From ceb2f6cb5ac373a22e0be5ce0db8254716afd40c Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 11:45:02 +0000 Subject: [PATCH 01/23] chore: Update capnweb to 0.10.0 A caret range on a 0.x version admits only patch releases, so ^0.8.0 held the workspace at 0.8.x and installing could not move it. Both declarations, in packages/rpc and packages/computer, now ask for ^0.10.0. Of the releases in between, 0.9.1 fixed nodeHttpBatchRpcResponse leaving the connection open and failing with ERR_HTTP_HEADERS_SENT on a request that is not a POST, and 0.10.0 added receiver-side limits on message size, nesting depth, and bigint length. The MessagePort wire format changed in 0.9.0; nothing here opens one. @cloudflare/sandbox pins capnweb at 0.8.0, so examples/think-compare-runtimes, which imports both it and @cloudflare/computer, resolves two copies where it previously shared one. Each drives its own session and no stub crosses between them. --- package-lock.json | 24 ++++++++++++++++++++++-- packages/computer/package.json | 2 +- packages/rpc/package.json | 2 +- 3 files changed, 24 insertions(+), 4 deletions(-) diff --git a/package-lock.json b/package-lock.json index a53cb7a1..17616fe2 100644 --- a/package-lock.json +++ b/package-lock.json @@ -13495,7 +13495,7 @@ "license": "MIT", "dependencies": { "acorn": "^8.17.0", - "capnweb": "^0.8.0", + "capnweb": "^0.10.0", "just-bash": "^3.0.1" }, "devDependencies": { @@ -13540,6 +13540,16 @@ "dev": true, "license": "MIT OR Apache-2.0" }, + "packages/computer/node_modules/capnweb": { + "version": "0.10.0", + "resolved": "https://registry.npmjs.org/capnweb/-/capnweb-0.10.0.tgz", + "integrity": "sha512-lQ2nlzaifkqBeNfs8RLErn7NZ8FNf3vn1W7J3nSH73edWG+0B5nHZGR51S2S9HMZEuZcEypOO469dZRY/BluFA==", + "license": "MIT", + "workspaces": [ + ".", + "packages/*" + ] + }, "packages/computerd": { "name": "@cloudflare/computerd", "version": "0.2.1", @@ -13583,7 +13593,7 @@ "version": "0.2.1", "dependencies": { "@cloudflare/dofs": "*", - "capnweb": "^0.8.0" + "capnweb": "^0.10.0" }, "devDependencies": { "@cloudflare/workers-types": "^4.20260616.1", @@ -13599,6 +13609,16 @@ "integrity": "sha512-mOhf5TUEB1m2vPrxtqoIGfz0fUC9xyxRDx5gWHy5s+OCo6dcV+g7wI1R7gYCMFohhqF/2y2xeKVwMwCJjfn/WA==", "dev": true, "license": "MIT OR Apache-2.0" + }, + "packages/rpc/node_modules/capnweb": { + "version": "0.10.0", + "resolved": "https://registry.npmjs.org/capnweb/-/capnweb-0.10.0.tgz", + "integrity": "sha512-lQ2nlzaifkqBeNfs8RLErn7NZ8FNf3vn1W7J3nSH73edWG+0B5nHZGR51S2S9HMZEuZcEypOO469dZRY/BluFA==", + "license": "MIT", + "workspaces": [ + ".", + "packages/*" + ] } } } diff --git a/packages/computer/package.json b/packages/computer/package.json index f0eefbe1..84da591e 100644 --- a/packages/computer/package.json +++ b/packages/computer/package.json @@ -117,7 +117,7 @@ }, "dependencies": { "acorn": "^8.17.0", - "capnweb": "^0.8.0", + "capnweb": "^0.10.0", "just-bash": "^3.0.1" }, "peerDependencies": { diff --git a/packages/rpc/package.json b/packages/rpc/package.json index c3eb78a2..518c59a5 100644 --- a/packages/rpc/package.json +++ b/packages/rpc/package.json @@ -35,7 +35,7 @@ }, "dependencies": { "@cloudflare/dofs": "*", - "capnweb": "^0.8.0" + "capnweb": "^0.10.0" }, "devDependencies": { "@cloudflare/workers-types": "^4.20260616.1", From c54406ba6527ed9d36de219902c251a09423f6b0 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 12:30:51 +0000 Subject: [PATCH 02/23] computerd: Remove UPSTREAM_URL Setting UPSTREAM_URL made the daemon dial the given address as a client, pull once before it started listening, and then run a sync tick every 250 milliseconds. It was the only path where the daemon drove its own sync loop. It predates POST /connect, which solved the same problem for hosts that cannot reach into the container, and no deployment sets it: the container backend passes PORT and MOUNT_POINT and nothing else. Sync is driven from the host across the capnweb session. Two rough edges go with it. An address that was wrong or not yet listening made the process exit before the HTTP server bound, so /health never answered and the operator got "WebSocket connection failed" and no further detail. And once a healthy peer dropped the socket nothing redialed, so the tick failed every 250 milliseconds while /health kept answering ok, leaving a dead sync loop behind a healthy probe and a log that grew without bound. The upstream option on createNodeVirtualFileSystem goes too, along with the polling loop and the stopSync handle it returned. Of the calls to that function, all but one passed no argument; the exception was this path. What is left has no opinion about syncing at all. --- packages/computerd/src/cli/computerd.ts | 25 +-------- packages/computerd/src/fuse/vfs.test.ts | 52 ------------------ packages/computerd/src/fuse/vfs.ts | 70 +++---------------------- 3 files changed, 9 insertions(+), 138 deletions(-) diff --git a/packages/computerd/src/cli/computerd.ts b/packages/computerd/src/cli/computerd.ts index 2727c6e3..f93f32fb 100644 --- a/packages/computerd/src/cli/computerd.ts +++ b/packages/computerd/src/cli/computerd.ts @@ -5,7 +5,6 @@ import { createServer, type IncomingMessage, type Server, type ServerResponse } import type { Socket } from "node:net"; import { isAbsolute } from "node:path"; import type { ExecEvent as RpcExecEvent } from "@cloudflare/computer-rpc"; -import { createWorkspaceClient, type WorkspaceClient } from "@cloudflare/computer-rpc/client"; import { isStubTrackingEnabled, stubSnapshot } from "@cloudflare/computer-rpc/debug"; import type { RunnerLike } from "@cloudflare/computer-rpc/server"; import { @@ -464,21 +463,7 @@ async function main(): Promise { const backend: FUSEBackend = await resolveFuseBackend(fuseMountMode); console.log(`[info] FUSE_MOUNT=${fuseMountMode} resolved to backend=${backend.kind}`); - const upstreamUrl = process.env.UPSTREAM_URL?.trim(); - let upstreamClient: WorkspaceClient | undefined; - if (upstreamUrl !== undefined && upstreamUrl.length > 0) { - // Use the `ws` package's WebSocket (not Node's built-in - // global) so the dial negotiates permessage-deflate against - // the upstream's WebSocketServer. Node 22's built-in - // WebSocket doesn't advertise the deflate extension. - upstreamClient = createWorkspaceClient({ - url: upstreamUrl, - WebSocketImpl: WebSocket as unknown as typeof globalThis.WebSocket, - }); - } - const { vfs, db, stopSync } = await createNodeVirtualFileSystem({ - upstream: upstreamClient?.sync, - }); + const { vfs, db } = await createNodeVirtualFileSystem(); const info: ComputerdInfo = { backend, mountPoint, port }; let fuse: FuseMount | undefined; @@ -584,14 +569,6 @@ async function main(): Promise { if (fuse !== undefined) { await unmount(fuse); } - if (upstreamClient !== undefined) { - try { - stopSync(); - await upstreamClient.close(); - } catch (error) { - console.error(error); - } - } teardownLogging(); process.exit(signal === "SIGINT" ? 130 : 143); }; diff --git a/packages/computerd/src/fuse/vfs.test.ts b/packages/computerd/src/fuse/vfs.test.ts index c6111f96..43ac6fbb 100644 --- a/packages/computerd/src/fuse/vfs.test.ts +++ b/packages/computerd/src/fuse/vfs.test.ts @@ -1,4 +1,3 @@ -import { createHash } from "node:crypto"; import { expect, test } from "vitest"; import { createNodeVirtualFileSystem } from "./index.js"; @@ -20,54 +19,3 @@ test("createNodeVirtualFileSystem returns a @platformatic/vfs filesystem", async vfs.unlinkSync("/project/greeting.txt"); expect(vfs.readdirSync("/project")).toEqual([]); }); - -test("createNodeVirtualFileSystem pulls initial state from an upstream SyncRPC", async () => { - const bytes = Buffer.from("hi"); - const hash = new Uint8Array(createHash("sha256").update(bytes).digest()); - - let fetchChangesCalls = 0; - const upstream = { - async fetchChanges() { - fetchChangesCalls++; - return { - currentCursor: { rev: 1, path: null }, - appliedPushCursor: { rev: 0, path: null }, - stream: new ReadableStream({ - start(c) { - c.enqueue({ - kind: "file", - rev: 1, - path: "/hi.txt", - mode: 0o644, - mtime: 100, - size: 2, - chunks: [{ hash, size: 2 }], - }); - c.close(); - }, - }), - }; - }, - async hasObjects(hashes) { - // The fake upstream is the source of truth for this file's - // chunk. Reply that we have every hash the client probes. - return hashes; - }, - async fetchObjects(hashes) { - return new ReadableStream({ - start(c) { - for (const h of hashes) c.enqueue({ hash: h, bytes }); - c.close(); - }, - }); - }, - async push() { - return { rev: 0, appliedPushCursor: { rev: 0, path: null } }; - }, - async pushObjects() {}, - }; - - const { vfs } = await createNodeVirtualFileSystem({ upstream }); - expect(fetchChangesCalls).toBe(1); - expect(vfs.readFileSync("/hi.txt").toString()).toBe("hi"); -}); diff --git a/packages/computerd/src/fuse/vfs.ts b/packages/computerd/src/fuse/vfs.ts index 432eeff1..c4023be7 100644 --- a/packages/computerd/src/fuse/vfs.ts +++ b/packages/computerd/src/fuse/vfs.ts @@ -1,5 +1,3 @@ -import type { SyncRPC } from "@cloudflare/computer-rpc"; -import { pullOnce, tick } from "@cloudflare/computer-rpc/driver"; import { Database, initializeSchema, SQLiteWorkspaceProvider } from "@cloudflare/dofs"; import { SQLiteTestStorage } from "@cloudflare/dofs/testing"; import { create, type VirtualFileSystem, VirtualProvider } from "@platformatic/vfs"; @@ -51,54 +49,24 @@ const EXTRA_VFS_METHODS = [ "releaseWriteBufferSync", ] as const; -export interface CreateOptions { - // Optional upstream sync surface. When set, the local store - // performs an initial pull on construction. When unset, computerd runs - // standalone against an in-memory store. - // - // The caller owns the carrier (WebSocket, in-process direct - // binding, or any future flavour). This package only needs the - // typed surface; the transport seam lives in computer-rpc. - // Future RPCs (exec, mounts, watchers) will travel beside - // SyncRPC on the same connection, so the caller may pass a - // composite stub — we accept the narrow SyncRPC subset - // structurally. - upstream?: SyncRPC; -} - export interface NodeVfsHandle { // @platformatic/vfs filesystem the FUSE driver consumes. vfs: NodeVirtualFileSystem; - // dofs Database backing the same store. Exposed so the - // CLI can construct a createSyncServer(db) and serve the local - // store to upstream callers over capnweb. + // dofs Database backing the same store. Exposed so the CLI can + // construct a createWorkspaceServer(db) and serve the local store + // to whoever holds the capnweb session. db: Database; - // Stop the periodic sync loop, if one was started. No-op when - // no upstream was provided. Idempotent. - stopSync: () => void; } -// Polling cadence for the background sync loop. Picked to match -// human-typing latency expectations without saturating the wire. -const SYNC_TICK_MS = 250; - -export async function createNodeVirtualFileSystem( - options: CreateOptions = {}, -): Promise { +// The store is local and process-lifetime. Sync is driven from the +// other end of the capnweb session: the host pushes changes in and +// pulls them back out, so nothing here polls. +export async function createNodeVirtualFileSystem(): Promise { ensureVirtualProviderPrototype(); const storage = new SQLiteTestStorage(); const db = new Database(storage); initializeSchema(db, () => Date.now()); - let stopSync = () => {}; - if (options.upstream !== undefined) { - // Initial pull. The polling loop would catch up eventually, - // but the FUSE mount comes up populated by waiting for the - // first pull to settle before returning. - await pullOnce(db, options.upstream); - stopSync = startSyncLoop(db, options.upstream); - } - const provider = new SQLiteWorkspaceProvider(db); const vfs = create(provider as unknown as VirtualProvider, { moduleHooks: false }); // Forward the extra dofs methods that @platformatic/vfs's @@ -117,27 +85,5 @@ export async function createNodeVirtualFileSystem( configurable: true, }); } - return { vfs, db, stopSync }; -} - -// Drive tick(db, upstream) on a setInterval. Errors during a tick -// log and continue — a transient upstream failure shouldn't -// kill the daemon. The watermarks are durable, so the next tick -// resumes from where the failed one would have. -function startSyncLoop(db: Database, upstream: SyncRPC): () => void { - let stopped = false; - const handle = setInterval(() => { - if (stopped) return; - tick(db, upstream).catch((error) => { - console.error("sync tick failed:", error); - }); - }, SYNC_TICK_MS); - // Don't block process exit on the timer. computerd's shutdown path - // calls stopSync() explicitly; this is belt-and-braces. - handle.unref?.(); - return () => { - if (stopped) return; - stopped = true; - clearInterval(handle); - }; + return { vfs, db }; } From 1a1998fea311f96f2f38227388d24fabfa333e6c Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 12:32:27 +0000 Subject: [PATCH 03/23] computerd: Take explicit paths in the /connect request POST /connect used to accept a base URL and build two paths from it: the base plus /health for the readiness poll, and the base plus /ws for the dial. Both belong to the host, so the daemon was deciding routes it does not serve, and renaming either one meant releasing a new binary alongside the host change. The request now names all three parts as base, health, and api. All are required: defaulting health to /health and api to /ws would leave the paths compiled in after all. A base must carry an http or https scheme, and each path must be absolute, so toHttpUrl is gone and toWebSocketUrl keeps normalizing the dial to ws or wss. Both paths are rejected if they parse as an address of their own or begin with a double slash. The daemon dials whatever it is handed, and either form would resolve somewhere other than under the given base. --- packages/computerd/src/cli/computerd.test.ts | 55 ++++++++++++- packages/computerd/src/cli/computerd.ts | 81 +++++++++++++------- 2 files changed, 107 insertions(+), 29 deletions(-) diff --git a/packages/computerd/src/cli/computerd.test.ts b/packages/computerd/src/cli/computerd.test.ts index f9b38b9d..263687d0 100644 --- a/packages/computerd/src/cli/computerd.test.ts +++ b/packages/computerd/src/cli/computerd.test.ts @@ -291,8 +291,10 @@ test("/connect re-dial tears down the prior WebSocket session", async (_ctx) => const peerPort = await getAvailablePort(); const opened = []; const peerSockets = new Set(); + // Deliberately non-default paths: the daemon must use what the + // request names, not paths of its own. const peerServer = http.createServer((req, res) => { - if (req.url === "/health") { + if (req.url === "/probe") { res.writeHead(200, { "content-type": "text/plain" }); res.end("ok\n"); return; @@ -305,7 +307,7 @@ test("/connect re-dial tears down the prior WebSocket session", async (_ctx) => }); const wss = new WebSocketServer({ noServer: true }); peerServer.on("upgrade", (req, socket, head) => { - if (req.url !== "/ws") { + if (req.url !== "/socket") { socket.destroy(); return; } @@ -337,7 +339,11 @@ test("/connect re-dial tears down the prior WebSocket session", async (_ctx) => const peerUrl = `http://127.0.0.1:${peerPort}`; const connect = async () => { - const res = await postJson(`http://127.0.0.1:${port}/connect`, { url: peerUrl }); + const res = await postJson(`http://127.0.0.1:${port}/connect`, { + base: peerUrl, + health: "/probe", + api: "/socket", + }); expect(res.statusCode).toBe(200); }; @@ -354,6 +360,49 @@ test("/connect re-dial tears down the prior WebSocket session", async (_ctx) => expect(opened[1].closed).toBe(false, "second peer WS should still be open"); }); +test("/connect rejects a body that does not name every part", async (_ctx) => { + // The daemon assembles no host paths of its own, so a request that + // leaves one out has to be refused rather than defaulted. + const port = await getAvailablePort(); + const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + await startComputerd({ port, mountPoint, env: { FUSE_MOUNT: "none" } }); + const url = `http://127.0.0.1:${port}/connect`; + + const cases = [ + [{}, "empty body"], + [{ health: "/health", api: "/api" }, "no base"], + [{ base: "http://127.0.0.1:1", api: "/api" }, "no health"], + [{ base: "http://127.0.0.1:1", health: "/health" }, "no api"], + [{ base: "ftp://127.0.0.1:1", health: "/health", api: "/api" }, "unsupported scheme"], + [{ base: "http://127.0.0.1:1", health: "health", api: "/api" }, "health lacks a slash"], + [{ base: "http://127.0.0.1:1", health: "/health", api: "api" }, "api lacks a slash"], + [ + { base: "http://127.0.0.1:1", health: "/health", api: "http://elsewhere/api" }, + "api is an absolute address", + ], + [ + { base: "http://127.0.0.1:1", health: "//elsewhere/health", api: "/api" }, + "health is protocol relative", + ], + ]; + + for (const [body, label] of cases) { + const res = await postJson(url, body); + expect(res.statusCode, `${label} should be refused`).toBe(400); + } + + // Control: a well-formed body must clear validation. Port 1 has + // nothing listening, so it gets as far as the health probe and + // fails there instead, which is a different status. + const ok = await postJson(url, { + base: "http://127.0.0.1:1", + health: "/health", + api: "/api", + healthTimeoutMs: 250, + }); + expect(ok.statusCode, "a complete body should reach the health probe").toBe(502); +}); + async function startComputerd({ port, mountPoint, diff --git a/packages/computerd/src/cli/computerd.ts b/packages/computerd/src/cli/computerd.ts index f93f32fb..8ecf7d9b 100644 --- a/packages/computerd/src/cli/computerd.ts +++ b/packages/computerd/src/cli/computerd.ts @@ -199,11 +199,11 @@ function createHTTPServer( return; } - // /connect — POST { url } where url is the http(s) base of an - // egress endpoint the host wants us to dial back into. We open a - // capnweb WebSocket session against `${url}/ws` and serve our RPC - // over it, exactly like /ws but with the carrier inverted (we - // dial out instead of accepting an inbound upgrade). + // /connect — POST { base, health, api } naming an endpoint the + // host wants us to dial back into. We poll `base + health` until + // it answers, then open a capnweb WebSocket session against + // `base + api` and serve our RPC over it: the same session as an + // inbound upgrade, with the carrier inverted because we dial out. if (path === "/connect") { if (request.method !== "POST") { send(response, 405, "method not allowed\n", { @@ -321,16 +321,37 @@ async function closeServer(server: Server): Promise { }); } +// The caller names every part of the endpoint it wants us to reach. +// We assemble `base + health` and `base + api` and hold no opinion +// about either path, so the host is free to rename its own routes +// without a matching release of this binary. interface ConnectBody { - // Base URL of the egress endpoint. ws[s]:// or http[s]://; we - // normalise http(s) to ws(s) and append /ws. - url?: unknown; - // How long to poll the upstream /health before giving up. - // Defaults to 30s; the egress proxy is up at boot but the worker - // that hosts it may take a tick. + // http:// or https:// origin of the endpoint, optionally with a + // path prefix. Trailing slashes are trimmed. + base?: unknown; + // Absolute path of the readiness probe, e.g. "/health". + health?: unknown; + // Absolute path of the capnweb endpoint, e.g. "/api". Dialed as + // ws:// or wss:// according to the scheme on `base`. + api?: unknown; + // How long to poll the readiness probe before giving up. Defaults + // to 30s; the egress proxy is up at boot but the worker that hosts + // it may take a tick. healthTimeoutMs?: unknown; } +// A path we're willing to append to `base`. Rejecting anything that +// parses as an address of its own keeps a caller from redirecting the +// dial somewhere the base never pointed. "//host/path" is +// protocol-relative, which resolves against the base's scheme rather +// than staying under the base, so it goes too. +function isEndpointPath(value: unknown): value is string { + if (typeof value !== "string") return false; + if (!value.startsWith("/")) return false; + if (value.startsWith("//")) return false; + return !URL.canParse(value); +} + async function handleConnect( request: IncomingMessage, response: ServerResponse, @@ -347,22 +368,38 @@ async function handleConnect( return; } - if (typeof body.url !== "string" || body.url.length === 0) { - send(response, 400, "missing 'url' in body\n", { + if ( + typeof body.base !== "string" || + !(body.base.startsWith("http://") || body.base.startsWith("https://")) + ) { + send(response, 400, "'base' must be an http:// or https:// URL\n", { + "content-type": "text/plain; charset=utf-8", + }); + return; + } + if (!isEndpointPath(body.health)) { + send(response, 400, "'health' must be an absolute path such as /health\n", { "content-type": "text/plain; charset=utf-8", }); return; } - const baseUrl = body.url.replace(/\/+$/, ""); + if (!isEndpointPath(body.api)) { + send(response, 400, "'api' must be an absolute path such as /api\n", { + "content-type": "text/plain; charset=utf-8", + }); + return; + } + const baseUrl = body.base.replace(/\/+$/, ""); + const healthUrl = `${baseUrl}${body.health}`; const healthTimeoutMs = typeof body.healthTimeoutMs === "number" && body.healthTimeoutMs > 0 ? body.healthTimeoutMs : 30_000; try { - await waitForHealth(baseUrl, healthTimeoutMs); + await waitForHealth(healthUrl, healthTimeoutMs); } catch (error) { - send(response, 502, `upstream /health unreachable: ${(error as Error).message}\n`, { + send(response, 502, `readiness probe unreachable: ${(error as Error).message}\n`, { "content-type": "text/plain; charset=utf-8", }); return; @@ -382,7 +419,7 @@ async function handleConnect( } } - const wsUrl = `${toWebSocketUrl(baseUrl)}/ws`; + const wsUrl = `${toWebSocketUrl(baseUrl)}${body.api}`; const ws = new WebSocket(wsUrl); upstreamSlot.ws = ws; ws.once("open", () => { @@ -413,15 +450,13 @@ async function readJson(request: IncomingMessage): Promise { } function toWebSocketUrl(input: string): string { - if (input.startsWith("ws://") || input.startsWith("wss://")) return input; if (input.startsWith("http://")) return `ws://${input.slice("http://".length)}`; if (input.startsWith("https://")) return `wss://${input.slice("https://".length)}`; throw new Error(`unsupported URL scheme: ${input}`); } -async function waitForHealth(baseUrl: string, timeoutMs: number): Promise { +async function waitForHealth(healthUrl: string, timeoutMs: number): Promise { const deadline = Date.now() + timeoutMs; - const healthUrl = `${toHttpUrl(baseUrl)}/health`; let lastError: unknown; while (Date.now() < deadline) { try { @@ -438,12 +473,6 @@ async function waitForHealth(baseUrl: string, timeoutMs: number): Promise ); } -function toHttpUrl(input: string): string { - if (input.startsWith("ws://")) return `http://${input.slice("ws://".length)}`; - if (input.startsWith("wss://")) return `https://${input.slice("wss://".length)}`; - return input; -} - async function main(): Promise { // Install logging + crash handlers first thing so any early throws // (parsePort, parseMountPoint, resolveFuseBackend) still land in From ff2a618c6db109c145403d5feab3d3b81dbcb587 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 12:34:01 +0000 Subject: [PATCH 04/23] computerd: Serve the capnweb session on /api The daemon mounted one RPC object on two carriers: an HTTP batch handler on POST /api and a websocket upgrade on /ws. It now serves the websocket alone, on /api. The batch route goes because it could not carry the interface. Every call that reads data back returns a ReadableStream, and delivering one means the server calls write() on a stub the client exposed and waits for an acknowledgement. A batch session cannot complete a call from server to client, so fetchChanges, fetchObjects, exec, and getExec all fail once there is anything to deliver, while watermarks and hasObjects succeed. An endpoint that answers the cheap calls and fails the rest is worse than one that is absent. A request that reaches /api without an Upgrade header now gets 400 and a sentence saying a websocket is required. Handshakes that do try to upgrade are answered by the ws package, except an unsupported Sec-WebSocket-Version: it answers 400 where the specification calls for 426, so that case is handled before handing over and reports the versions the server speaks. Samplers that only want revision numbers have GET /__computerd/watermarks, which reads through the same watermarks() the wire serves so the two cannot drift. --- packages/computerd/src/cli/computerd.test.ts | 102 ++++++++++++++++-- packages/computerd/src/cli/computerd.ts | 92 ++++++++++------ .../computerd/src/exec/runner.fuse.test.ts | 2 +- packages/computerd/src/fuse/backend.ts | 2 +- 4 files changed, 155 insertions(+), 43 deletions(-) diff --git a/packages/computerd/src/cli/computerd.test.ts b/packages/computerd/src/cli/computerd.test.ts index 263687d0..a28c29ae 100644 --- a/packages/computerd/src/cli/computerd.test.ts +++ b/packages/computerd/src/cli/computerd.test.ts @@ -105,13 +105,13 @@ test("computerd exposes file IO through real FUSE when FUSE_MOUNT=fuse", async ( expect(await fs.readFile(path.join(mountPoint, "dir", "hello.txt"), "utf8")).toBe("hello fuse"); }); -test("/ws serves a capnweb WorkspaceRPC session", async (_ctx) => { +test("/api serves a capnweb WorkspaceRPC session", async (_ctx) => { const { createWorkspaceClient } = await import("@cloudflare/computer-rpc/client"); const port = await getAvailablePort(); const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); await startComputerd({ port, mountPoint, env: { FUSE_MOUNT: "none" } }); - const client = createWorkspaceClient({ url: `ws://127.0.0.1:${port}/ws` }); + const client = createWorkspaceClient({ url: `ws://127.0.0.1:${port}/api` }); try { // hasObjects against a fresh DB returns the empty subset. expect(await client.sync.hasObjects([])).toEqual([]); @@ -133,15 +133,73 @@ test("/ws serves a capnweb WorkspaceRPC session", async (_ctx) => { } }); -test("/api serves a capnweb HTTP-batch WorkspaceRPC session", async (_ctx) => { - const { newHttpBatchRpcSession } = await import("capnweb"); +test("/api refuses anything that is not a websocket handshake", async (_ctx) => { + // /api carries one transport. A caller that arrives over plain HTTP, + // or botches the handshake, should learn that from the status rather + // than from a capnweb error several calls later. const port = await getAvailablePort(); const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); await startComputerd({ port, mountPoint, env: { FUSE_MOUNT: "none" } }); + const base = `http://127.0.0.1:${port}/api`; + + // No Upgrade header at all: never reaches the upgrade listener. + const plainGet = await fetch(base); + expect(plainGet.status).toBe(400); + expect(await plainGet.text()).toMatch(/websocket/i); + + const post = await fetch(base, { method: "POST", body: "[]" }); + expect(post.status).toBe(400); + + // An upgrade attempt naming a version we do not speak gets 426 and + // the versions we do speak, per the websocket specification. + const badVersion = await rawRequest(port, [ + "GET /api HTTP/1.1", + `Host: 127.0.0.1:${port}`, + "Upgrade: websocket", + "Connection: Upgrade", + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==", + "Sec-WebSocket-Version: 7", + ]); + expect(badVersion).toMatch(/^HTTP\/1\.1 426 /); + expect(badVersion).toMatch(/Sec-WebSocket-Version: 13, 8/i); + + // A handshake missing its key is malformed, not a version problem. + const noKey = await rawRequest(port, [ + "GET /api HTTP/1.1", + `Host: 127.0.0.1:${port}`, + "Upgrade: websocket", + "Connection: Upgrade", + "Sec-WebSocket-Version: 13", + ]); + expect(noKey).toMatch(/^HTTP\/1\.1 400 /); + + // An unknown path is still a 404, upgrade header or not. + const unknown = await rawRequest(port, [ + "GET /nope HTTP/1.1", + `Host: 127.0.0.1:${port}`, + "Upgrade: websocket", + "Connection: Upgrade", + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==", + "Sec-WebSocket-Version: 13", + ]); + expect(unknown).toMatch(/^HTTP\/1\.1 404 /); +}); + +test("/__computerd/watermarks reports sync revisions over plain HTTP", async (_ctx) => { + // Samplers want three numbers on an interval. Opening an RPC session + // per sample is the wrong shape for that. + const port = await getAvailablePort(); + const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-watermarks-")); + await startComputerd({ port, mountPoint, env: { FUSE_MOUNT: "none" } }); - // HTTP batch flushes on first await; each call is a fresh session. - const stub = newHttpBatchRpcSession(`http://127.0.0.1:${port}/api`); - expect(await stub.sync.hasObjects([])).toEqual([]); + const res = await fetch(`http://127.0.0.1:${port}/__computerd/watermarks`); + expect(res.status).toBe(200); + const body = await res.json(); + expect(body).toMatchObject({ + currentRev: expect.any(Number), + pushRev: expect.any(Number), + fetchCursor: { rev: expect.any(Number) }, + }); }); test("/__computerd/stats returns DOFS table sizes and process memory", async (_ctx) => { @@ -222,7 +280,7 @@ test("FUSE_MOUNT=shim materialises an RPC push under the mount point", async (_c const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-shim-push-")); await startComputerd({ port, mountPoint, env: { FUSE_MOUNT: "shim" } }); - const client = createWorkspaceClient({ url: `ws://127.0.0.1:${port}/ws` }); + const client = createWorkspaceClient({ url: `ws://127.0.0.1:${port}/api` }); onTestFinished(() => client.close()); const db = new Database(new SQLiteTestStorage()); @@ -533,6 +591,34 @@ async function waitFor(predicate, { timeoutMs = 2_000, intervalMs = 10 } = {}) { throw new Error("waitFor: predicate did not become true within the timeout"); } +// Write a request onto a raw socket and return the response head. +// http.request() will not send a malformed websocket handshake, and +// fetch() will not send one at all, so the handshake cases need this. +function rawRequest(port, lines) { + return new Promise((resolve, reject) => { + const socket = net.connect(port, "127.0.0.1", () => { + socket.write(`${lines.join("\r\n")}\r\n\r\n`); + }); + let buf = ""; + socket.setEncoding("utf8"); + socket.on("data", (chunk) => { + buf += chunk; + }); + socket.once("error", reject); + // The server may hold the socket open after refusing the + // handshake, so settle on the end of the response head. + const settle = () => { + socket.destroy(); + resolve(buf); + }; + socket.on("data", () => { + if (buf.includes("\r\n\r\n")) settle(); + }); + socket.once("close", () => resolve(buf)); + setTimeout(settle, 2_000); + }); +} + function postJson(url, body) { const payload = JSON.stringify(body); return new Promise((resolve, reject) => { diff --git a/packages/computerd/src/cli/computerd.ts b/packages/computerd/src/cli/computerd.ts index 8ecf7d9b..dd26fbc4 100644 --- a/packages/computerd/src/cli/computerd.ts +++ b/packages/computerd/src/cli/computerd.ts @@ -7,11 +7,7 @@ import { isAbsolute } from "node:path"; import type { ExecEvent as RpcExecEvent } from "@cloudflare/computer-rpc"; import { isStubTrackingEnabled, stubSnapshot } from "@cloudflare/computer-rpc/debug"; import type { RunnerLike } from "@cloudflare/computer-rpc/server"; -import { - acceptWebSocketSession, - createWorkspaceServer, - serveHTTPBatch, -} from "@cloudflare/computer-rpc/server"; +import { acceptWebSocketSession, createWorkspaceServer } from "@cloudflare/computer-rpc/server"; import type { Database } from "@cloudflare/dofs"; import { WebSocket, WebSocketServer } from "ws"; import { Runner } from "../exec/index.js"; @@ -176,25 +172,14 @@ function createHTTPServer( const server = createServer((request, response) => { const path = requestPath(request); - // /api — capnweb HTTP-batch endpoint. Single POST per call; - // request body carries the serialized message, response body - // carries the reply. Useful for environments that can't open - // a WebSocket (curl, fetch from a Worker without ws upgrade). + // /api — the capnweb endpoint. It carries one transport, a + // websocket, so a request that reaches the ordinary handler here + // has no Upgrade header and cannot be served. Say so plainly: + // the alternative is a caller that connects, makes one call that + // appears to work, and then fails on a stream. if (path === "/api") { - if (request.method !== "POST") { - send(response, 405, "method not allowed\n", { - allow: "POST", - "content-type": "text/plain; charset=utf-8", - }); - return; - } - void serveHTTPBatch(request, response, rpc).catch((error) => { - console.error("/api batch failed:", error); - if (!response.headersSent) { - send(response, 500, "internal error\n", { - "content-type": "text/plain; charset=utf-8", - }); - } + send(response, 400, "/api requires a websocket upgrade\n", { + "content-type": "text/plain; charset=utf-8", }); return; } @@ -255,6 +240,32 @@ function createHTTPServer( return; } + // Sync revisions over plain HTTP, for samplers that want a few + // numbers on an interval rather than an RPC session. Reads through + // the same watermarks() the wire serves, so the two cannot drift. + if (path === "/__computerd/watermarks") { + if (request.method === "HEAD") { + send(response, 200, "", { "content-type": "application/json; charset=utf-8" }); + return; + } + void rpc.sync + .watermarks() + .then((watermarks) => { + send(response, 200, JSON.stringify(watermarks), { + "content-type": "application/json; charset=utf-8", + }); + }) + .catch((error: unknown) => { + console.error("/__computerd/watermarks failed:", error); + if (!response.headersSent) { + send(response, 500, "internal error\n", { + "content-type": "text/plain; charset=utf-8", + }); + } + }); + return; + } + if (path === "/__computerd/info") { const body = request.method === "HEAD" ? "" : JSON.stringify(info); send(response, 200, body, { @@ -276,24 +287,39 @@ function createHTTPServer( }); }); - // /ws — capnweb WebSocket endpoint. Long-lived, bidirectional, - // streaming-friendly. The container's primary sync carrier. - // perMessageDeflate compresses each WS frame with zlib. Defaults - // off in the `ws` package; we turn it on so computerd-to-computerd peers - // (and any Node-side client that negotiates the extension) save - // bytes on the wire. Clients that don't advertise the extension - // negotiate down to plain frames, so no flag day for workerd or - // browser callers. + // /api — the capnweb endpoint. Long-lived, bidirectional, and + // streaming-friendly, which is what the wire contract needs: every + // call that reads data back returns a ReadableStream. + // perMessageDeflate compresses each frame with zlib. Defaults off + // in the `ws` package; we turn it on so any Node-side client that + // negotiates the extension saves bytes on the wire. Clients that + // don't advertise it negotiate down to plain frames, so no flag day + // for workerd or browser callers. const wss = new WebSocketServer({ noServer: true, perMessageDeflate: true }); wss.on("connection", (ws) => { acceptWebSocketSession(ws, rpc); }); server.on("upgrade", (request, socket, head) => { - if (requestPath(request) !== "/ws") { + if (requestPath(request) !== "/api") { socket.write("HTTP/1.1 404 Not Found\r\n\r\n"); socket.destroy(); return; } + // `ws` answers a malformed handshake for us: 405 for a method + // other than GET, 400 for a bad Upgrade header or a missing key. + // It answers 400 for an unsupported version too, where the + // websocket specification calls for 426, so that one case is + // handled here before handing over. + const version = Number(request.headers["sec-websocket-version"]); + if (version !== 13 && version !== 8) { + socket.write( + "HTTP/1.1 426 Upgrade Required\r\n" + + "Sec-WebSocket-Version: 13, 8\r\n" + + "Connection: close\r\n\r\n", + ); + socket.destroy(); + return; + } wss.handleUpgrade(request, socket as Socket, head, (ws) => { wss.emit("connection", ws, request); }); @@ -486,7 +512,7 @@ async function main(): Promise { // FUSE_MOUNT picks the backend. auto (default) probes /dev/fuse // or macFUSE and falls back to the userspace shim. fuse / macfuse // require their respective real backend. shim forces the userspace - // polling shim. none skips the mount entirely; HTTP + /api + /ws + // polling shim. none skips the mount entirely; HTTP and /api // still come up so tests and tooling can talk to computerd's RPC surface. const fuseMountMode = parseFuseMountMode(process.env.FUSE_MOUNT); const backend: FUSEBackend = await resolveFuseBackend(fuseMountMode); diff --git a/packages/computerd/src/exec/runner.fuse.test.ts b/packages/computerd/src/exec/runner.fuse.test.ts index da5d2e8c..23b85877 100644 --- a/packages/computerd/src/exec/runner.fuse.test.ts +++ b/packages/computerd/src/exec/runner.fuse.test.ts @@ -107,7 +107,7 @@ describeIfReal("Runner shell.exec under real FUSE", () => { test("exec(cwd inside the FUSE mount) returns quickly and runs the command", async () => { if (!url) throw new Error("computerd container did not start"); const client = createWorkspaceClient({ - url: `${url.replace(/^http(s?):\/\//, "ws$1://")}/ws`, + url: `${url.replace(/^http(s?):\/\//, "ws$1://")}/api`, WebSocketImpl: WebSocket, }); try { diff --git a/packages/computerd/src/fuse/backend.ts b/packages/computerd/src/fuse/backend.ts index 5f90eb77..3d4f5cf3 100644 --- a/packages/computerd/src/fuse/backend.ts +++ b/packages/computerd/src/fuse/backend.ts @@ -8,7 +8,7 @@ import { access as defaultAccess } from "node:fs/promises"; // - fuse : require the linux kernel FUSE backend. // - macfuse: require macFUSE on darwin. // - shim : force the userspace shim. Works on any platform. -// - none : skip the mount entirely; HTTP + /api + /ws still come up. +// - none : skip the mount entirely; HTTP and /api still come up. export type FuseMountMode = "auto" | "fuse" | "macfuse" | "shim" | "none"; // The resolved backend selection. No `reason` field — the choice is From 3add88e5b6b82fa4491f1ac8a702db23f4683788 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 12:34:01 +0000 Subject: [PATCH 05/23] rpc: Remove serveHTTPBatch The daemon served capnweb's HTTP batch transport through this wrapper and no longer does, leaving it without a caller. Removing it also drops the package's last use of capnweb's batch surface. --- packages/rpc/src/server.ts | 14 +------------- 1 file changed, 1 insertion(+), 13 deletions(-) diff --git a/packages/rpc/src/server.ts b/packages/rpc/src/server.ts index 9d0f23b8..3556663a 100644 --- a/packages/rpc/src/server.ts +++ b/packages/rpc/src/server.ts @@ -21,7 +21,7 @@ import { stageBlob, writeFetchCursor, } from "@cloudflare/dofs"; -import { newWebSocketRpcSession, nodeHttpBatchRpcResponse, RpcTarget } from "capnweb"; +import { newWebSocketRpcSession, RpcTarget } from "capnweb"; import { trackStub, untrackStub } from "./debug.js"; import type { ExecEvent, ShellRPC, SyncRPC, WorkspaceRPC } from "./interface.js"; @@ -349,18 +349,6 @@ export function acceptWebSocketSession( newWebSocketRpcSession(ws as unknown as WebSocket, rpc as unknown as RpcTarget); } -// Serve a single capnweb HTTP-batch session against a SyncRPC. Wraps -// capnweb's nodeHttpBatchRpcResponse so computerd never directly imports -// capnweb (which would split capnweb's module identity in mixed -// ESM/CJS contexts — the RpcTarget instanceof check then fails). -export function serveHTTPBatch( - request: import("node:http").IncomingMessage, - response: import("node:http").ServerResponse, - rpc: SyncRPC | ShellRPC | WorkspaceRPC, -): Promise { - return nodeHttpBatchRpcResponse(request, response, rpc as unknown as RpcTarget); -} - function iterableToReadableStream(it: AsyncIterable): ReadableStream { const iterator = it[Symbol.asyncIterator](); return new ReadableStream({ From fd7789ba550e9879815244e061638b03826115fe Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 12:16:51 +0000 Subject: [PATCH 06/23] computer: Name the egress paths in the /connect request The container backend sent only the egress base and left the daemon to append /health and /ws. The daemon now expects the request to name both paths, so send them. EGRESS_HEALTH_PATH and EGRESS_API_PATH sit next to DEFAULT_EGRESS_HOST because they describe the same endpoint: the routes WorkspaceProxy answers on the far side of the egress interceptor. Keeping them beside the host name means a rename touches one place. --- .../container/cloudflare-container.test.ts | 26 +++++++++++++++++++ .../container/cloudflare-container.ts | 9 ++++++- 2 files changed, 34 insertions(+), 1 deletion(-) diff --git a/packages/computer/src/backends/container/cloudflare-container.test.ts b/packages/computer/src/backends/container/cloudflare-container.test.ts index 09989a54..33cf36a2 100644 --- a/packages/computer/src/backends/container/cloudflare-container.test.ts +++ b/packages/computer/src/backends/container/cloudflare-container.test.ts @@ -34,6 +34,7 @@ interface FakeHostOptions { interface FakeHost { host: IWorkspaceContainerAPI; calls: { name: string; args: unknown[] }[]; + connectBody?: Record; startEnv?: Record; enableInternet?: boolean; interceptedHost?: string; @@ -102,6 +103,10 @@ function makeFakeHost(opts: FakeHostOptions = {}): FakeHost { return new Response(null, { status: 200 }); } if (url.pathname === "/connect") { + state.connectBody = await request + .clone() + .json() + .catch(() => undefined); if (connectStatus !== 200) { return new Response(`/connect ${connectStatus}`, { status: connectStatus }); } @@ -380,6 +385,27 @@ describe("CloudflareContainerBackend", () => { expect(String(error)).toMatch(/POST \/connect returned 502/); }); + test("connect() names the egress base and both paths in the /connect body", async () => { + // The container assembles no host paths of its own, so the + // request has to carry them. A missing field would leave the + // daemon with nothing to dial. + const fake = makeFakeHost(); + const backend = new CloudflareContainerBackend({ + container: () => ({ getWorkspaceContainer: () => fake.host }), + workspace: fakeWorkspace, + connectTimeoutMs: 600, + }); + + await backend.connect().catch(() => {}); + + expect(fake.connectBody).toMatchObject({ + base: "http://computer.internal", + health: "/health", + api: "/ws", + }); + expect(typeof fake.connectBody?.healthTimeoutMs).toBe("number"); + }); + test("connect() throws a transport error when the /ws upgrade never arrives", async () => { const fake = makeFakeHost(); const backend = new CloudflareContainerBackend({ diff --git a/packages/computer/src/backends/container/cloudflare-container.ts b/packages/computer/src/backends/container/cloudflare-container.ts index 5ecd25a8..fab9a8fc 100644 --- a/packages/computer/src/backends/container/cloudflare-container.ts +++ b/packages/computer/src/backends/container/cloudflare-container.ts @@ -138,6 +138,11 @@ export interface CloudflareContainerBackendOptions { } const DEFAULT_EGRESS_HOST = "computer.internal"; +// Paths the egress proxy serves. The container assembles no paths of +// its own, so these travel in the /connect request and both ends stay +// in step from one place. +const EGRESS_HEALTH_PATH = "/health"; +const EGRESS_API_PATH = "/ws"; const DEFAULT_CONTAINER_PORT = 8080; const DEFAULT_CONNECT_TIMEOUT_MS = 30_000; const DEFAULT_HEARTBEAT_INTERVAL_MS = 20_000; @@ -519,7 +524,9 @@ export class CloudflareContainerBackend implements WorkspaceBackend { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ - url: `http://${this.#options.egressHost}`, + base: `http://${this.#options.egressHost}`, + health: EGRESS_HEALTH_PATH, + api: EGRESS_API_PATH, healthTimeoutMs: remaining, }), }); From ad7e073b4d043feb64ac74fb9f9be2da8af65011 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 12:26:17 +0000 Subject: [PATCH 07/23] computer: Serve the capnweb session on /api The daemon now serves its capnweb session on /api, and the host end of the same connection follows, so both sides of the wire use one name. WorkspaceProxy answers /api, the container backend routes /api upgrades into the in-flight connect(), and the path travels to the daemon in the /connect request rather than being assembled there. The tokenized callback under /__workspace_connect keeps its existing spelling. Nothing in this repository builds those URLs, so renaming that segment would change an inbound contract without moving anything; both forms normalize to /api before reaching the durable object. handleFetch answered 426 for a request with no Upgrade header. That is a malformed handshake, which is a 400. 426 belongs to the narrower case of an unsupported Sec-WebSocket-Version, which is what the daemon now answers for it, so both ends agree on what each status means. --- .../container/cloudflare-container.test.ts | 22 +++++++++---------- .../container/cloudflare-container.ts | 13 ++++++----- .../src/backends/container/container-host.ts | 2 +- packages/computer/src/backends/test.ts | 2 +- packages/computer/src/proxy.ts | 16 +++++++++----- packages/computer/tests/proxy-worker.ts | 6 ++--- packages/computer/tests/proxy.test.ts | 18 +++++++-------- 7 files changed, 43 insertions(+), 36 deletions(-) diff --git a/packages/computer/src/backends/container/cloudflare-container.test.ts b/packages/computer/src/backends/container/cloudflare-container.test.ts index 33cf36a2..51eebe92 100644 --- a/packages/computer/src/backends/container/cloudflare-container.test.ts +++ b/packages/computer/src/backends/container/cloudflare-container.test.ts @@ -4,7 +4,7 @@ // The successful connect() path constructs a WebSocketPair, which // is a workerd global not available under the vitest node runner. // These tests cover the paths that bail before the upgrade (port -// never opens, /connect non-2xx, /ws upgrade timeout), the +// never opens, /connect non-2xx, /api upgrade timeout), the // handleFetch input validation, and the factory + workspace-ref // plumbing. The full happy-path round-trip is covered by the live // example. @@ -246,7 +246,7 @@ describe("CloudflareContainerBackend", () => { egress: { mode: "http-gateway", gateway }, }); await expect(backend.connect()).rejects.toThrow(); - const request = new Request("https://workspace.internal/ws", { + const request = new Request("https://workspace.internal/api", { method: "POST", body: "payload", headers: { @@ -280,7 +280,7 @@ describe("CloudflareContainerBackend", () => { await expect(backend.connect()).rejects.toThrow(); const response = await backend.handleFetch( - new Request("https://workspace.internal/ws", { + new Request("https://workspace.internal/api", { headers: { "x-workspace-egress-token": fake.gatewayToken ?? "", "x-workspace-egress-url": "ftp://api.example.test/data", @@ -401,12 +401,12 @@ describe("CloudflareContainerBackend", () => { expect(fake.connectBody).toMatchObject({ base: "http://computer.internal", health: "/health", - api: "/ws", + api: "/api", }); expect(typeof fake.connectBody?.healthTimeoutMs).toBe("number"); }); - test("connect() throws a transport error when the /ws upgrade never arrives", async () => { + test("connect() throws a transport error when the /api upgrade never arrives", async () => { const fake = makeFakeHost(); const backend = new CloudflareContainerBackend({ container: () => ({ getWorkspaceContainer: () => fake.host }), @@ -416,10 +416,10 @@ describe("CloudflareContainerBackend", () => { const error = await backend.connect().catch((caught: unknown) => caught); expect(error).toBeInstanceOf(WorkspaceTransportError); - expect(String(error)).toMatch(/\/ws upgrade did not arrive/); + expect(String(error)).toMatch(/\/api upgrade did not arrive/); }); - test("handleFetch rejects non-/ws paths", async () => { + test("handleFetch rejects non-/api paths", async () => { const fake = makeFakeHost(); const backend = new CloudflareContainerBackend({ container: () => ({ getWorkspaceContainer: () => fake.host }), @@ -435,8 +435,8 @@ describe("CloudflareContainerBackend", () => { container: () => ({ getWorkspaceContainer: () => fake.host }), workspace: fakeWorkspace, }); - const res = await backend.handleFetch(new Request("http://computer.internal/ws")); - expect(res.status).toBe(426); + const res = await backend.handleFetch(new Request("http://computer.internal/api")); + expect(res.status).toBe(400); }); test("connect() consults host.exitInfo() before host.start()", async () => { @@ -480,9 +480,9 @@ describe("CloudflareContainerBackend", () => { test("connect() restarts the host when initial readiness fails and recovers", async () => { // First attempt drains all probes as failures; restart() runs; // the second attempt's very first probe answers healthy. - // connect() still fails at the /ws upgrade (no WebSocketPair + // connect() still fails at the /api upgrade (no WebSocketPair // under node) — the point is that readiness recovered after - // restart and we reached the /connect POST and /ws upgrade. + // restart and we reached the /connect POST and /api upgrade. const fake = makeFakeHost({ healthSequence: [ // First attempt — enough failures to exhaust the budget. diff --git a/packages/computer/src/backends/container/cloudflare-container.ts b/packages/computer/src/backends/container/cloudflare-container.ts index fab9a8fc..06bd72e9 100644 --- a/packages/computer/src/backends/container/cloudflare-container.ts +++ b/packages/computer/src/backends/container/cloudflare-container.ts @@ -142,7 +142,7 @@ const DEFAULT_EGRESS_HOST = "computer.internal"; // its own, so these travel in the /connect request and both ends stay // in step from one place. const EGRESS_HEALTH_PATH = "/health"; -const EGRESS_API_PATH = "/ws"; +const EGRESS_API_PATH = "/api"; const DEFAULT_CONTAINER_PORT = 8080; const DEFAULT_CONNECT_TIMEOUT_MS = 30_000; const DEFAULT_HEARTBEAT_INTERVAL_MS = 20_000; @@ -333,7 +333,7 @@ export class CloudflareContainerBackend implements WorkspaceBackend { return handle; } - // Routes a /ws upgrade Request into the in-flight connect(). + // Routes an /api upgrade Request into the in-flight connect(). // Returns the 101 response that the WorkspaceProxy fetch handler // forwards back to the container. async handleFetch(req: Request): Promise { @@ -359,11 +359,14 @@ export class CloudflareContainerBackend implements WorkspaceBackend { return this.#egress.gateway.fetch(new Request(parsedUrl, sanitized)); } const url = new URL(req.url); - if (url.pathname !== "/ws") { + if (url.pathname !== EGRESS_API_PATH) { return new Response("not found", { status: 404 }); } + // A request with no Upgrade header is a malformed handshake, which + // is a 400. 426 belongs to the narrower case of a version this end + // does not speak, and is what the daemon answers for it. if (req.headers.get("upgrade") !== "websocket") { - return new Response("expected websocket upgrade", { status: 426 }); + return new Response(`${EGRESS_API_PATH} requires a websocket upgrade`, { status: 400 }); } const pair = new WebSocketPair(); @@ -564,7 +567,7 @@ export class CloudflareContainerBackend implements WorkspaceBackend { () => reject( new WorkspaceTransportError( - `CloudflareContainerBackend(${this.id}) [stage=ws]: /ws upgrade did not arrive within ${this.#options.connectTimeoutMs}ms`, + `CloudflareContainerBackend(${this.id}) [stage=ws]: /api upgrade did not arrive within ${this.#options.connectTimeoutMs}ms`, ), ), remaining, diff --git a/packages/computer/src/backends/container/container-host.ts b/packages/computer/src/backends/container/container-host.ts index 1ba1a8f3..eba11ba9 100644 --- a/packages/computer/src/backends/container/container-host.ts +++ b/packages/computer/src/backends/container/container-host.ts @@ -34,7 +34,7 @@ import { export type { ContainerExitInfo } from "./container-lifecycle.js"; // Identifies the Durable Object that owns the Workspace and answers -// the /ws upgrade. Plain data so it can travel over Workers RPC. +// the /api upgrade. Plain data so it can travel over Workers RPC. export interface WorkspaceRef { // Binding name in the host Worker's env that resolves to the // DurableObjectNamespace for the Workspace-owning DO class. diff --git a/packages/computer/src/backends/test.ts b/packages/computer/src/backends/test.ts index 75ca24c2..66e9e7ff 100644 --- a/packages/computer/src/backends/test.ts +++ b/packages/computer/src/backends/test.ts @@ -41,7 +41,7 @@ export class TestBackend implements WorkspaceBackend { // first RPC. The probe surfaces "harness forgot to start the // container" up front. await probeHealth(this.#url); - const client = createWorkspaceClient({ url: `${wsUrl}/ws` }); + const client = createWorkspaceClient({ url: `${wsUrl}/api` }); return { rpc: client, close: async () => { diff --git a/packages/computer/src/proxy.ts b/packages/computer/src/proxy.ts index 3da08082..b9222908 100644 --- a/packages/computer/src/proxy.ts +++ b/packages/computer/src/proxy.ts @@ -33,10 +33,10 @@ // // override async fetch(req: Request): Promise { // // The DO answers /health (port-readiness poll from the -// // backend) and /ws (capnweb upgrade) on its own fetch(). +// // backend) and /api (capnweb upgrade) on its own fetch(). // const url = new URL(req.url); // if (url.pathname === "/health") return new Response("ok\n"); -// if (url.pathname === "/ws") return this.#backend.handleFetch(req); +// if (url.pathname === "/api") return this.#backend.handleFetch(req); // return new Response("not found", { status: 404 }); // } // } @@ -46,7 +46,7 @@ // proxy looks up `env[binding]` at fetch time and falls back to a // clear error if the name doesn't resolve. The DO class doesn't // need to live in @cloudflare/computer — the proxy works for any -// DO that implements a fetch() handler answering /health and /ws. +// DO that implements a fetch() handler answering /health and /api. import { RpcTarget, WorkerEntrypoint } from "cloudflare:workers"; @@ -71,7 +71,7 @@ export class WorkspaceProxy extends WorkerEntrypoint { override async fetch(request: Request): Promise { const url = new URL(request.url); const egressToken = request.headers.get("x-workspace-egress-token"); - if (url.pathname === "/ws" && egressToken !== null) { + if (url.pathname === "/api" && egressToken !== null) { return Response.json({ callbackUrl: request.url, originalUrl: request.headers.get("x-workspace-egress-url"), @@ -35,7 +35,7 @@ export class TestStorageDO extends DurableObject { body: await request.text(), }); } - if (url.pathname === "/ws") { + if (url.pathname === "/api") { return new Response( url.searchParams.has("token") ? `from-do:${url.searchParams.get("token")}` : "from-do", { status: 200 }, diff --git a/packages/computer/tests/proxy.test.ts b/packages/computer/tests/proxy.test.ts index eb3db15d..85bc3a45 100644 --- a/packages/computer/tests/proxy.test.ts +++ b/packages/computer/tests/proxy.test.ts @@ -3,7 +3,7 @@ // into the DO. Two-and-a-half pieces of logic to pin: // // - /health answers 200 ok\n (the port-readiness probe). -// - /ws looks up env[binding] and forwards the request to the +// - /api looks up env[binding] and forwards the request to the // named DO instance. // - anything else is 404. // @@ -34,15 +34,15 @@ describe("WorkspaceProxy", () => { expect(res.headers.get("content-type")).toMatch(/text\/plain/); }); - it("/ws forwards to the DO at env[binding]", async () => { - const res = await SELF.fetch("http://proxy.test/ws", { + it("/api forwards to the DO at env[binding]", async () => { + const res = await SELF.fetch("http://proxy.test/api", { headers: { "x-test-id": freshId(), "x-test-binding": "COMPUTERD" }, }); expect(res.status).toBe(200); expect(await res.text()).toBe("from-do"); }); - it("accepts tokenized callback health and forwards a normalized tokenized /ws", async () => { + it("accepts tokenized callback health and normalizes the tokenized callback to /api", async () => { const token = "123e4567-e89b-12d3-a456-426614174000"; const health = await SELF.fetch(`http://proxy.test/__workspace_connect/${token}/health`, { headers: { "x-test-id": freshId() }, @@ -54,7 +54,7 @@ describe("WorkspaceProxy", () => { expect(await websocket.text()).toBe(`from-do:${token}`); }); - it("routes egress callbacks through /ws while preserving the original request", async () => { + it("routes egress callbacks through /api while preserving the original request", async () => { const res = await SELF.fetch("https://api.example.test/v1/data?format=json", { method: "POST", body: "payload", @@ -66,7 +66,7 @@ describe("WorkspaceProxy", () => { }); expect(await res.json()).toEqual({ - callbackUrl: "https://api.example.test/ws", + callbackUrl: "https://api.example.test/api", originalUrl: "https://api.example.test/v1/data?format=json", egressToken: "secret-token", method: "POST", @@ -74,15 +74,15 @@ describe("WorkspaceProxy", () => { }); }); - it("/ws returns 500 when env[binding] is missing", async () => { - const res = await SELF.fetch("http://proxy.test/ws", { + it("/api returns 500 when env[binding] is missing", async () => { + const res = await SELF.fetch("http://proxy.test/api", { headers: { "x-test-id": freshId(), "x-test-binding": "NOT_A_BINDING" }, }); expect(res.status).toBe(500); expect(await res.text()).toMatch(/NOT_A_BINDING is not a DurableObjectNamespace/); }); - it("returns 404 for paths other than /health and /ws", async () => { + it("returns 404 for paths other than /health and /api", async () => { const res = await SELF.fetch("http://proxy.test/anything-else", { headers: { "x-test-id": freshId() }, }); From f02bbecfbae5eb93df49af2dcd164a082091ebbd Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 12:35:29 +0000 Subject: [PATCH 08/23] examples: Route the capnweb upgrade on /api The daemon and the host both serve the capnweb session on /api now, so the durable objects that route the container's outbound upgrade follow. Each example matched /ws in its fetch handler; the container example forwards every request to the backend and needed no change beyond its comments. --- examples/container/src/index.ts | 6 +++--- examples/mcp/src/index.test.ts | 2 +- examples/mcp/src/index.ts | 2 +- examples/think-compare-runtimes/worker/think/agents.ts | 2 +- examples/think/src/agent.ts | 8 ++++---- examples/think/src/index.ts | 2 +- examples/tutorial/src/index.ts | 2 +- 7 files changed, 12 insertions(+), 12 deletions(-) diff --git a/examples/container/src/index.ts b/examples/container/src/index.ts index 28510bd1..791cd58e 100644 --- a/examples/container/src/index.ts +++ b/examples/container/src/index.ts @@ -4,7 +4,7 @@ // The DO is a thin shell over CloudflareContainerBackend: it picks // the container (this.ctx.container) and the egress fetcher // (ctx.exports.WorkspaceProxy bound to this DO instance), forwards -// container-bound /ws upgrades back through the backend, and +// container-bound /api upgrades back through the backend, and // otherwise just calls into a single Workspace instance. // // Wire shape: @@ -14,7 +14,7 @@ // ▼ // ContainerExample DO ──► Container ──► computerd (:8080) // ▲ │ -// │ ws://computer.internal/ws │ +// │ ws://computer.internal/api │ // └─── capnweb session ◀─────────────┘ import { DurableObject, tracing } from "cloudflare:workers"; @@ -92,7 +92,7 @@ function workspaceOptions(self: InstanceType): WorkspaceOp // hands back round-trip into this DO; the actual SyncRPC + ShellRPC // traffic stays on the computerd ↔ DO capnweb wire. export class ContainerExample extends withWorkspace(ContainerBase, workspaceOptions) { - // ---- WebSocket: computerd's outbound /ws upgrade --------------------- + // ---- WebSocket: computerd's outbound /api upgrade -------------------- override async fetch(request: Request): Promise { return this.backend.handleFetch(request); diff --git a/examples/mcp/src/index.test.ts b/examples/mcp/src/index.test.ts index 8cbe9933..189e062e 100644 --- a/examples/mcp/src/index.test.ts +++ b/examples/mcp/src/index.test.ts @@ -26,7 +26,7 @@ describe("Computer Code Mode MCP", () => { expect(health.status).toBe(200); expect(await health.text()).toBe("ok\n"); - const internal = await SELF.fetch("https://example.test/ws"); + const internal = await SELF.fetch("https://example.test/api"); expect(internal.status).toBe(404); }); diff --git a/examples/mcp/src/index.ts b/examples/mcp/src/index.ts index e8f56f82..d546d4fe 100644 --- a/examples/mcp/src/index.ts +++ b/examples/mcp/src/index.ts @@ -60,7 +60,7 @@ export class ComputerMCP extends withWorkspace(ComputerMCPBase, workspaceOptions const path = new URL(request.url).pathname; // computerd reaches this callback through an internal binding. The public // Worker forwards only /mcp. - if (path === "/ws") return this.containerShell.handleFetch(request); + if (path === "/api") return this.containerShell.handleFetch(request); if (path !== "/mcp") return new Response("not found", { status: 404 }); const unauthorized = authorize(request, this.env.MCP_TOKEN); diff --git a/examples/think-compare-runtimes/worker/think/agents.ts b/examples/think-compare-runtimes/worker/think/agents.ts index 04425f0f..52adf962 100644 --- a/examples/think-compare-runtimes/worker/think/agents.ts +++ b/examples/think-compare-runtimes/worker/think/agents.ts @@ -260,7 +260,7 @@ export class WorkspaceThinkAgent extends RuntimeThinkAgent { override async fetch(request: Request): Promise { const url = new URL(request.url); - if (url.pathname === "/ws" && this.#activeBackend) { + if (url.pathname === "/api" && this.#activeBackend) { return this.#activeBackend.handleFetch(request); } return super.fetch(request); diff --git a/examples/think/src/agent.ts b/examples/think/src/agent.ts index 7a82b725..624af906 100644 --- a/examples/think/src/agent.ts +++ b/examples/think/src/agent.ts @@ -46,7 +46,7 @@ import { createWorkersAI } from "workers-ai-provider"; // WorkerShellBackend reaches WorkspaceServiceProxy through // `ctx.exports.WorkspaceServiceProxy(...)` so the in-isolate shell // can call back into the host workspace. WorkspaceProxy carries the -// container's outbound /ws egress back to this DO. +// container's outbound /api egress back to this DO. export { WorkspaceProxy, WorkspaceServiceProxy }; const MODEL_ID = "@cf/zai-org/glm-5.2"; @@ -73,7 +73,7 @@ export class Assistant extends withWorkspaceContainer(AssistantBase) { * Container backend used when `exec` needs a real Linux userland. * The DO itself owns the container binding through the * withWorkspaceContainer mixin; CloudflareContainerBackend handles - * startup, outbound egress interception, the /ws upgrade, and the + * startup, outbound egress interception, the /api upgrade, and the * capnweb session. */ readonly #containerBackend = new CloudflareContainerBackend({ @@ -103,10 +103,10 @@ export class Assistant extends withWorkspaceContainer(AssistantBase) { useThink: true, }) as Workspace & ThinkWorkspaceCompatibility; - /** Forwarded by WorkspaceProxy for computerd's outbound /ws upgrade. */ + /** Forwarded by WorkspaceProxy for computerd's outbound /api upgrade. */ override async fetch(request: Request): Promise { const url = new URL(request.url); - if (url.pathname === "/ws") { + if (url.pathname === "/api") { return this.#containerBackend.handleFetch(request); } return super.fetch(request); diff --git a/examples/think/src/index.ts b/examples/think/src/index.ts index 1fb31ee2..ac043217 100644 --- a/examples/think/src/index.ts +++ b/examples/think/src/index.ts @@ -11,7 +11,7 @@ * The Assistant, WorkspaceProxy, and WorkspaceServiceProxy classes are * re-exported so the runtime can resolve them by name: Assistant is * the DO binding and container class, WorkspaceProxy carries computerd's - * outbound /ws upgrade back to the DO, and WorkspaceServiceProxy is + * outbound /api upgrade back to the DO, and WorkspaceServiceProxy is * the loopback Fetcher the worker backend hands into its Dynamic * Worker so the in-isolate shell can reach back into the host * workspace. diff --git a/examples/tutorial/src/index.ts b/examples/tutorial/src/index.ts index 9a0aae33..c2b077f2 100644 --- a/examples/tutorial/src/index.ts +++ b/examples/tutorial/src/index.ts @@ -80,7 +80,7 @@ export class RecipeAgent extends withWorkspaceContainer(RecipeBase) { } override async fetch(request: Request): Promise { - return new URL(request.url).pathname === "/ws" + return new URL(request.url).pathname === "/api" ? this.#backend.handleFetch(request) : super.fetch(request); } From 0f2ae7493b6f7675c0777997ab2470616b0c6a58 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 12:35:29 +0000 Subject: [PATCH 09/23] script: Dial the capnweb session on /api The daemon serves its session on /api. The FUSE flush check and the stub soak both dial it directly, so both move with it. --- script/computerd-fuse-flush.mjs | 2 +- script/computerd-stub-soak.mjs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/script/computerd-fuse-flush.mjs b/script/computerd-fuse-flush.mjs index 420cf3c7..0f91f579 100755 --- a/script/computerd-fuse-flush.mjs +++ b/script/computerd-fuse-flush.mjs @@ -157,7 +157,7 @@ async function main() { `${REPO_ROOT}/node_modules/@cloudflare/computer-rpc/dist/sync-driver.js` ); - const wsUrl = `${container.url.replace("http://", "ws://")}/ws`; + const wsUrl = `${container.url.replace("http://", "ws://")}/api`; const client = createWorkspaceClient({ url: wsUrl }); const recvStorage = new SQLiteTestStorage(); diff --git a/script/computerd-stub-soak.mjs b/script/computerd-stub-soak.mjs index 2bfedc34..857cfe6d 100755 --- a/script/computerd-stub-soak.mjs +++ b/script/computerd-stub-soak.mjs @@ -236,7 +236,7 @@ async function main() { await waitForHealth(port, child); console.error("[soak] computerd healthy"); - const ws = new WebSocket(`ws://127.0.0.1:${port}/ws`); + const ws = new WebSocket(`ws://127.0.0.1:${port}/api`); await new Promise((res, rej) => { ws.once("open", res); ws.once("error", rej); From eaa4638a7d7ccb24b7092b4a888fa124a0deb869 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 12:39:23 +0000 Subject: [PATCH 10/23] script: Soak sync between a host and one computerd The soak booted two daemons and pointed UPSTREAM_URL at the first, then measured how fast it caught up. The second daemon was never the subject: what the script measured was a daemon driving its own sync loop, and driving a loop needs a peer to drive against. That mode is gone, and no deployment used it. A workspace pairs one host with one container, so the script now boots one daemon and takes the host's part. It holds the authoritative store, writes into it at the configured rate, and ticks pushOnce and pullOnce across the capnweb session. The table reports the host revision beside the daemon revision, and the gap between them is the convergence lag. Resident memory still gets sampled every row, which is the other reason this script exists. Revisions now come from GET /__computerd/watermarks rather than an RPC session per sample, and SOAK_TICK_MS sets how often the tick runs. --- script/computerd-soak.mjs | 286 ++++++++++++++++---------------------- 1 file changed, 119 insertions(+), 167 deletions(-) diff --git a/script/computerd-soak.mjs b/script/computerd-soak.mjs index fbbaf019..5f913339 100755 --- a/script/computerd-soak.mjs +++ b/script/computerd-soak.mjs @@ -1,37 +1,48 @@ #!/usr/bin/env node -// computerd-soak.mjs — soak test for the computerd sync loop. +// computerd-soak.mjs — soak test for the sync loop between a host and +// one computerd. // -// Boots two computerd containers wired as peer-to-peer: -// A: standalone computerd, port mapped to the host. -// B: standalone computerd, port mapped to the host AND -// UPSTREAM_URL pointing at A's host port. B's sync -// loop pulls from A and pushes to A. +// Boots one computerd container and plays the part the durable object +// plays in production: this script holds the authoritative store and +// drives sync across the capnweb session. Files are written into the +// host store at a steady rate while a tick loop pushes them to the +// daemon. While that runs, sample: // -// Hammers A by writing files through its /api endpoint -// (capnweb HTTP batch). While the writes flow, sample: +// - the host store's currentRev — how far the writer has got. +// - the daemon's currentRev — how far sync has carried it. +// - docker stats — resident memory for the container. // -// - A.watermarks() — currentRev, pushRev, fetchRev. -// - B.watermarks() — same. -// - docker stats — RSS for both containers. +// The gap between the two revision columns is the convergence lag. The +// memory column is the other reason this script exists: it is the +// cheapest way to watch the daemon's footprint under write pressure. // -// Output is a TSV table on stdout, one row per sample, -// suitable for piping into a CSV reader or just eyeballing. +// This used to boot a second daemon and point UPSTREAM_URL at the +// first. That measured a daemon driving its own sync loop, which is a +// mode no deployment uses and which no longer exists. A workspace pairs +// one host with one container, so that is what this soaks. +// +// Output is a TSV table on stdout, one row per sample, suitable for +// piping into a CSV reader or just eyeballing. // // Knobs (env vars): // -// COMPUTERD_BINARY path to computerd-linux-x64 binary +// COMPUTERD_BINARY path to computerd-linux-x64 binary // SOAK_DURATION_MS total wall time of the soak phase (default 30000) -// SOAK_WRITES_PER_S target writes/second sustained against A (default 200) +// SOAK_WRITES_PER_S target writes/second sustained (default 200) // SOAK_PAYLOAD_B bytes per write (default 64) // SOAK_SAMPLE_MS sampling interval (default 250) +// SOAK_TICK_MS sync tick interval (default 100) import { execFile, spawn } from "node:child_process"; import { existsSync } from "node:fs"; import { dirname, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import { promisify } from "node:util"; -import { newHttpBatchRpcSession, newWebSocketRpcSession } from "capnweb"; +import { createWorkspaceClient } from "@cloudflare/computer-rpc/client"; +import { pullOnce, pushOnce } from "@cloudflare/computer-rpc/driver"; +import { currentRev, Database, initializeSchema, SQLiteWorkspaceProvider } from "@cloudflare/dofs"; +import { SQLiteTestStorage } from "@cloudflare/dofs/testing"; import WebSocket from "ws"; const execFileP = promisify(execFile); @@ -55,10 +66,11 @@ const SAMPLE_MS = Number(process.env.SOAK_SAMPLE_MS ?? 250); // skipped. const DISABLE_FUSE = process.env.SOAK_DISABLE_FUSE === "1" || !existsSync("/dev/fuse"); -// On Linux, host.docker.internal isn't resolved automatically; -// we map it to the host-gateway address so B can reach A. -// docker-desktop on macOS/Windows already provides this. -const ADD_HOST_GATEWAY = process.platform === "linux"; +// Matches MOUNT_POINT in the container env below. The store keeps +// everything under the mount point so pushed paths land where the FUSE +// mount expects them. +const MOUNT_POINT = "/workspace"; +const TICK_MS = Number(process.env.SOAK_TICK_MS ?? 100); const IMAGE_TAG = "computerd-harness:libfuse2"; @@ -90,7 +102,7 @@ RUN apt-get update >/dev/null && apt-get install -y --no-install-recommends \\ }); } -async function bootContainer(extraEnv = {}) { +async function bootContainer() { const args = ["run", "--rm", "-d", "--platform", "linux/amd64"]; if (!DISABLE_FUSE) { args.push( @@ -107,9 +119,6 @@ async function bootContainer(extraEnv = {}) { "seccomp=unconfined", ); } - if (ADD_HOST_GATEWAY) { - args.push("--add-host", "host.docker.internal:host-gateway"); - } args.push( "-v", `${BINARY}:/usr/local/bin/computerd:ro`, @@ -118,14 +127,11 @@ async function bootContainer(extraEnv = {}) { "-e", "PORT=8080", "-e", - "MOUNT_POINT=/workspace", + `MOUNT_POINT=${MOUNT_POINT}`, ); if (DISABLE_FUSE) { args.push("-e", "FUSE_MOUNT=none"); } - for (const [k, v] of Object.entries(extraEnv)) { - args.push("-e", `${k}=${v}`); - } const image = DISABLE_FUSE ? "debian:stable-slim" : IMAGE_TAG; args.push(image, "/usr/local/bin/computerd"); const { stdout } = await execFileP("docker", args); @@ -174,16 +180,12 @@ async function dockerStats(cids) { return result; } -// Connect a capnweb HTTP batch client to /api. Each call is -// its own session; we accept the cost (one round-trip per -// hammer-write) so the soak measures the entire stack -// including session setup. -function batchStub(url) { - return newHttpBatchRpcSession(`${url}/api`); -} - +// The daemon's revision numbers over plain HTTP. A sampler wanting +// three integers on an interval does not need an RPC session. async function fetchWatermarks(url) { - return await batchStub(url).sync.watermarks(); + const res = await fetch(`${url}/__computerd/watermarks`); + if (!res.ok) throw new Error(`watermarks HTTP ${res.status}`); + return await res.json(); } // Build a payload-bytes Uint8Array. @@ -207,149 +209,108 @@ function payloadBytes(seed) { return out; } -// Persistent capnweb WebSocket session against B. Reused -// across the soak; one upgrade, many push round-trips. +// Persistent capnweb session against the daemon. Reused across the +// soak: one upgrade, many push round-trips. // -// SOAK_NO_DEFLATE=1 forces the dial to negotiate without -// permessage-deflate so the soak can compare compressed and -// uncompressed wire costs without rebuilding the computerd binary. -function wsStub(url) { - const wsUrl = `${url.replace("http://", "ws://")}/ws`; - const ws = new WebSocket(wsUrl, { - perMessageDeflate: process.env.SOAK_NO_DEFLATE !== "1", +// SOAK_NO_DEFLATE=1 dials without permessage-deflate so the soak can +// compare compressed and uncompressed wire costs without rebuilding +// the computerd binary. +function openSession(url) { + return createWorkspaceClient({ + url: `${url.replace("http://", "ws://")}/api`, + WebSocketImpl: class extends WebSocket { + constructor(target) { + super(target, { perMessageDeflate: process.env.SOAK_NO_DEFLATE !== "1" }); + } + }, }); - return newWebSocketRpcSession(ws); } -// One write into computerd via the SyncRPC push path. senderRev=0 -// marks us as an external writer — the server applies as -// a local write (bumps vfs_meta.rev, leaves pushRev alone) so -// the outbound sync loop picks the entry up on the next tick. -// This is the F2 fix in practice; before it landed, this -// path silenced the outbound loop and A never saw any of B's -// writes. -async function pushOneWrite(stub, i, bytes) { - const hash = new Uint8Array(await crypto.subtle.digest("SHA-256", bytes)); - await stub.sync.pushObjects( - new ReadableStream({ - start(c) { - c.enqueue({ hash, bytes }); - c.close(); - }, - }), - ); - await stub.sync.push({ - senderRev: 0, - changes: new ReadableStream({ - start(c) { - c.enqueue({ - kind: "file", - path: `/soak_${i}.bin`, - mode: 0o644, - mtime: Date.now(), - size: bytes.byteLength, - chunks: [{ hash, size: bytes.byteLength }], - }); - c.close(); - }, - }), - }); +// The authoritative store, standing in for the durable object's SQLite. +function openHostStore() { + const db = new Database(new SQLiteTestStorage()); + initializeSchema(db, () => Date.now()); + return { db, provider: new SQLiteWorkspaceProvider(db, { now: () => Date.now() }) }; +} + +// One write into the host store. The sync tick below is what carries +// it to the daemon, so this is a plain filesystem write rather than a +// wire call. +function writeOne(provider, i, bytes) { + provider.writeFileSync(`${MOUNT_POINT}/soak_${i}.bin`, Buffer.from(bytes)); } async function main() { if (!DISABLE_FUSE) await ensureImage(); - process.stderr.write("booting A (sink) ...\n"); - const a = await bootContainer(); - process.stderr.write(` A: ${a.url} (${a.cid.slice(0, 12)})\n`); + process.stderr.write("booting computerd ...\n"); + const daemon = await bootContainer(); + process.stderr.write(` ${daemon.url} (${daemon.cid.slice(0, 12)})\n`); - process.stderr.write("booting B (source, UPSTREAM_URL -> A) ...\n"); - // B's sync loop will push to A. Hostname inside docker: - // we can't reach the host's 127.0.0.1 portably; use - // host.docker.internal which docker-desktop sets up on - // macOS/Windows. On linux we'd need --add-host=host.docker.internal:host-gateway. - // Inside a docker container we reach the host's mapped - // port via host.docker.internal. The capnweb client needs - // a ws:// URL pointing at the /ws endpoint (not just the - // host). - const upstreamForB = `${a.url.replace("http://127.0.0.1", "ws://host.docker.internal")}/ws`; - // Inside the docker container, the host port we're trying - // to reach is 127.0.0.1: on the host. Pass the - // mapped host port via host.docker.internal. - const b = await bootContainer({ - UPSTREAM_URL: upstreamForB, - }); - process.stderr.write(` B: ${b.url} (${b.cid.slice(0, 12)}) -> upstream ${upstreamForB}\n`); + const { db, provider } = openHostStore(); + provider.mkdirSync(MOUNT_POINT, { recursive: true }); + const session = openSession(daemon.url); - // Header row. - console.log( - "t_ms\tA_currentRev\tA_pushRev\tA_fetchRev\tB_currentRev\tB_pushRev\tB_fetchRev\tA_mem\tB_mem\twrites_sent", - ); + // Header row. host_rev is how far the writer has got; daemon_rev is + // how far sync has carried it. The difference is the lag. + console.log("t_ms\thost_rev\tdaemon_rev\tdaemon_pushRev\tmem\twrites_sent\tpushed"); const start = Date.now(); const stopAt = start + DURATION_MS; const intervalMs = Math.max(1, Math.floor(1000 / WRITES_PER_S)); let writeSeq = 0; let writesSent = 0; - let writesInFlight = 0; - // Writes go through B's SyncRPC /ws push path with - // senderRev=0. The server treats them as local writes; - // B's outbound sync loop ships them to A on the next - // tick. This is the path an external orchestrator (a - // DO accepting agent requests, the agent itself) would - // take — the same wire surface a computerd-to-computerd peer - // uses, just with a different senderRev value. - const writeStub = wsStub(b.url); - - // Fire-and-forget write loop. We don't await every write - // because the goal is to saturate; we cap the in-flight - // count to keep memory bounded. - const MAX_INFLIGHT = 32; + // Writes land in the host store. The sync tick below carries them + // over the wire, which is the direction production runs: the host + // owns the truth and pushes it to the container. const writeLoop = (async () => { while (Date.now() < stopAt) { - if (writesInFlight >= MAX_INFLIGHT) { - await new Promise((r) => setTimeout(r, 1)); - continue; - } - const i = writeSeq++; - writesInFlight++; - pushOneWrite(writeStub, i, payloadBytes(i)) - .then(() => { - writesSent++; - writesInFlight--; - }) - .catch((err) => { - writesInFlight--; - process.stderr.write(`write ${i} failed: ${err.message}\n`); - }); + writeOne(provider, writeSeq++, payloadBytes(writeSeq)); + writesSent++; await new Promise((r) => setTimeout(r, intervalMs)); } })(); - // Sample loop. Runs in parallel with the writes. + // Sync tick. pushOnce ships whatever the writer has committed since + // the last tick; pullOnce brings back anything the daemon changed on + // its own, which is nothing here but exercises the other direction. + // A failed tick logs and continues: watermarks are durable, so the + // next tick resumes where this one stopped. + let pushed = 0; + let ticking = false; + const tickTimer = setInterval(() => { + if (ticking) return; + ticking = true; + (async () => { + pushed += await pushOnce(db, session.sync); + await pullOnce(db, session.sync); + })() + .catch((err) => { + process.stderr.write(`sync tick failed: ${err.message}\n`); + }) + .finally(() => { + ticking = false; + }); + }, TICK_MS); + + // Sample loop. Runs in parallel with the writes and the tick. const samples = []; const sampleLoop = (async () => { while (Date.now() < stopAt + 5000) { const t = Date.now() - start; - const [aWm, bWm, stats] = await Promise.all([ - fetchWatermarks(a.url).catch(() => null), - fetchWatermarks(b.url).catch(() => null), - dockerStats([a.cid, b.cid]).catch(() => ({})), + const [wm, stats] = await Promise.all([ + fetchWatermarks(daemon.url).catch(() => null), + dockerStats([daemon.cid]).catch(() => ({})), ]); - const aMem = stats[a.cid.slice(0, 12)] ?? "?"; - const bMem = stats[b.cid.slice(0, 12)] ?? "?"; const row = [ t, - aWm?.currentRev ?? -1, - aWm?.pushRev ?? -1, - aWm?.fetchRev ?? -1, - bWm?.currentRev ?? -1, - bWm?.pushRev ?? -1, - bWm?.fetchRev ?? -1, - aMem, - bMem, + currentRev(db), + wm?.currentRev ?? -1, + wm?.pushRev ?? -1, + stats[daemon.cid.slice(0, 12)] ?? "?", writesSent, + pushed, ]; console.log(row.join("\t")); samples.push(row); @@ -358,31 +319,22 @@ async function main() { })(); await writeLoop; - process.stderr.write(`writes done (${writesSent} sent, ${writesInFlight} still in flight)\n`); - // Let in-flight writes drain. - while (writesInFlight > 0) await new Promise((r) => setTimeout(r, 100)); - // Let the sync loop catch up. + process.stderr.write(`writes done (${writesSent} committed locally)\n`); + // Let the tick drain what the writer just committed. await new Promise((r) => setTimeout(r, 3000)); await sampleLoop; - - // capnweb's WebSocket session doesn't expose an explicit - // close on the stub; the container kill below tears it - // down at the computerd end. - await Promise.all([kill(a.cid), kill(b.cid)]); + clearInterval(tickTimer); + await session.close().catch(() => {}); + await kill(daemon.cid); // Summary on stderr. const final = samples[samples.length - 1] ?? []; process.stderr.write(`\n--- soak complete ---\n`); - process.stderr.write(`writes attempted: ${writeSeq}\n`); - process.stderr.write(`writes acked: ${writesSent}\n`); - process.stderr.write(`final B.currentRev: ${final[4]}\n`); - process.stderr.write( - `final B.pushRev: ${final[5]} (gap to currentRev: ${final[4] - final[5]})\n`, - ); - process.stderr.write( - `final A.fetchRev: ${final[3]} (lag behind B.currentRev: ${final[4] - final[3]})\n`, - ); - process.stderr.write(`final B mem: ${final[8]}\n`); + process.stderr.write(`writes committed: ${writeSeq}\n`); + process.stderr.write(`entries pushed: ${pushed}\n`); + process.stderr.write(`final host rev: ${final[1]}\n`); + process.stderr.write(`final daemon rev: ${final[2]} (lag: ${final[1] - final[2]})\n`); + process.stderr.write(`final mem: ${final[4]}\n`); } main().catch((err) => { From 78e9e72af391644ce37d8ae7d8e17ba628f51376 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 12:43:15 +0000 Subject: [PATCH 11/23] docs: Describe the single /api endpoint Bring the written surface in line with the daemon. It serves one endpoint, /api, carrying one transport, and the HTTP batch alternative is gone: every call that returns data returns a ReadableStream, and delivering one needs a call from server to client that a batch session cannot complete. UPSTREAM_URL is removed from the environment tables and from the prose that described a daemon-driven sync loop. The lifecycle document claimed the Cloudflare backend wired that variable automatically, which was never true; the carrier is POST /connect and the durable object drives sync across it. That sentence is the one a reader lands on when working out how a container recovers after a restart, so it pointed at the wrong mechanism. The /connect description now shows the request naming a base and both paths, records the readiness probe reading base plus health, and covers the new watermarks endpoint and the handshake statuses. --- .../skills/debugging-computerd-fuse/SKILL.md | 14 +++++----- docs/07_injected_service.md | 26 +++++++++---------- docs/08_capnweb_interface.md | 9 ++++--- docs/11_lifecycle.md | 14 +++++----- docs/README.md | 2 +- examples/container/README.md | 13 +++++----- examples/tutorial/README.md | 4 +-- .../computer-computerd-linux-x64/README.md | 1 - packages/computerd/README.md | 17 ++++++------ 9 files changed, 51 insertions(+), 49 deletions(-) diff --git a/.agents/skills/debugging-computerd-fuse/SKILL.md b/.agents/skills/debugging-computerd-fuse/SKILL.md index c609979d..fca758fe 100644 --- a/.agents/skills/debugging-computerd-fuse/SKILL.md +++ b/.agents/skills/debugging-computerd-fuse/SKILL.md @@ -1,6 +1,6 @@ --- name: debugging-computerd-fuse -description: Debug computerd in real-FUSE mode end-to-end without workerd, vitest-pool-workers, or wrangler in the loop. Boot the linux-x64 binary in a privileged docker container, drive its capnweb /ws endpoint from Node, simulate DO-side sync from a SQLiteTestStorage, and isolate FUSE-related deadlocks. Load when a real-FUSE bug reproduces locally but unit tests pass, when the harness vitest tests hang against a real container, or when you need to attribute a wedge to FUSE vs sync vs exec. +description: Debug computerd in real-FUSE mode end-to-end without workerd, vitest-pool-workers, or wrangler in the loop. Boot the linux-x64 binary in a privileged docker container, drive its capnweb /api endpoint from Node, simulate DO-side sync from a SQLiteTestStorage, and isolate FUSE-related deadlocks. Load when a real-FUSE bug reproduces locally but unit tests pass, when the harness vitest tests hang against a real container, or when you need to attribute a wedge to FUSE vs sync vs exec. --- # Debugging computerd against real FUSE @@ -87,10 +87,12 @@ with a missing `/dev/fuse` would have failed startup outright. ## Drive computerd from a Node script -computerd serves a composite `WorkspaceRPC` over `/ws` (capnweb WebSocket) -and `/api` (capnweb HTTP batch). The `@cloudflare/computer-rpc/client` -package wraps the WS form and the `/driver` subpath exposes -`pushOnce`/`pullOnce` against a Node-side `Database`. +computerd serves a composite `WorkspaceRPC` over `/api`, a capnweb +WebSocket. That is the only RPC carrier: capnweb's HTTP batch transport +cannot deliver a returned stream, which is what every read on this +interface is. The `@cloudflare/computer-rpc/client` package wraps the +socket and the `/driver` subpath exposes `pushOnce`/`pullOnce` against a +Node-side `Database`. Set up a probe project once: @@ -123,7 +125,7 @@ import { pullOnce, pushOnce } from "@cloudflare/computer-rpc/driver"; import { WebSocket } from "ws"; const url = process.env.COMPUTERD_URL; // e.g. http://127.0.0.1:18080 -const wsUrl = `${url.replace(/^http(s?):\/\//, "ws$1://")}/ws`; +const wsUrl = `${url.replace(/^http(s?):\/\//, "ws$1://")}/api`; const storage = new SQLiteTestStorage(); const db = new Database(storage); diff --git a/docs/07_injected_service.md b/docs/07_injected_service.md index c323f754..93fa95f5 100644 --- a/docs/07_injected_service.md +++ b/docs/07_injected_service.md @@ -32,9 +32,8 @@ staging the binary into a container image. same paths. The backend is picked by `FUSE_MOUNT` (default `auto`, see the env-var table below). 2. **Dirty tracking.** Writes that flow through FUSE land in the - in-container VFS database; sync (when `UPSTREAM_URL` is set) is - what surfaces those revisions back out. See doc 02 for the sync - protocol. + in-container VFS database; the host pulls those revisions back out + across the capnweb session. See doc 02 for the sync protocol. 3. **Exec.** Runs shell commands and streams stdout/stderr back over capnweb. See [05. Shell Interface](./05_runtime_interface.md). 4. **Apply.** Accepts changes pushed by the DO and writes them into @@ -52,9 +51,9 @@ backend pins it to `8080`) and serves: | --- | --- | --- | | `/health` | `GET`, `HEAD` | Liveness probe; `200 ok\n` as soon as the HTTP server binds. | | `/__computerd/info` | `GET` | Runtime info: FUSE backend, mount point, port. | -| `/api` | `POST` | HTTP-batch capnweb transport. | -| `/ws` | `GET` (upgrade) | WebSocket capnweb transport — the bootstrap stub is `WorkspaceRPC`. | -| `/connect` | `POST` | Tells `computerd` to dial *out* to a caller-supplied URL and serve a `WorkspaceRPC` session over that outbound WebSocket. Used by the Cloudflare backend (see below). | +| `/__computerd/watermarks` | `GET` | Sync revisions: `currentRev`, `pushRev`, `fetchCursor`. | +| `/api` | `GET` (upgrade) | WebSocket capnweb transport — the bootstrap stub is `WorkspaceRPC`. A request without an `Upgrade` header gets `400`; an unsupported `Sec-WebSocket-Version` gets `426` and the versions the server speaks. | +| `/connect` | `POST` | Tells `computerd` to dial *out* to a caller-supplied endpoint and serve a `WorkspaceRPC` session over that outbound WebSocket. Used by the Cloudflare backend (see below). | | `/` | `GET` | Banner/info page. | The capnweb bootstrap interface is **`WorkspaceRPC`** (defined in @@ -108,9 +107,9 @@ Provider-agnostic shape — three steps, in order: path the mount is awaited *before* `listen`, so by the time `/health` answers FUSE is up too. With `FUSE_MOUNT=none` there is no FUSE step at all. -3. **Open the capnweb session.** Either the host upgrades to `/ws` - directly, or it asks `computerd` (via `POST /connect`) to dial *out* to a - URL it controls and serve the session over that outbound socket. +3. **Open the capnweb session.** Either the host upgrades to `/api` + directly, or it asks `computerd` (via `POST /connect`) to dial *out* to an + endpoint it controls and serve the session over that outbound socket. Either way, the bootstrap stub is `WorkspaceRPC`. ### Cloudflare Containers specifics @@ -132,8 +131,10 @@ wires it like this: repeated until it returns `200`. 4. **Invert the WebSocket.** The DO arms an upgrade slot (`#armUpgrade`) and then `POST`s to `/connect` on the container - (`#postConnect`). `computerd` reads that request and dials *out* to the - egress at `${egressHost}/ws`. Because the egress is intercepted, + (`#postConnect`). The request names the egress base and both paths, + so `computerd` polls `base + health` and then dials `base + api`; + the daemon assembles no paths of its own. Because the egress is + intercepted, that outbound dial loops back to the DO's `handleFetch()`, which accepts the upgrade and resolves the in-flight `#pendingUpgrade`. The capnweb session then runs over that socket. **The WebSocket @@ -154,10 +155,9 @@ These are the variables `computerd` actually consumes (see | Variable | Default | Meaning | | --- | --- | --- | -| `PORT` | `45678` | Port the HTTP/WS server listens on. CF backend pins this to `8080`. | +| `PORT` | `45678` | Port the HTTP server listens on. CF backend pins this to `8080`. | | `MOUNT_POINT` | `/workspace` | Absolute path inside the container to mount the FUSE filesystem at. Ignored when `FUSE_MOUNT=none`. | | `FUSE_MOUNT` | `auto` | Backend selector: `auto` probes `/dev/fuse` (linux) or macFUSE (darwin) and falls back to the userspace shim; `fuse` / `macfuse` require the corresponding real backend; `shim` forces the userspace shim; `none` skips the mount entirely. | -| `UPSTREAM_URL` | unset | If set, `computerd` starts a sync client against this URL to push/pull VFS revisions. | | `EXEC_LOG_MAX_BYTES` | runner default | Caps the per-exec stdout/stderr log retained in-memory. | | `LOG_FILE` | unset | If set, every `console.log` / `console.error` line and any `uncaughtException` / `unhandledRejection` is also appended to this file. Stdout/stderr behaviour is unchanged. | diff --git a/docs/08_capnweb_interface.md b/docs/08_capnweb_interface.md index c4ed867a..e51e389a 100644 --- a/docs/08_capnweb_interface.md +++ b/docs/08_capnweb_interface.md @@ -12,10 +12,11 @@ alternative). The interface served is `WorkspaceRPC`, defined in ## Transport -- **Carrier.** One long-lived WebSocket per Workspace. The DO opens it - against the computerd's `/ws` endpoint, with `/api` available - as an HTTP-batch alternative (single POST per call) for callers that - can't hold a socket. Default port is `45678`; it will become a +- **Carrier.** One long-lived WebSocket per Workspace, on the + computerd's `/api` endpoint. There is no HTTP alternative: every call + that returns data returns a `ReadableStream`, and delivering one + needs a call from server to client, which capnweb's HTTP batch + transport cannot complete. Default port is `45678`; it will become a build-time variable so hosts can pin a non-default port. - **Framing.** capnweb text frames. Binary frames are unsupported. **(planned)** the server will fail the session loudly on the first diff --git a/docs/11_lifecycle.md b/docs/11_lifecycle.md index 4166a907..9e72d830 100644 --- a/docs/11_lifecycle.md +++ b/docs/11_lifecycle.md @@ -24,7 +24,7 @@ capnweb WebSocket session. │ │ Workspace │ │ │ │ computerd │ │ │ │ fs: WorkspaceFS │ │ │ │ HTTP server │ │ │ │ shell: ShellRPC │ │ │ │ /health │ │ -│ │ sync: SyncRPC │◀─┼── capnweb WS ───▶│ │ /connect /ws │ │ +│ │ sync: SyncRPC │◀─┼── capnweb WS ───▶│ │ /connect /api │ │ │ │ push() / pull() │ │ │ │ FUSE mount │ │ │ │ ready() │ │ │ │ exec runner │ │ │ └──────────┬──────────┘ │ │ └────────┬─────────┘ │ @@ -50,7 +50,7 @@ The 1:1 mapping is load-bearing for several reasons: doesn't have to multiplex multiple WS peers. The DO is the WebSocket *server* in this pairing, even though `computerd` -exposes its own `/ws` server-side and could be dialed directly. The +exposes its own `/api` server-side and could be dialed directly. The reason for the inversion is documented in [07. Injected Service §Bootstrap sequence](./07_injected_service.md): the egress interceptor needs to be wired before any traffic flows, and @@ -90,7 +90,7 @@ an incarnation boundary. What survives is: On every new incarnation `Workspace.ready()` re-runs `#connect()`, which re-enters the backend's bootstrap sequence. If the container is -still alive, the backend's `POST /connect` + `/ws` handshake produces +still alive, the backend's `POST /connect` + `/api` handshake produces a fresh capnweb session against the same in-memory VFS on the container side. If the container died too (e.g. host OOM took both), the next sync round is a rev-0 baseline rebuild from the DO's store. @@ -129,9 +129,9 @@ that point so the next call rebuilds from scratch (see the container host and ba The critical asymmetry: the **container's VFS is process-lifetime in-memory**, while the **DO's VFS is durable SQLite**. A container -restart loses container-side state. Sync via `UPSTREAM_URL` (which -the Cloudflare backend wires automatically) is what brings state back -on the next push/pull round. +restart loses container-side state. The durable object drives sync +across the capnweb session it opens through `POST /connect`, and that +is what brings state back on the next push/pull round. ## Capnweb lifecycle @@ -373,7 +373,7 @@ between agent turns, and it's exactly where hibernation pays off. Adopting PartySocket for reconnect/backoff would require the DO to be the WebSocket *client* dialing `computerd`'s -`/ws` endpoint. That model is appealing for reconnect, but +`/api` endpoint. That model is appealing for reconnect, but hibernatable WebSockets only work server-side via `ctx.acceptWebSocket()` — there is no hibernation API for outbound client sockets. **Inverting the dial direction permanently forecloses diff --git a/docs/README.md b/docs/README.md index b36ce6ef..0a359ce3 100644 --- a/docs/README.md +++ b/docs/README.md @@ -87,7 +87,7 @@ To build the binary from source instead, run `npm run build:bin `artifacts/computerd/computerd-linux-x64`, then `COPY` that into the image. -`computerd`'s own default port is `45678`; the Cloudflare container backend pins the in-image listener to `8080`, which is what `examples/container/` uses. See [07. Injected Service](./07_injected_service.md) for the env vars (`PORT`, `MOUNT_POINT`, `FUSE_MOUNT`, `UPSTREAM_URL`, `EXEC_LOG_MAX_BYTES`) and the reverse-dial boot sequence. +`computerd`'s own default port is `45678`; the Cloudflare container backend pins the in-image listener to `8080`, which is what `examples/container/` uses. See [07. Injected Service](./07_injected_service.md) for the env vars (`PORT`, `MOUNT_POINT`, `FUSE_MOUNT`, `EXEC_LOG_MAX_BYTES`) and the reverse-dial boot sequence. ## Example diff --git a/examples/container/README.md b/examples/container/README.md index 0e813bc3..9f6f220b 100644 --- a/examples/container/README.md +++ b/examples/container/README.md @@ -21,7 +21,7 @@ client ─► Worker /c//{file,exec} ▼ DO (ContainerExample) ──► Container ──► computerd (:8080) ▲ │ - │ ws://computer.internal/ws │ + │ ws://computer.internal/api │ └────────── capnweb session ◄──────┘ ``` @@ -30,19 +30,20 @@ client ─► Worker /c//{file,exec} `Workspace` instance. That backend owns the entire computerd lifecycle: container start, outbound egress interception, port-readiness polling, POST - `/connect` to computerd, `/ws` upgrade routing, and capnweb session + `/connect` to computerd, `/api` upgrade routing, and capnweb session attach. 2. computerd reaches the Worker through the container's **outbound interception** (`ctx.container.interceptOutboundHttp("computer.internal", …)`, set up by the backend). The DO passes `ctx.exports.WorkspaceProxy({ props: { binding, id } })` as the egress fetcher; that `WorkerEntrypoint` (re-exported from - `@cloudflare/computer`) routes `/ws` upgrades back to the owning DO. + `@cloudflare/computer`) routes `/api` upgrades back to the owning DO. 3. When `Workspace.ready()` is called for the first time, the backend posts `/connect` into computerd with - `{ url: "http://computer.internal" }`. computerd polls - `computer.internal/health`, then dials - `ws://computer.internal/ws`. + `{ base: "http://computer.internal", health: "/health", api: "/api" }`. + computerd polls `computer.internal/health`, then dials + `ws://computer.internal/api`. Naming both paths in the request keeps + the daemon from assembling routes it does not serve. 4. `WorkspaceProxy.fetch` forwards the upgrade to the DO's `fetch()` via the DO binding looked up from its props. The DO's `fetch()` delegates to `backend.handleFetch(req)`, which performs the diff --git a/examples/tutorial/README.md b/examples/tutorial/README.md index 47015283..aedf4f17 100644 --- a/examples/tutorial/README.md +++ b/examples/tutorial/README.md @@ -164,7 +164,7 @@ export class RecipeAgent extends withWorkspaceContainer(RecipeBase) { }) as Workspace & ThinkWorkspaceCompatibility; override async fetch(request: Request): Promise { - return new URL(request.url).pathname === "/ws" + return new URL(request.url).pathname === "/api" ? this.#backend.handleFetch(request) : super.fetch(request); } @@ -179,7 +179,7 @@ network access. the durable object can start and stop its own container. The `workspace: { binding, id }` pair is how the container finds its way home: it dials the named binding at that id, which is why `fetch` has to -hand `/ws` to the backend before the base class sees it. +hand `/api` to the backend before the base class sees it. ## 5. Hook the workspace up to the agent diff --git a/packages/computer-computerd-linux-x64/README.md b/packages/computer-computerd-linux-x64/README.md index 1f4254bf..e508f0ce 100644 --- a/packages/computer-computerd-linux-x64/README.md +++ b/packages/computer-computerd-linux-x64/README.md @@ -49,4 +49,3 @@ examples that intentionally track release candidates. | `PORT` | `8080` | HTTP + WebSocket listener port. | | `MOUNT_POINT` | `/workspace` | Path the FUSE filesystem mounts at. | | `FUSE_MOUNT` | `auto` | Backend selector. `auto` probes `/dev/fuse` (linux) or macFUSE (darwin) and falls back to the userspace shim. `fuse` and `macfuse` require their respective real backend. `shim` forces the userspace shim. `none` skips the mount entirely; HTTP / WS still come up. | -| `UPSTREAM_URL` | unset | If set, computerd dials this WebSocket on boot and runs a bidirectional sync loop against it. | diff --git a/packages/computerd/README.md b/packages/computerd/README.md index 4d18d696..138a9ac1 100644 --- a/packages/computerd/README.md +++ b/packages/computerd/README.md @@ -28,8 +28,8 @@ Current endpoints: - `GET /__computerd/info` returns JSON with the selected FUSE backend, mount point, and bound port. - `GET /__computerd/stats` returns JSON with DOFS table row counts, total inline and blob byte sizes, the orphan-blob subset, and process resident memory. Useful for watching how the store grows under load. - `GET /` returns `200 OK` with an empty JSON object: `{}`. -- `POST /api` is a capnweb HTTP-batch RPC endpoint backed by `@cloudflare/computer-rpc`. Non-POST methods return `405`. -- `GET /ws` upgrades to a WebSocket carrying the same capnweb RPC surface. This is the container's primary sync carrier. +- `GET /__computerd/watermarks` returns JSON with `currentRev`, `pushRev`, and `fetchCursor`, read through the same `watermarks()` the wire serves. For samplers that want a few numbers without opening a session. +- `GET /api` upgrades to a WebSocket carrying the capnweb RPC surface backed by `@cloudflare/computer-rpc`. This is the container's only RPC carrier. A request without an `Upgrade` header returns `400`; a handshake naming an unsupported `Sec-WebSocket-Version` returns `426` along with the versions the server speaks. All other paths and methods return `404`/`405` with a `text/plain` body. @@ -38,9 +38,9 @@ Current filesystem support: - `@platformatic/vfs` in-memory filesystem provided by `@cloudflare/dofs`'s node provider. - FUSE operation adapter covering the full `fuse-native` operation surface. - Unsupported FUSE operations return `ENOSYS` to the kernel; the binding logs a one-shot warning per operation. -- capnweb RPC over `/api` and `/ws` exposes the workspace database and an `exec` runner to clients. -- Optional host/DO synchronization: when `UPSTREAM_URL` is set, `computerd` opens a `SyncClient` from `@cloudflare/computer-rpc/client` against that URL and runs the sync loop in the background. -- No on-disk persistence yet — the in-memory VFS is rebuilt on each start, with sync pulling state back from the upstream when configured. +- capnweb RPC over `/api` exposes the workspace database and an `exec` runner to clients. +- Synchronization is driven by whoever holds the other end of the session. The daemon serves `SyncRPC`; it does not run a sync loop of its own. +- No on-disk persistence yet — the in-memory VFS is rebuilt on each start, and the host pushes state back after a restart. ## FUSE write model @@ -101,14 +101,13 @@ FUSE_MOUNT=auto # default: probe /dev/fuse or macFUSE, fall back to the users FUSE_MOUNT=fuse # require the linux kernel FUSE backend (/dev/fuse) FUSE_MOUNT=macfuse # require macFUSE on darwin FUSE_MOUNT=shim # force the userspace dev shim (no FUSE) -FUSE_MOUNT=none # skip the mount entirely; HTTP + /api + /ws still come up +FUSE_MOUNT=none # skip the mount entirely; HTTP and /api still come up ``` Additional environment variables: ```sh -UPSTREAM_URL=https://example/ws # open a SyncClient against this capnweb endpoint -EXEC_LOG_MAX_BYTES=1048576 # cap the in-memory exec log buffer (bytes) +EXEC_LOG_MAX_BYTES=1048576 # cap the in-memory exec log buffer (bytes) ``` `FUSE_MOUNT=auto` is the friendly default: if `/dev/fuse` (or macFUSE) is available `computerd` mounts a real FUSE filesystem, otherwise it transparently falls back to the userspace shim. Pin the value (`fuse` / `macfuse` / `shim` / `none`) when a test needs to assert a specific code path. @@ -124,7 +123,7 @@ How it works: - A periodic poll (~250 ms) walks `MOUNT_POINT`, diffs it against a content-hash shadow, and pushes any new or changed entries into the VFS. - The shadow doubles as a loop suppressor: after a write in either direction the shadow matches both sides, so the next tick on the opposite side sees no diff. -`exec` runs with `cwd=MOUNT_POINT` exactly as it does under real FUSE, so a child process that writes into the mount point ends up writing through the shim into the VFS — and onward to the DO when `UPSTREAM_URL` is set. +`exec` runs with `cwd=MOUNT_POINT` exactly as it does under real FUSE, so a child process that writes into the mount point ends up writing through the shim into the VFS, and onward to the host on its next pull. Caveats. The shim is dev-only: From 76863af3cc2b836c9b62ddf96c3f121075aca2c1 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 12:44:00 +0000 Subject: [PATCH 12/23] chore: Add a changeset for the /api endpoint move --- .changeset/one-endpoint-on-api.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/one-endpoint-on-api.md diff --git a/.changeset/one-endpoint-on-api.md b/.changeset/one-endpoint-on-api.md new file mode 100644 index 00000000..81bf03cb --- /dev/null +++ b/.changeset/one-endpoint-on-api.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": minor +--- + +The capnweb session now runs on `/api` at both ends of the connection, replacing `/ws`. A durable object that routes the container's outbound upgrade must match `/api` in its `fetch()` handler; `WorkspaceProxy` already does. `POST /connect` carries the endpoint as `{ base, health, api }` so the container assembles no host paths of its own, and a request to `/api` without an `Upgrade` header answers `400` rather than upgrading. From 64df4351337d37dbd6df07f50e349474c9b20012 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 14:20:55 +0000 Subject: [PATCH 13/23] computer: Rename the tokenized callback segment to /api The tokenized connect callback matched /__workspace_connect//ws while everything else moved to /api, leaving the proxy answering two spellings for one endpoint. It now matches /api, so both the plain and tokenized forms read the same and normalize to the same path before reaching the durable object. Nothing in this repository builds those URLs, so this changes an inbound shape that only external callers can be using. A caller still requesting the /ws form gets a 404 and needs the new segment. --- .../src/backends/container/cloudflare-container.ts | 6 +++--- packages/computer/src/proxy.ts | 12 +++++------- packages/computer/tests/proxy.test.ts | 2 +- 3 files changed, 9 insertions(+), 11 deletions(-) diff --git a/packages/computer/src/backends/container/cloudflare-container.ts b/packages/computer/src/backends/container/cloudflare-container.ts index 06bd72e9..155a7cb1 100644 --- a/packages/computer/src/backends/container/cloudflare-container.ts +++ b/packages/computer/src/backends/container/cloudflare-container.ts @@ -79,7 +79,7 @@ export interface CloudflareContainerBackendOptions { container: () => ContainerHostHolder | Promise; // Identifies the Workspace-owning DO. Fixed for the lifetime of - // the backend: the backend lives inside this DO and the /ws + // the backend: the backend lives inside this DO and the /api // upgrade always lands here. Plain {binding, id} data so it // survives the Workers RPC hop to a cross-DO container host. workspace: WorkspaceRef; @@ -100,7 +100,7 @@ export interface CloudflareContainerBackendOptions { containerEnv?: Record; // Total time the backend waits for: container port to open, - // /connect POST to return, /ws upgrade to arrive. Default 30s. + // /connect POST to return, /api upgrade to arrive. Default 30s. connectTimeoutMs?: number; // Period for the application-level heartbeat — a watermarks() @@ -165,7 +165,7 @@ export class CloudflareContainerBackend implements WorkspaceBackend { readonly #egress: WorkspaceEgressPolicy; readonly #egressToken: string | undefined; - // State for the in-flight /ws upgrade. handleFetch() resolves + // State for the in-flight /api upgrade. handleFetch() resolves // #pendingUpgrade; connect() awaits it. #pendingUpgrade: Promise | undefined; #resolveUpgrade: ((ws: WebSocket) => void) | undefined; diff --git a/packages/computer/src/proxy.ts b/packages/computer/src/proxy.ts index b9222908..6d21d6f3 100644 --- a/packages/computer/src/proxy.ts +++ b/packages/computer/src/proxy.ts @@ -78,7 +78,7 @@ export class WorkspaceProxy extends WorkerEntrypoint { headers: { "x-test-id": freshId() }, }); expect(health.status).toBe(200); - const websocket = await SELF.fetch(`http://proxy.test/__workspace_connect/${token}/ws`, { + const websocket = await SELF.fetch(`http://proxy.test/__workspace_connect/${token}/api`, { headers: { "x-test-id": freshId(), "x-test-binding": "COMPUTERD" }, }); expect(await websocket.text()).toBe(`from-do:${token}`); From 78cd36b4cb2a50d67e57666f193ca52b71da84e0 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 14:20:55 +0000 Subject: [PATCH 14/23] rpc: Refer to the endpoint as /api in comments The wire moved to /api. Two comments still named /ws as the path the composite server is served on and the URL a client dials. --- packages/rpc/src/client.ts | 2 +- packages/rpc/src/server.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/rpc/src/client.ts b/packages/rpc/src/client.ts index e4fef044..300ff479 100644 --- a/packages/rpc/src/client.ts +++ b/packages/rpc/src/client.ts @@ -15,7 +15,7 @@ export interface RPCEvent { } export interface ClientOptions { - // WebSocket URL. Typically ws://container-host:45678/ws. + // WebSocket URL. Typically ws://container-host:45678/api. url: string; // Optional WebSocket constructor. Defaults to the global // WebSocket (node 22+ ships one; older runtimes can pass the diff --git a/packages/rpc/src/server.ts b/packages/rpc/src/server.ts index 3556663a..ebe7c467 100644 --- a/packages/rpc/src/server.ts +++ b/packages/rpc/src/server.ts @@ -323,7 +323,7 @@ export function createShellServer(runner: RunnerLike): ShellRPC { } // Construct the composite WorkspaceRPC. The wire serves this on -// /ws so clients reach `.sync` and `.shell` through one session. +// /api so clients reach `.sync` and `.shell` through one session. export function createWorkspaceServer( db: Database, runner: RunnerLike, From 3ff0c634c91bbf23651978f228ae87cfb27f5375 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 14:39:05 +0000 Subject: [PATCH 15/23] docs: Drop the HTTP batch alternative from the wire description Two places still introduced the wire as a WebSocket "with an HTTP batch alternative" while the paragraphs below them explained why there is no such alternative. Say it once, in the opening, and give the reason: capnweb's batch transport cannot deliver a stream returned from a call, and every read on this interface returns one. --- .agents/skills/capnweb/SKILL.md | 12 +++++++----- docs/08_capnweb_interface.md | 4 ++-- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/.agents/skills/capnweb/SKILL.md b/.agents/skills/capnweb/SKILL.md index e650304b..bc4e8610 100644 --- a/.agents/skills/capnweb/SKILL.md +++ b/.agents/skills/capnweb/SKILL.md @@ -14,7 +14,9 @@ description: | between the Durable Object and `computerd`. It's an object-capability RPC system with promise pipelining, structured-clone-style transfer of stubs, and bidirectional calls. The wire format used here is text -JSON over a long-lived WebSocket, with an HTTP batch alternative. +JSON over a long-lived WebSocket. That is the only carrier: capnweb's +HTTP batch transport cannot deliver a stream returned from a call, +which is what every read on this interface is. ## Where things live @@ -48,10 +50,10 @@ you don't explicitly dispose stubs, you leak resources on the other side of the connection. This matters more here than in many capnweb deployments because the -connection is **long-lived**. HTTP batch sessions auto-dispose -everything when the batch ends, but our WebSocket between the -Durable Object and `computerd` stays up for the lifetime of the workspace. -Every undisposed stub stays alive until that connection drops. +connection is **long-lived**. A short-lived session disposes everything +when it ends; our WebSocket between the Durable Object and `computerd` +stays up for the lifetime of the workspace, so every undisposed stub +stays alive until that connection drops. ## The caller-disposes rule diff --git a/docs/08_capnweb_interface.md b/docs/08_capnweb_interface.md index e51e389a..59763084 100644 --- a/docs/08_capnweb_interface.md +++ b/docs/08_capnweb_interface.md @@ -6,8 +6,8 @@ [capnweb](https://github.com/cloudflare/capnweb) is the RPC framing used between the Durable Object and the in-container `computerd` computerd. -The wire format is text JSON over a single WebSocket (with an HTTP-batch -alternative). The interface served is `WorkspaceRPC`, defined in +The wire format is text JSON over a single WebSocket. The interface +served is `WorkspaceRPC`, defined in `packages/rpc/src/interface.ts` and consumed by both sides. ## Transport From b7af944e13cb08aae4c2ad06e4c45bc54abbb8c5 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 14:57:33 +0000 Subject: [PATCH 16/23] script: Dial the capnweb session on /api in exec-tests The endpoint moved to /api. This script embeds its client in a heredoc rather than a .mjs file, so it was missed when the other harnesses were updated. --- script/exec-tests | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/script/exec-tests b/script/exec-tests index 05217693..aa5a8e39 100755 --- a/script/exec-tests +++ b/script/exec-tests @@ -77,7 +77,7 @@ const assert = (cond, msg) => { } }; -const client = createWorkspaceClient({ url: `ws://localhost:${port}/ws` }); +const client = createWorkspaceClient({ url: `ws://localhost:${port}/api` }); try { // 1. echo + exit code. const h = await client.shell.exec({ source: "echo hello && exit 7" }); From fcb79cfc26800c8326e2e4abce4a9e6e83cfed9b Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 14:57:33 +0000 Subject: [PATCH 17/23] script: Fix the paths and result shape in the FUSE flush check The check wrote through FUSE to /workspace/x.txt and then read /x.txt back from the receiver store, which cannot resolve: the virtual filesystem keys entries by their absolute in-container path, so the mount point is part of the key. Both ends now use one constant. It also logged pullOnce's return as if it were a count. pullOnce resolves to an ApplyResult, so the line printed "[object Object] entries" and the guard comparing it against 0 could never fire. Destructure applied. --- script/computerd-fuse-flush.mjs | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/script/computerd-fuse-flush.mjs b/script/computerd-fuse-flush.mjs index 0f91f579..e5d1e7b4 100755 --- a/script/computerd-fuse-flush.mjs +++ b/script/computerd-fuse-flush.mjs @@ -138,9 +138,12 @@ async function main() { // 2. Write through FUSE. echo > triggers FUSE create, write, // flush (on close), release \u2014 every spill point landed in // the 68407fc fix. + // The VFS keys everything under the mount point, so the path the + // receiver reads back is the absolute in-container path, not /x.txt. + const target = "/workspace/x.txt"; const payload = `from-fuse ${Date.now()}\n`; - await dockerExec(container.cid, "bash", "-c", `printf '%s' '${payload}' > /workspace/x.txt`); - process.stderr.write(` wrote ${payload.trim()} to /workspace/x.txt via FUSE\n`); + await dockerExec(container.cid, "bash", "-c", `printf '%s' '${payload}' > ${target}`); + process.stderr.write(` wrote ${payload.trim()} to ${target} via FUSE\n`); // 3. Pull from the container's WebSocket on the host. Mirrors // what the host DO does after exec returns. @@ -165,16 +168,16 @@ async function main() { initializeSchema(recvDb, () => Date.now()); try { - const applied = await pullOnce(recvDb, client.sync); + const { applied } = await pullOnce(recvDb, client.sync); process.stderr.write(` pullOnce applied ${applied} entries\n`); if (applied === 0) { - throw new Error("pullOnce applied 0 entries; expected at least 1 for /x.txt"); + throw new Error(`pullOnce applied 0 entries; expected at least 1 for ${target}`); } // 4. Read the file back through the receiver-side provider. // The bytes have to match what we wrote through FUSE. const provider = new SQLiteWorkspaceProvider(recvDb, { now: () => Date.now() }); - const back = provider.readFileSync("/x.txt", "utf8"); + const back = provider.readFileSync(target, "utf8"); if (back !== payload) { throw new Error( `byte mismatch:\n wrote: ${JSON.stringify(payload)}\n read: ${JSON.stringify(back)}`, From 5c2393e94061031c37e212684e4627bec556802c Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 15:10:09 +0000 Subject: [PATCH 18/23] script: Run the whole exec-tests suite outside workerd The last section of exec-tests exercises Workspace.runtime through the high-level facade, and it had never run. It imported TestBackend and Workspace from per-file dist paths, which the bundler does not emit and never has, so the script died on a module resolution error after the seven wire-level checks passed. Those symbols live on the package's main entry, which cannot be imported under plain node: it re-exports WorkspaceProxy, extending WorkerEntrypoint from cloudflare:workers, and that specifier only resolves inside workerd. A node --import hook now registers a resolve hook for the specifier and answers with stub classes, which is enough to satisfy `class X extends Y` at module evaluation time. Nothing in the stub is ever constructed. The alternative was adding build entries for each file the script wants, which would duplicate bundled code in a published package to suit a harness. Also correct the soak entry in the script list, which still described the two-container topology. --- AGENTS.md | 9 +++- script/exec-tests | 26 ++++++------ script/lib/cloudflare-workers-stub.mjs | 59 ++++++++++++++++++++++++++ 3 files changed, 79 insertions(+), 15 deletions(-) create mode 100644 script/lib/cloudflare-workers-stub.mjs diff --git a/AGENTS.md b/AGENTS.md index c886cc99..7469bf19 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -133,8 +133,9 @@ loop. Reach for these when you're chasing a behavior the unit tests don't cover. - `shell` boots a debian-slim container with the linux `computerd` binary mounted under `/usr/local/bin`. The starting point for anything that needs a real FUSE mount. -- `computerd-soak.mjs` boots two `computerd` containers wired peer-to-peer and - soaks the sync loop. Use it to chase convergence or churn bugs. +- `computerd-soak.mjs` boots one `computerd` container, takes the host's + part of the sync loop from a local store, and reports convergence lag + beside container memory. Use it to chase convergence or churn bugs. - `computerd-stub-soak.mjs` soaks the long-lived WebSocket session and reads `session.getStats()` to detect stub-disposal drift. Run it for changes around the capnweb lifecycle. @@ -147,6 +148,10 @@ loop. Reach for these when you're chasing a behavior the unit tests don't cover. tasks against the mount with a tmpfs baseline for comparison. - `exec-tests` boots `computerd` in docker with FUSE disabled and exercises a few `shell.exec` scenarios. +- `lib/cloudflare-workers-stub.mjs` is a node `--import` hook that stubs + the `cloudflare:workers` module, so a host-side script can import + `@cloudflare/computer`'s main entry outside workerd. Not runnable on + its own; `exec-tests` uses it. - `npm-bench.sh` / `run-npm-bench.sh` benchmark npm package installs on native disk vs the FUSE mount. `run-npm-bench.sh` is the user-facing entry point that boots a privileged Docker container; diff --git a/script/exec-tests b/script/exec-tests index aa5a8e39..a9d77d7e 100755 --- a/script/exec-tests +++ b/script/exec-tests @@ -13,7 +13,7 @@ # - artifacts/computerd/computerd-linux-x64 (run `npm run build:bin --workspace # @cloudflare/computerd` first) # - docker -# - node 22+ on the host +# - node on the host, recent enough for module.registerHooks (22.15+) set -euo pipefail repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" @@ -61,9 +61,12 @@ if ! curl -sf "http://localhost:${port}/health" >/dev/null 2>&1; then fi echo "==> computerd healthy on :${port}" -# Drive the smoke from a node script. Inline so we don't add a -# new file to mount; the script is self-contained. -PORT="${port}" REPO_ROOT="${repo_root}" node --input-type=module -e ' +# Drive the smoke from a node script, inline so nothing extra has to be +# mounted into the container. The --import hook makes the package's main +# entry loadable outside workerd; see the note at its use below. +PORT="${port}" REPO_ROOT="${repo_root}" node \ + --import "${repo_root}/script/lib/cloudflare-workers-stub.mjs" \ + --input-type=module -e ' const repoRoot = process.env.REPO_ROOT; const port = process.env.PORT; const { createWorkspaceClient } = await import( @@ -234,15 +237,12 @@ try { // 7. Workspace.runtime host wrapper (E3). Uses the composite // client through the high-level facade: encoding utf8 + // ExecHandle.result() in one call. - // Import the specific files rather than the package barrel. - // The barrel re-exports WorkspaceProxy, which extends - // WorkerEntrypoint from cloudflare:workers — that scheme - // only resolves under workerd, not plain Node. - const { TestBackend } = await import( - `${repoRoot}/packages/computer/dist/backends/test.js` - ); - const { Workspace } = await import( - `${repoRoot}/packages/computer/dist/workspace.js` + // The package barrel re-exports WorkspaceProxy, which extends + // WorkerEntrypoint from cloudflare:workers. That specifier only + // resolves under workerd, so this runs with a --import hook that + // stubs it; see script/lib/cloudflare-workers-stub.mjs. + const { TestBackend, Workspace } = await import( + `${repoRoot}/packages/computer/dist/index.js` ); // Workspace needs DurableObjectStorageLike; supply the // node:sqlite-backed test storage so this runs under plain diff --git a/script/lib/cloudflare-workers-stub.mjs b/script/lib/cloudflare-workers-stub.mjs new file mode 100644 index 00000000..4ab7f63d --- /dev/null +++ b/script/lib/cloudflare-workers-stub.mjs @@ -0,0 +1,59 @@ +// Makes `@cloudflare/computer`'s main entry importable under plain node. +// +// The entry re-exports WorkspaceProxy, which extends WorkerEntrypoint +// from `cloudflare:workers`. That specifier only resolves inside +// workerd, so a plain `import` of the entry fails with +// ERR_UNSUPPORTED_ESM_URL_SCHEME before any of the exports a host-side +// script actually wants become reachable. +// +// Registering a resolve hook for the specifier costs less than the +// alternatives. Importing dist files individually needs per-file build +// entries the bundler does not emit, and adding them would duplicate +// bundled code to suit a script. +// +// The stubs only have to satisfy `class X extends Y` at module +// evaluation time. Nothing here is ever constructed: a script that +// reaches for Workers runtime behavior wants workerd, not this. +// +// Use it with node's --import flag: +// +// node --import ./script/lib/cloudflare-workers-stub.mjs script.mjs +import { registerHooks } from "node:module"; + +const SPECIFIER = "cloudflare:workers"; +const STUB_URL = "cloudflare-workers-stub:main"; + +const SOURCE = ` +export class RpcTarget {} + +class Entrypoint { + constructor(ctx, env) { + this.ctx = ctx; + this.env = env; + } +} + +export class WorkerEntrypoint extends Entrypoint {} +export class DurableObject extends Entrypoint {} + +export const tracing = { + enterSpan(_name, callback) { + return callback(); + }, +}; +`; + +registerHooks({ + resolve(specifier, context, next) { + if (specifier === SPECIFIER) { + return { url: STUB_URL, shortCircuit: true }; + } + return next(specifier, context); + }, + load(url, context, next) { + if (url === STUB_URL) { + return { format: "module", source: SOURCE, shortCircuit: true }; + } + return next(url, context); + }, +}); From c20255215a8635a2ae0a1bc2864999c44cd4dd9e Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 15:34:38 +0000 Subject: [PATCH 19/23] chore: Reword the changeset for the endpoint move --- .changeset/one-endpoint-on-api.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/one-endpoint-on-api.md b/.changeset/one-endpoint-on-api.md index 81bf03cb..003eb51a 100644 --- a/.changeset/one-endpoint-on-api.md +++ b/.changeset/one-endpoint-on-api.md @@ -2,4 +2,4 @@ "@cloudflare/computer": minor --- -The capnweb session now runs on `/api` at both ends of the connection, replacing `/ws`. A durable object that routes the container's outbound upgrade must match `/api` in its `fetch()` handler; `WorkspaceProxy` already does. `POST /connect` carries the endpoint as `{ base, health, api }` so the container assembles no host paths of its own, and a request to `/api` without an `Upgrade` header answers `400` rather than upgrading. +Refactor Cap'n Web endpoints. The /ws endpoint has been renamed to /api. The /connect caller now provides both API & healthcheck endpoints. Support for Cap'n Web HTTP batch transport has been removed. From 2132e8702fdff1df7173902a27e741eaae57faba Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 15:34:38 +0000 Subject: [PATCH 20/23] docs: Drop the missing-HTTP-transport note from the carrier The carrier entry stated that no HTTP alternative exists and then explained why. A contract document is better describing what the wire is than cataloguing what it is not, and the claim reads as though it were about capnweb, which does ship an HTTP batch transport. --- docs/08_capnweb_interface.md | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/docs/08_capnweb_interface.md b/docs/08_capnweb_interface.md index 59763084..d708b0d4 100644 --- a/docs/08_capnweb_interface.md +++ b/docs/08_capnweb_interface.md @@ -13,10 +13,7 @@ served is `WorkspaceRPC`, defined in ## Transport - **Carrier.** One long-lived WebSocket per Workspace, on the - computerd's `/api` endpoint. There is no HTTP alternative: every call - that returns data returns a `ReadableStream`, and delivering one - needs a call from server to client, which capnweb's HTTP batch - transport cannot complete. Default port is `45678`; it will become a + computerd's `/api` endpoint. Default port is `45678`; it will become a build-time variable so hosts can pin a non-default port. - **Framing.** capnweb text frames. Binary frames are unsupported. **(planned)** the server will fail the session loudly on the first From 712f04101344f9d26e289839fc60f4637f57a71d Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 16:28:40 +0000 Subject: [PATCH 21/23] script: Pair each soak file with its own payload The write loop passed writeSeq++ for the file index and writeSeq for the payload seed. Arguments evaluate left to right, so the increment landed before the seed was read: every file carried the bytes generated for the next index, and seed zero was never used. Nothing in the script asserts on content, so the soak still measured what it was built to measure. The pairing is wrong for anyone who later checks the bytes against the name. --- script/computerd-soak.mjs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/script/computerd-soak.mjs b/script/computerd-soak.mjs index 5f913339..d6472e05 100755 --- a/script/computerd-soak.mjs +++ b/script/computerd-soak.mjs @@ -266,7 +266,8 @@ async function main() { // owns the truth and pushes it to the container. const writeLoop = (async () => { while (Date.now() < stopAt) { - writeOne(provider, writeSeq++, payloadBytes(writeSeq)); + const seq = writeSeq++; + writeOne(provider, seq, payloadBytes(seq)); writesSent++; await new Promise((r) => setTimeout(r, intervalMs)); } From ae09ee0b5bee17379ac482193c5021898701f413 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 17 Aug 2026 16:28:58 +0000 Subject: [PATCH 22/23] chore: Split the changeset into one entry per change The endpoint rename, the /connect payload change, and the removal of UPSTREAM_URL landed as a single entry. They are three separate things a consumer may care about, and a reader scanning the changelog for the one that affects them should not have to unpick a paragraph covering all three. --- .changeset/connect-names-host-paths.md | 5 +++++ .changeset/one-endpoint-on-api.md | 5 ----- .changeset/remove-upstream-url.md | 5 +++++ .changeset/rename-ws-endpoint-to-api.md | 5 +++++ 4 files changed, 15 insertions(+), 5 deletions(-) create mode 100644 .changeset/connect-names-host-paths.md delete mode 100644 .changeset/one-endpoint-on-api.md create mode 100644 .changeset/remove-upstream-url.md create mode 100644 .changeset/rename-ws-endpoint-to-api.md diff --git a/.changeset/connect-names-host-paths.md b/.changeset/connect-names-host-paths.md new file mode 100644 index 00000000..b1cad5d0 --- /dev/null +++ b/.changeset/connect-names-host-paths.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": minor +--- + +The /connect caller now provides both API & healthcheck endpoints. The container no longer builds either path itself, so a host is free to serve them wherever it likes. diff --git a/.changeset/one-endpoint-on-api.md b/.changeset/one-endpoint-on-api.md deleted file mode 100644 index 003eb51a..00000000 --- a/.changeset/one-endpoint-on-api.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -"@cloudflare/computer": minor ---- - -Refactor Cap'n Web endpoints. The /ws endpoint has been renamed to /api. The /connect caller now provides both API & healthcheck endpoints. Support for Cap'n Web HTTP batch transport has been removed. diff --git a/.changeset/remove-upstream-url.md b/.changeset/remove-upstream-url.md new file mode 100644 index 00000000..a1ffec68 --- /dev/null +++ b/.changeset/remove-upstream-url.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": minor +--- + +The UPSTREAM_URL environment variable has been removed along with the container's own sync loop. Syncing is driven by whichever peer holds the other end of the Cap'n Web session. diff --git a/.changeset/rename-ws-endpoint-to-api.md b/.changeset/rename-ws-endpoint-to-api.md new file mode 100644 index 00000000..d21fef3f --- /dev/null +++ b/.changeset/rename-ws-endpoint-to-api.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": minor +--- + +The Cap'n Web /ws endpoint has been renamed to /api at both ends of the connection. A durable object that routes the container's outbound upgrade must match /api in its own fetch handler. Support for the Cap'n Web HTTP batch transport has been removed, so /api carries a websocket only. From 16eacf84328bc61e0d67a989d69223cc21dde3a4 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Tue, 18 Aug 2026 10:36:37 +0000 Subject: [PATCH 23/23] computerd: Move the watermarks route under /api The route read sync revisions through rpc.sync.watermarks() but sat at /__computerd/watermarks, alongside runtime info and process memory. Those report on the daemon; this reports the workspace, and it returns exactly what a session would. It is now /api/watermarks, which draws the line where it belongs: /api is the workspace surface, including anything that reads through it, and /__computerd is daemon introspection. Only the exact /api path upgrades, so a handshake aimed at the subpath still answers 404, and the shared secret covers the new path the same way it covered the old one. --- docs/07_injected_service.md | 8 ++- packages/computerd/README.md | 2 +- packages/computerd/src/cli/computerd.test.ts | 16 +++++- packages/computerd/src/cli/computerd.ts | 55 +++++++++++--------- script/computerd-soak.mjs | 2 +- 5 files changed, 51 insertions(+), 32 deletions(-) diff --git a/docs/07_injected_service.md b/docs/07_injected_service.md index 93fa95f5..e3b8b58f 100644 --- a/docs/07_injected_service.md +++ b/docs/07_injected_service.md @@ -51,11 +51,15 @@ backend pins it to `8080`) and serves: | --- | --- | --- | | `/health` | `GET`, `HEAD` | Liveness probe; `200 ok\n` as soon as the HTTP server binds. | | `/__computerd/info` | `GET` | Runtime info: FUSE backend, mount point, port. | -| `/__computerd/watermarks` | `GET` | Sync revisions: `currentRev`, `pushRev`, `fetchCursor`. | -| `/api` | `GET` (upgrade) | WebSocket capnweb transport — the bootstrap stub is `WorkspaceRPC`. A request without an `Upgrade` header gets `400`; an unsupported `Sec-WebSocket-Version` gets `426` and the versions the server speaks. | +| `/api` | `GET` (upgrade) | WebSocket capnweb transport — the bootstrap stub is `WorkspaceRPC`. Only the exact path upgrades. A request without an `Upgrade` header gets `400`; an unsupported `Sec-WebSocket-Version` gets `426` and the versions the server speaks. | +| `/api/watermarks` | `GET`, `HEAD` | Sync revisions: `currentRev`, `pushRev`, `fetchCursor`. The same values `sync.watermarks()` returns, for callers that want a few numbers without holding a session. | | `/connect` | `POST` | Tells `computerd` to dial *out* to a caller-supplied endpoint and serve a `WorkspaceRPC` session over that outbound WebSocket. Used by the Cloudflare backend (see below). | | `/` | `GET` | Banner/info page. | +`/api` is the workspace surface: the session itself, plus anything that +reads through it. `/__computerd` is daemon introspection, which is why +runtime info sits there and revisions do not. + The capnweb bootstrap interface is **`WorkspaceRPC`** (defined in `packages/rpc/`), split into `sync` and `shell` sub-stubs. diff --git a/packages/computerd/README.md b/packages/computerd/README.md index 138a9ac1..7ba1088e 100644 --- a/packages/computerd/README.md +++ b/packages/computerd/README.md @@ -28,8 +28,8 @@ Current endpoints: - `GET /__computerd/info` returns JSON with the selected FUSE backend, mount point, and bound port. - `GET /__computerd/stats` returns JSON with DOFS table row counts, total inline and blob byte sizes, the orphan-blob subset, and process resident memory. Useful for watching how the store grows under load. - `GET /` returns `200 OK` with an empty JSON object: `{}`. -- `GET /__computerd/watermarks` returns JSON with `currentRev`, `pushRev`, and `fetchCursor`, read through the same `watermarks()` the wire serves. For samplers that want a few numbers without opening a session. - `GET /api` upgrades to a WebSocket carrying the capnweb RPC surface backed by `@cloudflare/computer-rpc`. This is the container's only RPC carrier. A request without an `Upgrade` header returns `400`; a handshake naming an unsupported `Sec-WebSocket-Version` returns `426` along with the versions the server speaks. +- `GET /api/watermarks` returns JSON with `currentRev`, `pushRev`, and `fetchCursor`, read through the same `watermarks()` the wire serves. For samplers that want a few numbers without opening a session. It sits under `/api` because it reads the workspace surface; `/__computerd` is for daemon introspection. All other paths and methods return `404`/`405` with a `text/plain` body. diff --git a/packages/computerd/src/cli/computerd.test.ts b/packages/computerd/src/cli/computerd.test.ts index a28c29ae..9c524566 100644 --- a/packages/computerd/src/cli/computerd.test.ts +++ b/packages/computerd/src/cli/computerd.test.ts @@ -173,6 +173,18 @@ test("/api refuses anything that is not a websocket handshake", async (_ctx) => ]); expect(noKey).toMatch(/^HTTP\/1\.1 400 /); + // A subpath under /api is not the session endpoint: only the exact + // path upgrades. + const subpath = await rawRequest(port, [ + "GET /api/watermarks HTTP/1.1", + `Host: 127.0.0.1:${port}`, + "Upgrade: websocket", + "Connection: Upgrade", + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==", + "Sec-WebSocket-Version: 13", + ]); + expect(subpath).toMatch(/^HTTP\/1\.1 404 /); + // An unknown path is still a 404, upgrade header or not. const unknown = await rawRequest(port, [ "GET /nope HTTP/1.1", @@ -185,14 +197,14 @@ test("/api refuses anything that is not a websocket handshake", async (_ctx) => expect(unknown).toMatch(/^HTTP\/1\.1 404 /); }); -test("/__computerd/watermarks reports sync revisions over plain HTTP", async (_ctx) => { +test("/api/watermarks reports sync revisions over plain HTTP", async (_ctx) => { // Samplers want three numbers on an interval. Opening an RPC session // per sample is the wrong shape for that. const port = await getAvailablePort(); const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-watermarks-")); await startComputerd({ port, mountPoint, env: { FUSE_MOUNT: "none" } }); - const res = await fetch(`http://127.0.0.1:${port}/__computerd/watermarks`); + const res = await fetch(`http://127.0.0.1:${port}/api/watermarks`); expect(res.status).toBe(200); const body = await res.json(); expect(body).toMatchObject({ diff --git a/packages/computerd/src/cli/computerd.ts b/packages/computerd/src/cli/computerd.ts index dd26fbc4..a0e42e95 100644 --- a/packages/computerd/src/cli/computerd.ts +++ b/packages/computerd/src/cli/computerd.ts @@ -184,6 +184,35 @@ function createHTTPServer( return; } + // /api/watermarks — the same sync revisions the session serves, + // over plain HTTP, for samplers that want a few numbers on an + // interval rather than a session of their own. It reads through + // rpc.sync.watermarks(), so the two cannot drift. Part of the + // workspace API rather than daemon introspection, hence /api + // rather than /__computerd. + if (path === "/api/watermarks") { + if (request.method === "HEAD") { + send(response, 200, "", { "content-type": "application/json; charset=utf-8" }); + return; + } + void rpc.sync + .watermarks() + .then((watermarks) => { + send(response, 200, JSON.stringify(watermarks), { + "content-type": "application/json; charset=utf-8", + }); + }) + .catch((error: unknown) => { + console.error("/api/watermarks failed:", error); + if (!response.headersSent) { + send(response, 500, "internal error\n", { + "content-type": "text/plain; charset=utf-8", + }); + } + }); + return; + } + // /connect — POST { base, health, api } naming an endpoint the // host wants us to dial back into. We poll `base + health` until // it answers, then open a capnweb WebSocket session against @@ -240,32 +269,6 @@ function createHTTPServer( return; } - // Sync revisions over plain HTTP, for samplers that want a few - // numbers on an interval rather than an RPC session. Reads through - // the same watermarks() the wire serves, so the two cannot drift. - if (path === "/__computerd/watermarks") { - if (request.method === "HEAD") { - send(response, 200, "", { "content-type": "application/json; charset=utf-8" }); - return; - } - void rpc.sync - .watermarks() - .then((watermarks) => { - send(response, 200, JSON.stringify(watermarks), { - "content-type": "application/json; charset=utf-8", - }); - }) - .catch((error: unknown) => { - console.error("/__computerd/watermarks failed:", error); - if (!response.headersSent) { - send(response, 500, "internal error\n", { - "content-type": "text/plain; charset=utf-8", - }); - } - }); - return; - } - if (path === "/__computerd/info") { const body = request.method === "HEAD" ? "" : JSON.stringify(info); send(response, 200, body, { diff --git a/script/computerd-soak.mjs b/script/computerd-soak.mjs index d6472e05..bf30822d 100755 --- a/script/computerd-soak.mjs +++ b/script/computerd-soak.mjs @@ -183,7 +183,7 @@ async function dockerStats(cids) { // The daemon's revision numbers over plain HTTP. A sampler wanting // three integers on an interval does not need an RPC session. async function fetchWatermarks(url) { - const res = await fetch(`${url}/__computerd/watermarks`); + const res = await fetch(`${url}/api/watermarks`); if (!res.ok) throw new Error(`watermarks HTTP ${res.status}`); return await res.json(); }