diff --git a/docs/08_capnweb_interface.md b/docs/08_capnweb_interface.md index 8b9d4c83..6ec45c70 100644 --- a/docs/08_capnweb_interface.md +++ b/docs/08_capnweb_interface.md @@ -162,15 +162,19 @@ See SUMMARY §0 'Investigation notes' (commits dc692c0, c95c74d, ```ts interface ShellRPC { - // Spawn a command. Returns a handle whose `events` stream - // yields stdout / stderr / exit frames. The stream is the - // single source of truth — there is no buffered-return - // variant. The handle's id can be passed to getExec to - // reattach after a reconnect. + // Spawn an execution. `source` is a shell command line for a + // command backend, or module source for a callable backend. + // Returns a handle whose `events` stream yields stdout / stderr / + // exit frames. The stream is the single source of truth — there + // is no buffered-return variant. The handle's id can be passed to + // getExec to reattach after a reconnect. `input` carries a + // structured value for a callable backend; command backends + // ignore it. exec(input: { - command: string; - cwd?: string; - id?: string; + source: string; + cwd?: string; + id?: string; + input?: unknown; }): Promise<{ id: string; events: ReadableStream }>; // Reattach to an in-flight or recently-completed exec by id. @@ -195,9 +199,15 @@ interface ShellRPC { type ExecEvent = | { id: string; seq: number; name: "stdout"; value: Uint8Array } | { id: string; seq: number; name: "stderr"; value: Uint8Array } - | { id: string; seq: number; name: "exit"; value: number }; + | { id: string; seq: number; name: "exit"; value: number; result?: unknown }; ``` +The `exit` frame carries the process exit code and, for a callable +backend that ran to a zero exit, the structured return value on +`result`. The value and the exit code settle together, so a single +terminal frame carries both rather than splitting them across two +frames. Command backends never set `result`. + All payloads on the wire are binary. The host-side `Workspace.runtime` converts to `string` when the caller passes `encoding: "utf8"`. Every event carries a monotonic `seq` (per exec id) so callers can resume diff --git a/docs/11_lifecycle.md b/docs/11_lifecycle.md index cfa7bf37..6bf7952b 100644 --- a/docs/11_lifecycle.md +++ b/docs/11_lifecycle.md @@ -340,7 +340,7 @@ Two things have to change for capnweb + hibernation to work: is dropped by the idempotent apply path. No attachment write is required. - **Exec streams: store `{ [id]: seq }` per in-flight exec.** - The `WorkspaceShell` driver inside the DO is the only place + The `CommandExecutor` driver inside the DO is the only place that knows where the consumer got to in the event stream. Every time it surfaces an event to the caller (or on some reasonable debounce) it has to update the attachment so the diff --git a/docs/12_worker_backend.md b/docs/12_worker_backend.md index d258349c..a207794a 100644 --- a/docs/12_worker_backend.md +++ b/docs/12_worker_backend.md @@ -138,9 +138,9 @@ the same. `BackendHandle.sync` is `"none"`. With a single authoritative store there's nothing to ship or fetch; `Workspace.push` and `Workspace.pull` short-circuit on the bit and the reconcile pass -on connect is skipped. The shell exec bracket still calls them so -the surface stays uniform — every `ExecResult.pushed`, `pulled`, -and `skipped` is empty. +on connect is skipped. The exec sync bracket still calls them so +the surface stays uniform — the pushed, pulled, and skipped counts +on the runtime result are empty. ## Event stream framing @@ -152,9 +152,9 @@ shape across the isolate hop. The backend decodes the frames into structured `ExecEvent` values and re-encodes string payloads (`stdout` / `stderr`) into -`Uint8Array` so the existing `WorkspaceShell` utf8 decoder -transforms in `packages/computer/src/shell.ts` see the shape -they already handle. +`Uint8Array` so the runtime's utf8 decoder transforms, which +accumulate the result from raw events, see the shape they already +handle. ## Fetcher factory escape hatch diff --git a/docs/17_isolate_javascript.md b/docs/17_isolate_javascript.md index de3396c1..2db84ecf 100644 --- a/docs/17_isolate_javascript.md +++ b/docs/17_isolate_javascript.md @@ -8,7 +8,6 @@ import { WorkerJavaScriptBackend } from "@cloudflare/computer/backends/worker-ja const workspace = new Workspace({ storage: ctx.storage, - waitUntil: ctx.waitUntil.bind(ctx), backends: [ new WorkerJavaScriptBackend({ loader: env.LOADER, @@ -52,7 +51,7 @@ const result = await handle.result(); The source is a real ES module. Static imports, literal dynamic imports, and top-level await are supported. If the module default-exports a function, Workspace invokes it with `options.input`. Otherwise module evaluation completes with a `null` structured result. -`waitUntil` is required for this backend. `runtime.exec()` returns before the Dynamic Worker finishes, so the host must attach completion to the Durable Object event lifetime. Construction fails when a module backend connects without this hook. +`runtime.exec()` returns before the Dynamic Worker finishes: the run keeps advancing while its event stream is consumed and the host call into the Dynamic Worker stays in flight. That pending work keeps the Durable Object resident on its own. A run whose handle is returned but never read can be evicted once the object goes idle; drain the event stream (or `result()`) to keep the run alive, and schedule an alarm through `ctx.storage.setAlarm()` for work that must survive eviction. ## Durable relative imports diff --git a/docs/18_runtime_migration.md b/docs/18_runtime_migration.md index 541a051a..db68c8b1 100644 --- a/docs/18_runtime_migration.md +++ b/docs/18_runtime_migration.md @@ -12,7 +12,7 @@ This change is a breaking preview-API migration. Public execution now uses one r | `workspace.shell.dispose(id, options)` | `workspace.runtime.disposeExec(id, options)` | | `workspace.code` / script execution | `workspace.runtime.exec(source, { backend: "worker-javascript", input })` | -`WorkspaceShell` still exists internally to implement command backends. It is not a public Workspace property. +`CommandExecutor` exists internally to implement command backends. It is not a public Workspace property. ## Default backend IDs diff --git a/examples/worker-javascript/README.md b/examples/worker-javascript/README.md index 98221f66..473e25dd 100644 --- a/examples/worker-javascript/README.md +++ b/examples/worker-javascript/README.md @@ -50,9 +50,10 @@ client ─► Worker /c//{file,exec} store (the DO's SQLite); push and pull short-circuit. `pushed` / `pulled` are always zero. -The backend requires `waitUntil`, so the DO passes -`ctx.waitUntil.bind(ctx)` into the Workspace options. The DO is a -thin host; the Dynamic Worker lifecycle is the loader's problem. +The DO is a thin host; the Dynamic Worker lifecycle is the loader's +problem. A run keeps advancing while its event stream is consumed, so +the request that drains the handle holds the object resident until the +run finishes. ## Paths diff --git a/examples/worker-javascript/src/index.ts b/examples/worker-javascript/src/index.ts index 3aa5758f..5cd06032 100644 --- a/examples/worker-javascript/src/index.ts +++ b/examples/worker-javascript/src/index.ts @@ -13,7 +13,6 @@ export class ContainerExample extends withWorkspace(class extends DurableObject< const { ctx, env } = self as unknown as { ctx: DurableObjectState; env: Env }; return { storage: ctx.storage as unknown as DurableObjectStorageLike, - waitUntil: ctx.waitUntil.bind(ctx), backends: [new WorkerJavaScriptBackend({ loader: env.LOADER })], mounts: { "/workspace/r2": R2Bucket(env.Bucket), diff --git a/examples/worker-shell/README.md b/examples/worker-shell/README.md index ee1e0fb9..76231733 100644 --- a/examples/worker-shell/README.md +++ b/examples/worker-shell/README.md @@ -57,8 +57,8 @@ client ─► Worker /c//{file,exec} and one workspace per DO is the natural boundary. 5. `BackendHandle.sync` is `"none"`. There's a single authoritative store (the DO's SQLite); push and pull - short-circuit. `ExecResult.pushed` / `pulled` are always - zero. + short-circuit. The runtime result's `pushed` / `pulled` + counts are always zero. The DO is a thin host. There's no Dockerfile; the Dynamic Worker lifecycle is the loader's problem. diff --git a/packages/computer/README.md b/packages/computer/README.md index 1075d5f7..1f8a2c47 100644 --- a/packages/computer/README.md +++ b/packages/computer/README.md @@ -242,9 +242,8 @@ Alongside `exec`, the runtime exposes `getExec`, `killExec`, and - **Worker JavaScript** evaluates a module with structured input/results, durable relative imports, configured libraries, Workspace-backed `node:fs/promises`, and trusted `ws:git` / - `ws:artifacts` modules. It runs after `runtime.exec()` returns, so - pass `waitUntil: ctx.waitUntil.bind(ctx)` to `Workspace`; the backend - refuses to connect without it. See + `ws:artifacts` modules. It runs after `runtime.exec()` returns; the + run stays alive while its event stream is consumed. See [`docs/17_isolate_javascript.md`](../../docs/17_isolate_javascript.md) and [`examples/worker-javascript`](../../examples/worker-javascript). diff --git a/packages/computer/src/backend.ts b/packages/computer/src/backend.ts index ca9b517a..87482600 100644 --- a/packages/computer/src/backend.ts +++ b/packages/computer/src/backend.ts @@ -30,7 +30,6 @@ import type { WorkspaceRPC } from "@cloudflare/computer-rpc"; // their own transport); the in-process module backend uses it. export interface WorkspaceBackendHost { readonly db: import("@cloudflare/dofs").Database; - readonly waitUntil?: (promise: Promise) => void; readonly fs: import("@cloudflare/dofs").WorkspaceFilesystem; readonly git: import("./git/index.js").GitClient; readonly artifacts: import("./artifacts/index.js").ArtifactClient; diff --git a/packages/computer/src/backends/worker-javascript/frames.test.ts b/packages/computer/src/backends/worker-javascript/frames.test.ts index c065f481..fc529825 100644 --- a/packages/computer/src/backends/worker-javascript/frames.test.ts +++ b/packages/computer/src/backends/worker-javascript/frames.test.ts @@ -36,14 +36,19 @@ describe("parseRuntimeFrame", () => { expect(frame).toEqual({ name: "stderr", value: new TextEncoder().encode("oops") }); }); - it("decodes a result frame carrying a structured value", () => { - const frame = parseRuntimeFrame(`{"name":"result","value":{"a":[1,2,null]}}`); - expect(frame).toEqual({ name: "result", value: { a: [1, 2, null] } }); + it("decodes an exit frame carrying an integer", () => { + expect(parseRuntimeFrame(`{"name":"exit","code":0}`)).toEqual({ name: "exit", code: 0 }); + expect(parseRuntimeFrame(`{"name":"exit","code":130}`)).toEqual({ name: "exit", code: 130 }); }); - it("decodes an exit frame carrying an integer", () => { - expect(parseRuntimeFrame(`{"name":"exit","value":0}`)).toEqual({ name: "exit", value: 0 }); - expect(parseRuntimeFrame(`{"name":"exit","value":130}`)).toEqual({ name: "exit", value: 130 }); + it("decodes an exit frame carrying a structured result", () => { + const frame = parseRuntimeFrame(`{"name":"exit","code":0,"result":{"a":[1,2,null]}}`); + expect(frame).toEqual({ name: "exit", code: 0, result: { a: [1, 2, null] } }); + }); + + it("decodes an exit frame carrying a null result", () => { + const frame = parseRuntimeFrame(`{"name":"exit","code":0,"result":null}`); + expect(frame).toEqual({ name: "exit", code: 0, result: null }); }); it("rejects invalid JSON", () => { @@ -59,7 +64,7 @@ describe("parseRuntimeFrame", () => { }); it("rejects an exit frame whose value is not an integer", () => { - expect(() => parseRuntimeFrame(`{"name":"exit","value":"x"}`)).toThrow(); + expect(() => parseRuntimeFrame(`{"name":"exit","code":"x"}`)).toThrow(); }); }); @@ -67,12 +72,12 @@ describe("decodeRuntimeFrames", () => { it("decodes newline-delimited frames arriving in one chunk", async () => { const frames = await collect( decodeRuntimeFrames( - streamOf(`{"name":"stdout","b64":"${b64("hi")}"}\n{"name":"exit","value":0}\n`), + streamOf(`{"name":"stdout","b64":"${b64("hi")}"}\n{"name":"exit","code":0}\n`), ), ); expect(frames).toEqual([ { name: "stdout", value: new TextEncoder().encode("hi") }, - { name: "exit", value: 0 }, + { name: "exit", code: 0 }, ]); }); @@ -86,13 +91,13 @@ describe("decodeRuntimeFrames", () => { }); it("emits a trailing frame that arrives without a final newline", async () => { - const frames = await collect(decodeRuntimeFrames(streamOf(`{"name":"exit","value":1}`))); - expect(frames).toEqual([{ name: "exit", value: 1 }]); + const frames = await collect(decodeRuntimeFrames(streamOf(`{"name":"exit","code":1}`))); + expect(frames).toEqual([{ name: "exit", code: 1 }]); }); it("skips blank lines between frames", async () => { - const frames = await collect(decodeRuntimeFrames(streamOf(`{"name":"exit","value":0}\n\n`))); - expect(frames).toEqual([{ name: "exit", value: 0 }]); + const frames = await collect(decodeRuntimeFrames(streamOf(`{"name":"exit","code":0}\n\n`))); + expect(frames).toEqual([{ name: "exit", code: 0 }]); }); it("errors the stream on a malformed frame", async () => { diff --git a/packages/computer/src/backends/worker-javascript/frames.ts b/packages/computer/src/backends/worker-javascript/frames.ts index d924a6f5..6b738238 100644 --- a/packages/computer/src/backends/worker-javascript/frames.ts +++ b/packages/computer/src/backends/worker-javascript/frames.ts @@ -3,8 +3,7 @@ import type { WorkspaceRuntimeValue } from "../../runtime/types.js"; export type RuntimeFrame = | { name: "stdout"; value: Uint8Array } | { name: "stderr"; value: Uint8Array } - | { name: "result"; value: WorkspaceRuntimeValue } - | { name: "exit"; value: number }; + | { name: "exit"; code: number; result?: WorkspaceRuntimeValue }; export function parseRuntimeFrame(line: string): RuntimeFrame { let record: Record; @@ -20,14 +19,18 @@ export function parseRuntimeFrame(line: string): RuntimeFrame { } return { name, value: decodeBase64(record.b64) }; } - if (name === "result") { - return { name, value: record.value as WorkspaceRuntimeValue }; - } if (name === "exit") { - if (!Number.isSafeInteger(record.value)) { + if (!Number.isSafeInteger(record.code)) { throw new Error("WorkerJavaScriptBackend received a malformed exit frame"); } - return { name, value: record.value as number }; + if ("result" in record) { + return { + name, + code: record.code as number, + result: record.result as WorkspaceRuntimeValue, + }; + } + return { name, code: record.code as number }; } throw new Error("WorkerJavaScriptBackend received an unknown execution frame"); } diff --git a/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts b/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts index 686070be..0439478e 100644 --- a/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts +++ b/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts @@ -3,15 +3,7 @@ import { SQLiteTestStorage } from "@cloudflare/dofs/testing"; import { describe, expect, it, vi } from "vitest"; import { Workspace } from "../../workspace.js"; -import { WorkerJavaScriptBackend as ProductionWorkerJavaScriptBackend } from "./worker-javascript.js"; - -class WorkerJavaScriptBackend extends ProductionWorkerJavaScriptBackend { - override readonly requiresWaitUntil = false; - - override connect(host: Parameters[0]) { - return super.connect({ ...host, waitUntil: host.waitUntil ?? (() => {}) }); - } -} +import { WorkerJavaScriptBackend } from "./worker-javascript.js"; function throwingLoader(message: string) { return { @@ -34,12 +26,11 @@ async function evaluateResult( const frames: string[] = []; try { await host.assertResult(value); - frames.push(JSON.stringify({ name: "result", value })); - frames.push(JSON.stringify({ name: "exit", value: 0 })); + frames.push(JSON.stringify({ name: "exit", code: 0, result: value })); } catch (error) { const message = error instanceof Error ? error.message : String(error); frames.push(JSON.stringify({ name: "stderr", b64: btoa(`${message}\n`) })); - frames.push(JSON.stringify({ name: "exit", value: 1 })); + frames.push(JSON.stringify({ name: "exit", code: 1 })); } const encoder = new TextEncoder(); const readable = new ReadableStream({ @@ -52,18 +43,6 @@ async function evaluateResult( } describe("WorkerJavaScriptBackend", () => { - it("requires a host event-lifetime hook", async () => { - const backend = new ProductionWorkerJavaScriptBackend({ loader: throwingLoader("unused") }); - await expect( - backend.connect({ - db: undefined as never, - fs: undefined as never, - git: undefined as never, - artifacts: undefined as never, - }), - ).rejects.toThrow(/requires WorkspaceOptions.waitUntil/); - }); - it("validates timeout configuration", () => { expect( () => @@ -81,52 +60,11 @@ describe("WorkerJavaScriptBackend", () => { ).toThrow(/positive finite/); }); - it("cancels a started worker when waitUntil registration fails", async () => { - let entrypointDisposals = 0; - let workerDisposals = 0; - const workspace = new Workspace({ - storage: new SQLiteTestStorage(), - waitUntil() { - throw new Error("waitUntil unavailable"); - }, - backends: [ - new WorkerJavaScriptBackend({ - loader: { - load() { - return { - getEntrypoint() { - return { - evaluate: () => new Promise(() => undefined), - [Symbol.dispose]() { - entrypointDisposals += 1; - }, - }; - }, - [Symbol.dispose]() { - workerDisposals += 1; - }, - }; - }, - }, - }), - ], - }); - await workspace.fs.mkdir("/workspace", { recursive: true }); - const execution = await workspace.runtime.exec("export default 1", { encoding: "utf8" }); - await expect(execution.result()).resolves.toMatchObject({ - status: "failed", - stderr: expect.stringContaining("waitUntil unavailable"), - }); - expect(entrypointDisposals).toBe(1); - expect(workerDisposals).toBe(1); - }); - it("disposes Loader resources when evaluate throws synchronously", async () => { let entrypointDisposals = 0; let workerDisposals = 0; const workspace = new Workspace({ storage: new SQLiteTestStorage(), - waitUntil() {}, backends: [ new WorkerJavaScriptBackend({ loader: { @@ -411,15 +349,13 @@ describe("WorkerJavaScriptBackend", () => { await workspace.close(); }); - it("limits concurrent Dynamic Workers and attaches execution to waitUntil", async () => { + it("limits concurrent Dynamic Workers", async () => { let resolveEvaluation!: (value: { result: number }) => void; const evaluation = new Promise<{ result: number }>((resolve) => { resolveEvaluation = resolve; }); - const waitUntil = vi.fn<(promise: Promise) => void>(); const workspace = new Workspace({ storage: new SQLiteTestStorage(), - waitUntil, backends: [ new WorkerJavaScriptBackend({ maxConcurrentExecutions: 1, @@ -445,7 +381,6 @@ describe("WorkerJavaScriptBackend", () => { }); await workspace.fs.mkdir("/workspace", { recursive: true }); const first = await workspace.runtime.exec("export default 1", { id: "first" }); - expect(waitUntil).toHaveBeenCalledOnce(); await expect( workspace.runtime.exec("export default 2", { id: "second" }), ).rejects.toMatchObject({ code: "EEXEC_BUSY" }); @@ -513,7 +448,7 @@ describe("WorkerJavaScriptBackend", () => { releaseWrite(); const events = await terminal; expect(await fs.readFile("/workspace/output.txt", "utf8")).toBe("done"); - expect(events.at(-1)).toMatchObject({ name: "exit", value: 0 }); + expect(events.at(-1)).toMatchObject({ name: "exit", code: 0 }); }); it("streams stdout before user code returns", async () => { @@ -545,7 +480,7 @@ describe("WorkerJavaScriptBackend", () => { ); await exitReleased; controller.enqueue( - encoder.encode(`${JSON.stringify({ name: "exit", value: 0 })}\n`), + encoder.encode(`${JSON.stringify({ name: "exit", code: 0 })}\n`), ); controller.close(); }, @@ -632,7 +567,7 @@ describe("WorkerJavaScriptBackend", () => { for await (const event of execution.events) events.push(event); const exitIndex = events.findIndex((event) => event.name === "exit"); expect(exitIndex).toBeGreaterThanOrEqual(0); - expect(events[exitIndex]).toMatchObject({ name: "exit", value: 130 }); + expect(events[exitIndex]).toMatchObject({ name: "exit", code: 130 }); // The exit event is terminal: no stdout, stderr, or result follows it. expect(events.slice(exitIndex + 1)).toEqual([]); await handle.close(); @@ -684,7 +619,7 @@ describe("WorkerJavaScriptBackend", () => { const events = []; for await (const event of execution.events) events.push(event); const exit = events.find((event) => event.name === "exit"); - expect(exit).toMatchObject({ name: "exit", value: 1 }); + expect(exit).toMatchObject({ name: "exit", code: 1 }); expect(events.some((event) => event.name === "result")).toBe(false); await handle.close(); }); @@ -736,7 +671,7 @@ describe("WorkerJavaScriptBackend", () => { const events = []; for await (const event of execution.events) events.push(event); expect(aborted).toBe(true); - expect(events.at(-1)).toMatchObject({ name: "exit", value: 1 }); + expect(events.at(-1)).toMatchObject({ name: "exit", code: 1 }); }); it("waits for accepted host calls before reporting cancellation", async () => { @@ -806,7 +741,7 @@ describe("WorkerJavaScriptBackend", () => { expect(await fs.readFile("/workspace/output.txt", "utf8")).toBe("done"); const events = []; for await (const event of execution.events) events.push(event); - expect(events.at(-1)).toMatchObject({ name: "exit", value: 130 }); + expect(events.at(-1)).toMatchObject({ name: "exit", code: 130 }); }); it("settles subscribers when terminal persistence fails and repairs on reconnect", async () => { @@ -850,18 +785,18 @@ describe("WorkerJavaScriptBackend", () => { finish({ result: 1 }); const events = []; for await (const event of execution.events) events.push(event); - expect(events.at(-1)).toMatchObject({ name: "exit", value: 1 }); + expect(events.at(-1)).toMatchObject({ name: "exit", code: 1 }); const sameSessionReplay = await handle.getExec({ id: "storage-failure" }); const sameSessionEvents = []; for await (const event of sameSessionReplay.events) sameSessionEvents.push(event); - expect(sameSessionEvents.at(-1)).toMatchObject({ name: "exit", value: 1 }); + expect(sameSessionEvents.at(-1)).toMatchObject({ name: "exit", code: 1 }); db.run = originalRun as typeof db.run; const reconnected = await backend.connect(host); const replay = await reconnected.getExec({ id: "storage-failure" }); const repaired = []; for await (const event of replay.events) repaired.push(event); - expect(repaired.at(-1)).toMatchObject({ name: "exit", value: 1 }); + expect(repaired.at(-1)).toMatchObject({ name: "exit", code: 1 }); }); it("bounds durable completed-execution retention", async () => { diff --git a/packages/computer/src/backends/worker-javascript/worker-javascript.ts b/packages/computer/src/backends/worker-javascript/worker-javascript.ts index d0bdf43a..9023c2fe 100644 --- a/packages/computer/src/backends/worker-javascript/worker-javascript.ts +++ b/packages/computer/src/backends/worker-javascript/worker-javascript.ts @@ -133,7 +133,6 @@ interface ExecutionRecord { export class WorkerJavaScriptBackend implements WorkspaceModuleBackend { readonly protocol = "module" as const; - readonly requiresWaitUntil = true; readonly type = "worker-javascript"; readonly callable = true; readonly id: string; @@ -211,11 +210,6 @@ export class WorkerJavaScriptBackend implements WorkspaceModuleBackend { } async connect(host: WorkspaceModuleBackendHost): Promise { - if (!host.waitUntil) { - throw new Error( - "WorkerJavaScriptBackend requires WorkspaceOptions.waitUntil; pass ctx.waitUntil.bind(ctx).", - ); - } return new JavaScriptBackendHandle(this.#options, host); } } @@ -284,7 +278,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { "Execution was interrupted when its Workspace runtime restarted.\n", ), }, - { id, seq: record.events.length + 2, name: "exit", value: 1 }, + { id, seq: record.events.length + 2, name: "exit", code: 1 }, ]); } } @@ -415,7 +409,12 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { onComplete: () => this.#finalize(record), onError: (message) => this.#finalize(record, message), }); - this.#host.waitUntil?.(record.control.completion); + // The completion promise drives finalize (which writes the + // terminal SQL) and is no longer handed to a host lifetime + // hook. Observe it so a finalize rejection surfaces as a + // swallowed error rather than an unhandled rejection in the + // Durable Object. + void record.control.completion.catch(() => undefined); } catch (error) { record.control?.cancel(); await record.control?.completion.catch(() => undefined); @@ -507,7 +506,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { `${error instanceof Error ? error.message : String(error)}\n`, ), }, - { id: record.id, seq: record.events.length + 2, name: "exit", value: 1 }, + { id: record.id, seq: record.events.length + 2, name: "exit", code: 1 }, ]); return; } @@ -518,7 +517,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { name: "stderr", value: new TextEncoder().encode(message), }, - { id: record.id, seq: record.events.length + 2, name: "exit", value: 130 }, + { id: record.id, seq: record.events.length + 2, name: "exit", code: 130 }, ]); })(); record.finalization = finalization; @@ -562,7 +561,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { "Execution was interrupted when its Workspace runtime restarted or before its terminal state was persisted.\n", ), }, - { id, seq: record.events.length + 2, name: "exit", value: 1 }, + { id, seq: record.events.length + 2, name: "exit", code: 1 }, ]); } return record; @@ -612,11 +611,12 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { name: frame.name, value: frame.value, }); - } else if (frame.name === "result") { - record.result = frame.value; - record.hasResult = true; } else { - record.exitCode = frame.value; + record.exitCode = frame.code; + if ("result" in frame) { + record.result = frame.result; + record.hasResult = true; + } } } @@ -641,7 +641,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { `${error instanceof Error ? error.message : String(error)}\n`, ), }, - { id: record.id, seq: record.events.length + 2, name: "exit", value: 1 }, + { id: record.id, seq: record.events.length + 2, name: "exit", code: 1 }, ]); return; } @@ -655,7 +655,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { `${truncateUtf8(errorMessage, Math.max(0, this.#options.maxStdioBytes - 1))}\n`, ), }, - { id: record.id, seq: record.events.length + 2, name: "exit", value: 1 }, + { id: record.id, seq: record.events.length + 2, name: "exit", code: 1 }, ]); return; } @@ -671,27 +671,22 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { name: "stderr", value: new TextEncoder().encode("Execution ended without reporting a result.\n"), }, - { id: record.id, seq: record.events.length + 2, name: "exit", value: 1 }, + { id: record.id, seq: record.events.length + 2, name: "exit", code: 1 }, ]); return; } const exitCode = record.exitCode; - const terminal: WorkspaceRuntimeEvent[] = []; - if (exitCode === 0 && record.hasResult) { - terminal.push({ - id: record.id, - seq: record.events.length + 1, - name: "result", - value: record.result as WorkspaceRuntimeValue, - }); - } - terminal.push({ - id: record.id, - seq: record.events.length + terminal.length + 1, - name: "exit", - value: exitCode, - }); - this.#finish(record, exitCode === 0 ? "completed" : "failed", terminal); + const exit: WorkspaceRuntimeEvent = + exitCode === 0 && record.hasResult + ? { + id: record.id, + seq: record.events.length + 1, + name: "exit", + code: exitCode, + result: record.result as WorkspaceRuntimeValue, + } + : { id: record.id, seq: record.events.length + 1, name: "exit", code: exitCode }; + this.#finish(record, exitCode === 0 ? "completed" : "failed", [exit]); } #stream(record: ExecutionRecord, after?: number | "tail") { @@ -774,7 +769,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { "Execution failed because its terminal state could not be persisted.\n", ), }, - { id: record.id, seq: record.events.length + 2, name: "exit", value: 1 }, + { id: record.id, seq: record.events.length + 2, name: "exit", code: 1 }, ]; } record.status = settledStatus; @@ -889,7 +884,9 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { function encodeEvent(event: WorkspaceRuntimeEvent): Uint8Array { if (event.name === "stdout" || event.name === "stderr") return event.value; - return new TextEncoder().encode(JSON.stringify(event.value)); + const payload = + "result" in event ? { code: event.code, result: event.result } : { code: event.code }; + return new TextEncoder().encode(JSON.stringify(payload)); } function decodeEvent( @@ -899,10 +896,16 @@ function decodeEvent( payload: Uint8Array, ): WorkspaceRuntimeEvent { if (name === "stdout" || name === "stderr") return { id, seq, name, value: payload }; - const value = JSON.parse(new TextDecoder().decode(payload)) as unknown; - if (name === "exit") return { id, seq, name, value: Number(value) }; - assertRuntimeValue(value); - return { id, seq, name: "result", value }; + const decoded = JSON.parse(new TextDecoder().decode(payload)) as unknown; + if (typeof decoded === "object" && decoded !== null && "code" in decoded) { + const record = decoded as { code: unknown; result?: unknown }; + if ("result" in record) { + assertRuntimeValue(record.result); + return { id, seq, name: "exit", code: Number(record.code), result: record.result }; + } + return { id, seq, name: "exit", code: Number(record.code) }; + } + return { id, seq, name: "exit", code: Number(decoded) }; } function startJavaScriptExecution(options: { @@ -1158,12 +1161,11 @@ function runtimeWorkerModule(entryName: string, maxStdioBytes: number) { : module.default ?? null; const value = result ?? null; await host.assertResult(value); - enqueue({ name: "result", value }); - enqueue({ name: "exit", value: 0 }); + enqueue({ name: "exit", code: 0, result: value }); } catch (error) { const message = error instanceof Error ? error.message : String(error); enqueue({ name: "stderr", b64: toBase64(truncate(message) + "\\n") }); - enqueue({ name: "exit", value: 1 }); + enqueue({ name: "exit", code: 1 }); } await writeChain; try { diff --git a/packages/computer/src/backends/worker-shell/worker-shell.test.ts b/packages/computer/src/backends/worker-shell/worker-shell.test.ts index 83941c16..af021566 100644 --- a/packages/computer/src/backends/worker-shell/worker-shell.test.ts +++ b/packages/computer/src/backends/worker-shell/worker-shell.test.ts @@ -142,7 +142,7 @@ describe("WorkerShellBackend", () => { const backend = new WorkerShellBackend({ fetcher: () => fetcher }); const handle = await backend.connect(); - const envelope = await handle.rpc.shell.exec({ command: "echo hello" }); + const envelope = await handle.rpc.shell.exec({ source: "echo hello" }); const reader = envelope.events.getReader(); const seen: unknown[] = []; while (true) { @@ -155,7 +155,7 @@ describe("WorkerShellBackend", () => { expect(observedCommand).toBe("echo hello"); expect(seen).toEqual([ { id: "run-1", seq: 1, name: "stdout", value: encoder.encode("hello\n") }, - { id: "run-1", seq: 2, name: "exit", value: 0 }, + { id: "run-1", seq: 2, name: "exit", code: 0 }, ]); expect(envelope.id).toBe("run-1"); }); @@ -172,7 +172,7 @@ describe("WorkerShellBackend", () => { const backend = new WorkerShellBackend({ fetcher: () => fetcher }); const handle = await backend.connect(); const envelope = await handle.rpc.shell.exec({ - command: "printenv TOKEN", + source: "printenv TOKEN", env: { TOKEN: "secret", EMPTY: "" }, }); const reader = envelope.events.getReader(); @@ -194,7 +194,7 @@ describe("WorkerShellBackend", () => { }), })); const handle = await new WorkerShellBackend({ fetcher: () => fetcher }).connect(); - const envelope = await handle.rpc.shell.exec({ command: "bad" }); + const envelope = await handle.rpc.shell.exec({ source: "bad" }); await expect(envelope.events.getReader().read()).rejects.toMatchObject({ code: "EPROTOCOL" }); }); @@ -214,7 +214,7 @@ describe("WorkerShellBackend", () => { await ws.ready(); const backend = new WorkerShellBackend({ fetcher: () => fetcher }); const handle = await backend.connect(); - await handle.rpc.shell.exec({ command: "x", cwd: "/workspace/src", id: "fixed" }); + await handle.rpc.shell.exec({ source: "x", cwd: "/workspace/src", id: "fixed" }); expect(observed?.cwd).toBe("/workspace/src"); expect(observed?.id).toBe("fixed"); }); @@ -358,8 +358,8 @@ describe("WorkerShellBackend", () => { }, }); const handle = await backend.connect(); - await handle.rpc.shell.exec({ command: "true" }); - await handle.rpc.shell.exec({ command: "true" }); + await handle.rpc.shell.exec({ source: "true" }); + await handle.rpc.shell.exec({ source: "true" }); expect(factoryCalls).toBe(1); }); }); diff --git a/packages/computer/src/backends/worker-shell/worker-shell.ts b/packages/computer/src/backends/worker-shell/worker-shell.ts index b78c9abc..e413ed42 100644 --- a/packages/computer/src/backends/worker-shell/worker-shell.ts +++ b/packages/computer/src/backends/worker-shell/worker-shell.ts @@ -171,7 +171,7 @@ export class WorkerShellBackend implements WorkspaceBackend { const shell: ShellRPC = { async exec(input) { const envelope = await fetcher.exec({ - command: input.command, + command: input.source, cwd: input.cwd, id: input.id, timeoutMs: input.timeoutMs, @@ -191,7 +191,7 @@ export class WorkerShellBackend implements WorkspaceBackend { // The user Worker has no DB-backed log to dispose; the // event stream itself is the only resource and it ends // with the run. Treated as a no-op on this backend so - // the WorkspaceShell surface stays uniform. + // the ShellRPC surface stays uniform. }, }; @@ -262,7 +262,7 @@ export class WorkerShellBackend implements WorkspaceBackend { } // Decode a byte-framed event stream produced by ShellWorker -// into the structured ExecEvent shape WorkspaceShell expects. +// into the structured ExecEvent shape the runtime expects. // Frames are newline-delimited JSON objects. function decodeFramedEvents(source: ReadableStream): ReadableStream { const decoder = new TextDecoder(); @@ -337,8 +337,8 @@ function reshape(event: { }): ExecEvent { // ShellWorker ships stdout / stderr values as utf8 strings; // ExecEvent on the wire carries Uint8Array. Re-encode so the - // existing WorkspaceShell utf8 decoder transforms in shell.ts - // see the shape they already handle. + // runtime's utf8 decoder transforms see the shape they already + // handle. if (event.name === "stdout" || event.name === "stderr") { return { id: event.id, @@ -347,7 +347,7 @@ function reshape(event: { value: new TextEncoder().encode(event.value as string), }; } - return { id: event.id, seq: event.seq, name: "exit", value: event.value as number }; + return { id: event.id, seq: event.seq, name: "exit", code: event.value as number }; } function disposeQuietly(value: { [Symbol.dispose]?: () => void }) { diff --git a/packages/computer/src/client.test.ts b/packages/computer/src/client.test.ts index 2e4e734d..035b0e9b 100644 --- a/packages/computer/src/client.test.ts +++ b/packages/computer/src/client.test.ts @@ -37,7 +37,7 @@ function fakeRuntime(promisedProperties = false) { const stream = new ReadableStream({ start(c) { c.enqueue( - new TextEncoder().encode(`${JSON.stringify({ id, seq: 0, name: "exit", value: 0 })}\n`), + new TextEncoder().encode(`${JSON.stringify({ id, seq: 0, name: "exit", code: 0 })}\n`), ); c.close(); }, @@ -149,7 +149,7 @@ describe("getWorkspace — remote dispatch", () => { const handle = await ws.runtime.exec("echo ok"); const events = []; for await (const event of handle) events.push(event); - expect(events).toEqual([{ id: expect.any(String), seq: 0, name: "exit", value: 0 }]); + expect(events).toEqual([{ id: expect.any(String), seq: 0, name: "exit", code: 0 }]); }); it("exposes the complete runtime lifecycle and preserves handle ids", async () => { @@ -297,11 +297,16 @@ describe("client runtime.exec — remote handle rebuild", () => { const { host } = fakeRemote(); const ws = await getWorkspace(host); const handle = await ws.runtime.exec("echo hi"); - const events: Array<{ name: string; value: unknown }> = []; - for await (const event of handle as AsyncIterable<{ name: string; value: unknown }>) { - events.push({ name: event.name, value: event.value }); + type Collected = { name: "stdout" | "stderr"; value: unknown } | { name: "exit"; code: number }; + const events: Collected[] = []; + for await (const event of handle as AsyncIterable) { + events.push( + event.name === "exit" + ? { name: "exit", code: event.code } + : { name: event.name, value: event.value }, + ); } - expect(events).toEqual([{ name: "exit", value: 0 }]); + expect(events).toEqual([{ name: "exit", code: 0 }]); }); it("throws if result() is called after the stream has started", async () => { diff --git a/packages/computer/src/observe-integration.test.ts b/packages/computer/src/observe-integration.test.ts index 5e6a5ab0..628a889f 100644 --- a/packages/computer/src/observe-integration.test.ts +++ b/packages/computer/src/observe-integration.test.ts @@ -242,12 +242,12 @@ describe("Workspace observer — runtime stub", () => { const execed: string[] = []; const shellRpc: import("@cloudflare/computer-rpc").ShellRPC = { async exec(input) { - execed.push(input.command); + execed.push(input.source); return { id: "exec-1", events: new ReadableStream({ start(c) { - c.enqueue({ id: "exec-1", seq: 0, name: "exit", value: 0 }); + c.enqueue({ id: "exec-1", seq: 0, name: "exit", code: 0 }); c.close(); }, }), diff --git a/packages/computer/src/observe.ts b/packages/computer/src/observe.ts index c2e8d2ba..748ab904 100644 --- a/packages/computer/src/observe.ts +++ b/packages/computer/src/observe.ts @@ -7,7 +7,7 @@ // connect() + watermark reconcile. // workspace.sync.push — one per `Workspace.push()` call. // workspace.sync.pull — one per `Workspace.pull()` call. -// workspace.runtime.exec — one per `WorkspaceShell.exec()` call, +// workspace.runtime.exec — one per command executor exec() call, // covering pre-exec push, the spawn // request, and (when `result()` is // awaited) the post-drain pull. diff --git a/packages/computer/src/retry.test.ts b/packages/computer/src/retry.test.ts index 1a7855e3..f34a9734 100644 --- a/packages/computer/src/retry.test.ts +++ b/packages/computer/src/retry.test.ts @@ -68,7 +68,7 @@ function retryBackend(options: { id: "command-1", events: new ReadableStream({ start(controller) { - controller.enqueue({ id: "command-1", seq: 1, name: "exit", value: 0 }); + controller.enqueue({ id: "command-1", seq: 1, name: "exit", code: 0 }); controller.close(); }, }), diff --git a/packages/computer/src/runtime/runtime.test.ts b/packages/computer/src/runtime/runtime.test.ts index cc27ba26..e03b1ad3 100644 --- a/packages/computer/src/runtime/runtime.test.ts +++ b/packages/computer/src/runtime/runtime.test.ts @@ -1,8 +1,11 @@ import { describe, expect, it, vi } from "vitest"; -import type { WorkspaceShell } from "../shell.js"; import { WorkspaceRuntime } from "./runtime.js"; -import type { ModuleExecutionEnvelope, WorkspaceModuleBackendHandle } from "./types.js"; +import type { + ModuleExecutionEnvelope, + WorkspaceModuleBackendHandle, + WorkspaceRuntimeEvent, +} from "./types.js"; function emptyEnvelope(id: string): ModuleExecutionEnvelope { return { @@ -25,13 +28,161 @@ function moduleHandleStub(): WorkspaceModuleBackendHandle { }; } +function eventStream(events: WorkspaceRuntimeEvent[]): ReadableStream { + return new ReadableStream({ + start(controller) { + for (const event of events) controller.enqueue(event); + controller.close(); + }, + }); +} + +// A backend whose exec replays a fixed event sequence. Drives the +// runtime's encoding transform and result drain — the work the host +// shell facade used to own before every backend shared one path. +function replayBackend(events: WorkspaceRuntimeEvent[]): WorkspaceModuleBackendHandle { + return { + exec: vi.fn(async (input) => ({ id: input.id ?? "exec", events: eventStream(events) })), + getExec: vi.fn(async (input) => ({ id: input.id, events: eventStream(events) })), + killExec: vi.fn(async () => {}), + disposeExec: vi.fn(async () => {}), + close: vi.fn(async () => {}), + }; +} + +function runtimeFor(handle: WorkspaceModuleBackendHandle): WorkspaceRuntime { + return new WorkspaceRuntime({ + callableBackendIds: new Set(), + backendHandle: async () => handle, + resolveBackendId: () => "backend", + }); +} + +function stdout(seq: number, value: Uint8Array): WorkspaceRuntimeEvent { + return { id: "e", seq, name: "stdout", value }; +} +function stderr(seq: number, value: Uint8Array): WorkspaceRuntimeEvent { + return { id: "e", seq, name: "stderr", value }; +} +function bytes(text: string): Uint8Array { + return new TextEncoder().encode(text); +} + +describe("WorkspaceRuntime result accumulation", () => { + it("concatenates stdout chunks in arrival order as raw bytes", async () => { + const runtime = runtimeFor( + replayBackend([ + stdout(1, bytes("one")), + stdout(2, bytes("two")), + stdout(3, bytes("three")), + { id: "e", seq: 4, name: "exit", code: 0 }, + ]), + ); + const result = await (await runtime.exec("noop")).result(); + expect(result.stdout).toBeInstanceOf(Uint8Array); + expect(new TextDecoder().decode(result.stdout as Uint8Array)).toBe("onetwothree"); + }); + + it("keeps stdout and stderr separate", async () => { + const runtime = runtimeFor( + replayBackend([ + stdout(1, bytes("out")), + stderr(2, bytes("err")), + stdout(3, bytes("out2")), + { id: "e", seq: 4, name: "exit", code: 0 }, + ]), + ); + const result = await (await runtime.exec("noop")).result(); + expect(new TextDecoder().decode(result.stdout as Uint8Array)).toBe("outout2"); + expect(new TextDecoder().decode(result.stderr as Uint8Array)).toBe("err"); + }); + + it("captures the exit code from the exit event", async () => { + const runtime = runtimeFor(replayBackend([{ id: "e", seq: 1, name: "exit", code: 42 }])); + const result = await (await runtime.exec("noop")).result(); + expect(result.exitCode).toBe(42); + }); + + it("maps signal exit codes to a cancelled status", async () => { + for (const code of [129, 130, 137, 143]) { + const runtime = runtimeFor(replayBackend([{ id: "e", seq: 1, name: "exit", code: code }])); + const result = await (await runtime.exec("noop")).result(); + expect(result.status).toBe("cancelled"); + expect(result.exitCode).toBe(code); + } + }); + + it("reports exit code -1 when the stream closes without an exit event", async () => { + const runtime = runtimeFor(replayBackend([stdout(1, bytes("partial"))])); + const result = await (await runtime.exec("noop")).result(); + expect(result.exitCode).toBe(-1); + expect(result.status).toBe("failed"); + }); +}); + +describe("WorkspaceRuntime utf8 encoding", () => { + it("returns stdout / stderr as strings when encoding is 'utf8'", async () => { + const runtime = runtimeFor( + replayBackend([ + stdout(1, bytes("hello ")), + stderr(2, bytes("warn")), + stdout(3, bytes("world")), + { id: "e", seq: 4, name: "exit", code: 0 }, + ]), + ); + const result = await (await runtime.exec("noop", { encoding: "utf8" })).result(); + expect(result.stdout).toBe("hello world"); + expect(result.stderr).toBe("warn"); + }); + + it("decodes multi-byte UTF-8 split across chunks correctly", async () => { + const partyHat = new Uint8Array([0xf0, 0x9f, 0x8e, 0x89]); + const runtime = runtimeFor( + replayBackend([ + stdout(1, partyHat.subarray(0, 3)), + stdout(2, partyHat.subarray(3)), + { id: "e", seq: 3, name: "exit", code: 0 }, + ]), + ); + const result = await (await runtime.exec("noop", { encoding: "utf8" })).result(); + expect(result.stdout).toBe("\u{1f389}"); + }); + + it("keeps the stdout and stderr decoders independent", async () => { + const partyHat = new Uint8Array([0xf0, 0x9f, 0x8e, 0x89]); + const heart = new Uint8Array([0xe2, 0x9d, 0xa4]); + const runtime = runtimeFor( + replayBackend([ + stdout(1, partyHat.subarray(0, 2)), + stderr(2, heart.subarray(0, 2)), + stdout(3, partyHat.subarray(2)), + stderr(4, heart.subarray(2)), + { id: "e", seq: 5, name: "exit", code: 0 }, + ]), + ); + const result = await (await runtime.exec("noop", { encoding: "utf8" })).result(); + expect(result.stdout).toBe("\u{1f389}"); + expect(result.stderr).toBe("\u2764"); + }); + + it("preserves encoding when consuming the stream directly", async () => { + const runtime = runtimeFor( + replayBackend([stdout(1, bytes("stream-mode")), { id: "e", seq: 2, name: "exit", code: 0 }]), + ); + const handle = await runtime.exec("noop", { encoding: "utf8" }); + const seen: unknown[] = []; + for await (const event of handle) { + if (event.name === "stdout") seen.push(event.value); + } + expect(seen).toEqual(["stream-mode"]); + }); +}); + describe("WorkspaceRuntime callable gate", () => { it("rejects structured input for a non-callable backend", async () => { const runtime = new WorkspaceRuntime({ - commandBackendIds: new Set(["worker-shell"]), callableBackendIds: new Set(), - shell: () => ({}) as unknown as WorkspaceShell, - moduleHandle: async () => moduleHandleStub(), + backendHandle: async () => moduleHandleStub(), resolveBackendId: () => "worker-shell", }); @@ -43,10 +194,8 @@ describe("WorkspaceRuntime callable gate", () => { it("accepts structured input for a callable module backend", async () => { const handle = moduleHandleStub(); const runtime = new WorkspaceRuntime({ - commandBackendIds: new Set(), callableBackendIds: new Set(["worker-javascript"]), - shell: () => ({}) as unknown as WorkspaceShell, - moduleHandle: async () => handle, + backendHandle: async () => handle, resolveBackendId: () => "worker-javascript", }); diff --git a/packages/computer/src/runtime/runtime.ts b/packages/computer/src/runtime/runtime.ts index b426a214..9086b2c9 100644 --- a/packages/computer/src/runtime/runtime.ts +++ b/packages/computer/src/runtime/runtime.ts @@ -1,7 +1,8 @@ import type { SkippedEntry } from "@cloudflare/dofs"; -import type { ExecEncoding, ExecHandle, WorkspaceShell } from "../shell.js"; +import type { ExecEncoding } from "../shell.js"; import type { + ModuleExecutionEnvelope, WorkspaceModuleBackendHandle, WorkspaceRuntimeDisposeOptions, WorkspaceRuntimeEvent, @@ -13,13 +14,19 @@ import type { } from "./types.js"; interface WorkspaceRuntimeRouterOptions { - commandBackendIds: ReadonlySet; callableBackendIds: ReadonlySet; - shell: () => WorkspaceShell; - moduleHandle: (id: string) => Promise; + backendHandle: (id: string) => Promise; resolveBackendId: (id: string | undefined) => string; } +// The error a caller sees when it hands structured `input` to a +// backend that does not accept it. Exported so the exec tool rejects +// with the same wording the runtime raises, instead of a second copy +// that could drift. +export function notCallableMessage(backend: string): string { + return `Backend ${JSON.stringify(backend)} is not callable; it does not accept structured input.`; +} + export class WorkspaceRuntime { readonly #options: WorkspaceRuntimeRouterOptions; @@ -43,35 +50,9 @@ export class WorkspaceRuntime { if (options.id !== undefined) assertExecutionId(options.id); const backend = this.#backend(options.backend); if (options.input !== undefined && !this.#options.callableBackendIds.has(backend)) { - throw new Error( - `Backend ${JSON.stringify(backend)} is not callable; it does not accept structured input.`, - ); - } - if (this.#options.commandBackendIds.has(backend)) { - if (options.input !== undefined) { - throw new Error( - `Backend ${JSON.stringify(backend)} is not callable; it does not accept structured input.`, - ); - } - const shell = this.#options.shell(); - const handle = await ( - shell.exec as unknown as ( - command: string, - options: Record, - ) => Promise> - )(source, { - backend, - cwd: options.cwd, - encoding: options.encoding, - id: options.id, - timeoutMs: options.timeoutMs, - env: options.env, - stdin: options.stdin, - }); - return wrapCommandHandle(handle, backend); + throw new Error(notCallableMessage(backend)); } - - const runtime = await this.#options.moduleHandle(backend); + const runtime = await this.#options.backendHandle(backend); const envelope = await runtime.exec({ id: options.id, source, @@ -81,7 +62,15 @@ export class WorkspaceRuntime { stdin: options.stdin, timeoutMs: options.timeoutMs, }); - return wrapModuleHandle(runtime, backend, envelope.id, envelope.events, options.encoding); + return wrapModuleHandle( + runtime, + backend, + envelope.id, + envelope.events, + options.encoding, + true, + envelope.sync, + ); } getExec(id: string): Promise>; @@ -99,36 +88,7 @@ export class WorkspaceRuntime { ): Promise> { assertExecutionId(id); const backend = this.#backend(options.backend); - if (this.#options.commandBackendIds.has(backend)) { - const shell = this.#options.shell(); - const handle = await ( - shell.get as unknown as ( - id: string, - options: Record, - ) => Promise> - )(id, { - backend, - encoding: options.encoding, - resume: options.resume, - }); - const resultHandle = - options.resume === undefined || options.resume === "full" - ? undefined - : () => - ( - shell.get as unknown as ( - id: string, - options: Record, - ) => Promise> - )(id, { - backend, - encoding: options.encoding, - resume: "full", - }); - return wrapCommandHandle(handle, backend, resultHandle); - } - - const runtime = await this.#options.moduleHandle(backend); + const runtime = await this.#options.backendHandle(backend); const envelope = await runtime.getExec({ id, after: resumeToAfter(options.resume) }); return wrapModuleHandle( runtime, @@ -137,27 +97,20 @@ export class WorkspaceRuntime { envelope.events, options.encoding, options.resume === undefined || options.resume === "full", + envelope.sync, ); } async killExec(id: string, options: WorkspaceRuntimeKillOptions = {}): Promise { assertExecutionId(id); const backend = this.#backend(options.backend); - if (this.#options.commandBackendIds.has(backend)) { - await this.#options.shell().kill(id, options.signal, { backend }); - return; - } - await (await this.#options.moduleHandle(backend)).killExec({ id, signal: options.signal }); + await (await this.#options.backendHandle(backend)).killExec({ id, signal: options.signal }); } async disposeExec(id: string, options: WorkspaceRuntimeDisposeOptions = {}): Promise { assertExecutionId(id); const backend = this.#backend(options.backend); - if (this.#options.commandBackendIds.has(backend)) { - await this.#options.shell().dispose(id, { backend }); - return; - } - await (await this.#options.moduleHandle(backend)).disposeExec({ id }); + await (await this.#options.backendHandle(backend)).disposeExec({ id }); } #backend(requested: string | undefined): string { @@ -171,84 +124,6 @@ export class WorkspaceRuntime { } } -function wrapCommandHandle( - handle: ExecHandle, - backend: string, - lazyResultHandle?: () => Promise>, -): WorkspaceRuntimeExecHandle { - let claimed: "result" | "stream" | undefined; - let reader: ReadableStreamDefaultReader> | undefined; - let resultPromise: Promise> | undefined; - const stream = new ReadableStream>( - { - async pull(controller) { - if (claimed === "result") { - controller.error(new Error("runtime handle already consumed by result()")); - return; - } - claimed = "stream"; - reader ??= handle.getReader() as ReadableStreamDefaultReader>; - try { - const next = await reader.read(); - if (next.done) { - reader.releaseLock(); - reader = undefined; - controller.close(); - } else controller.enqueue(next.value); - } catch (error) { - reader?.releaseLock(); - reader = undefined; - controller.error(error); - } - }, - async cancel(reason) { - if (reader) { - try { - await reader.cancel(reason); - } finally { - reader.releaseLock(); - reader = undefined; - } - } else await handle.cancel(reason); - }, - }, - { highWaterMark: 0 }, - ) as WorkspaceRuntimeExecHandle; - Object.defineProperties(stream, { - id: { value: handle.id, enumerable: false }, - backend: { value: backend, enumerable: false }, - result: { - value: (): Promise> => { - if (claimed === "stream") { - throw new Error("runtime handle already streaming: result() and streaming are exclusive"); - } - claimed = "result"; - resultPromise ??= (async () => { - if (lazyResultHandle) await handle.cancel("result() requested a full replay"); - const result = lazyResultHandle - ? await (await lazyResultHandle()).result() - : await handle.result(); - return { - status: - result.exitCode === 0 - ? "completed" - : isCancellationExitCode(result.exitCode) - ? "cancelled" - : "failed", - ...result, - }; - })(); - return resultPromise; - }, - }, - kill: { - value: (signal?: WorkspaceRuntimeKillOptions["signal"]) => handle.kill(signal), - }, - [Symbol.dispose]: { value: () => handle[Symbol.dispose]() }, - }); - return stream; -} - function wrapModuleHandle( runtime: WorkspaceModuleBackendHandle, backend: string, @@ -256,6 +131,7 @@ function wrapModuleHandle( source: ReadableStream, encoding: E | undefined, resultMayUseSource = true, + sync?: ModuleExecutionEnvelope["sync"], ): WorkspaceRuntimeExecHandle { let claimed: "result" | "stream" | undefined; let sourceCancelled = false; @@ -314,10 +190,11 @@ function wrapModuleHandle( resultReader = active; }; if (resultMayUseSource && !sourceCancelled) { - return drainModuleResult(source, encoding, setReader); + return drainModuleResult(source, encoding, setReader, sync); } if (!sourceCancelled) await source.cancel("result() requested a full replay"); - return drainModuleResult((await runtime.getExec({ id })).events, encoding, setReader); + const replay = await runtime.getExec({ id }); + return drainModuleResult(replay.events, encoding, setReader, replay.sync); })(); return resultPromise; }, @@ -401,7 +278,10 @@ function transformModuleEvents( }), } as WorkspaceRuntimeEvent); } else { - if (event.name === "exit") flushPending(controller, event.seq); + // `exit` is the only non-stdio event; flush any buffered + // partial output before the terminal event so a trailing + // multi-byte remainder lands ahead of it. + flushPending(controller, event.seq); enqueue(controller, event as WorkspaceRuntimeEvent); } }, @@ -414,11 +294,14 @@ async function drainModuleResult( events: ReadableStream, encoding: E | undefined, setReader: (reader: ReadableStreamDefaultReader | undefined) => void, + sync?: ModuleExecutionEnvelope["sync"], ): Promise> { const stdout: Uint8Array[] = []; const stderr: Uint8Array[] = []; let value: WorkspaceRuntimeResult["value"]; - let exitCode = 1; + // -1 marks a stream that closed without an exit frame, keeping that + // case distinct from a genuine exit 1. Both settle as "failed". + let exitCode = -1; const reader = events.getReader(); setReader(reader); try { @@ -428,23 +311,29 @@ async function drainModuleResult( const event = next.value; if (event.name === "stdout") stdout.push(event.value); if (event.name === "stderr") stderr.push(event.value); - if (event.name === "result") value = event.value; - if (event.name === "exit") exitCode = event.value; + if (event.name === "exit") { + exitCode = event.code; + if ("result" in event) value = event.result; + } } } finally { reader.releaseLock(); setReader(undefined); } + // A backend with a remote store reports its sync bracket stats; + // the pull outcome settles once the event stream above drains. + const pull = sync ? await sync.outcome : undefined; return { - status: exitCode === 0 ? "completed" : exitCode === 130 ? "cancelled" : "failed", + status: + exitCode === 0 ? "completed" : isCancellationExitCode(exitCode) ? "cancelled" : "failed", exitCode, stdout: join(stdout, encoding) as WorkspaceRuntimeResult["stdout"], stderr: join(stderr, encoding) as WorkspaceRuntimeResult["stderr"], ...(value === undefined ? {} : { value }), - pushed: 0, - pulled: 0, - skipped: [] as SkippedEntry[], - sync: { status: "complete", applied: 0, skipped: [] }, + pushed: sync?.pushed ?? 0, + pulled: pull?.applied ?? 0, + skipped: pull?.skipped ?? ([] as SkippedEntry[]), + sync: pull?.sync ?? { status: "complete", applied: 0, skipped: [] }, }; } diff --git a/packages/computer/src/runtime/types.ts b/packages/computer/src/runtime/types.ts index 52d505ce..cf3d9931 100644 --- a/packages/computer/src/runtime/types.ts +++ b/packages/computer/src/runtime/types.ts @@ -87,8 +87,7 @@ type RuntimeChunk = E extends "utf8" ? string : Uint8Arr export type WorkspaceRuntimeEvent = | { id: string; seq: number; name: "stdout"; value: RuntimeChunk } | { id: string; seq: number; name: "stderr"; value: RuntimeChunk } - | { id: string; seq: number; name: "result"; value: WorkspaceRuntimeValue } - | { id: string; seq: number; name: "exit"; value: number }; + | { id: string; seq: number; name: "exit"; code: number; result?: WorkspaceRuntimeValue }; export interface WorkspaceRuntimeResult { status: WorkspaceRuntimeStatus; @@ -150,6 +149,15 @@ export interface ModuleExecutionInput { export interface ModuleExecutionEnvelope { id: string; events: ReadableStream; + // Sync bracket stats for a backend that pairs with a remote store. + // The pre-exec push count is known when the envelope is created; + // the post-drain pull outcome settles once `events` is consumed to + // its end. Absent for backends that reuse the host store, whose + // result reports zeroed stats. + sync?: { + pushed: number; + outcome: Promise<{ applied: number; skipped: SkippedEntry[]; sync: ExecSyncResult }>; + }; } export interface WorkspaceModuleBackendHandle { @@ -157,7 +165,10 @@ export interface WorkspaceModuleBackendHandle { getExec(input: { id: string; after?: number | "tail" }): Promise; killExec(input: { id: string; signal?: KillSignal }): Promise; disposeExec(input: { id: string }): Promise; - close(): Promise; + // Tear down a backend-owned transport. The command adapter omits + // it: a command backend's transport is closed through its + // BackendHandle, not through the adapter the runtime consumes. + close?(): Promise; } export type WorkspaceModuleBackendHost = import("../backend.js").WorkspaceBackendHost; @@ -165,7 +176,6 @@ export type WorkspaceModuleBackendHost = import("../backend.js").WorkspaceBacken export interface WorkspaceModuleBackend { readonly protocol: "module"; readonly id: string; - readonly requiresWaitUntil?: boolean; readonly type: string; readonly callable?: boolean; connect(host: WorkspaceModuleBackendHost): Promise; diff --git a/packages/computer/src/runtime/wire.test.ts b/packages/computer/src/runtime/wire.test.ts new file mode 100644 index 00000000..7363b0cd --- /dev/null +++ b/packages/computer/src/runtime/wire.test.ts @@ -0,0 +1,46 @@ +import { describe, expect, it } from "vitest"; + +import type { WorkspaceRuntimeEvent } from "./types.js"; +import { decodeRuntimeEvents, encodeRuntimeEvent } from "./wire.js"; + +function streamOf(...events: WorkspaceRuntimeEvent[]): ReadableStream { + return new ReadableStream({ + start(controller) { + for (const event of events) controller.enqueue(encodeRuntimeEvent(event)); + controller.close(); + }, + }); +} + +async function collect( + stream: ReadableStream, +): Promise { + const events: WorkspaceRuntimeEvent[] = []; + const reader = stream.getReader(); + while (true) { + const next = await reader.read(); + if (next.done) break; + events.push(next.value as WorkspaceRuntimeEvent); + } + return events; +} + +describe("runtime wire codec", () => { + it("round-trips an exit event carrying a structured result", async () => { + const events = await collect( + decodeRuntimeEvents( + streamOf({ id: "e-1", seq: 2, name: "exit", code: 0, result: { a: [1, 2, null] } }), + ), + ); + expect(events).toEqual([ + { id: "e-1", seq: 2, name: "exit", code: 0, result: { a: [1, 2, null] } }, + ]); + }); + + it("round-trips an exit event with no result", async () => { + const events = await collect( + decodeRuntimeEvents(streamOf({ id: "e-1", seq: 1, name: "exit", code: 1 })), + ); + expect(events).toEqual([{ id: "e-1", seq: 1, name: "exit", code: 1 }]); + }); +}); diff --git a/packages/computer/src/runtime/wire.ts b/packages/computer/src/runtime/wire.ts index fac275bf..f39acffe 100644 --- a/packages/computer/src/runtime/wire.ts +++ b/packages/computer/src/runtime/wire.ts @@ -4,8 +4,7 @@ import type { WorkspaceRuntimeEvent, WorkspaceRuntimeValue } from "./types.js"; type RuntimeFrame = | { id: string; seq: number; name: "stdout" | "stderr"; enc: "utf8"; value: string } | { id: string; seq: number; name: "stdout" | "stderr"; enc: "b64"; value: string } - | { id: string; seq: number; name: "result"; value: WorkspaceRuntimeValue } - | { id: string; seq: number; name: "exit"; value: number }; + | { id: string; seq: number; name: "exit"; code: number; result?: WorkspaceRuntimeValue }; function toBase64(bytes: Uint8Array): string { let binary = ""; @@ -24,7 +23,7 @@ function fromBase64(text: string): Uint8Array { export function encodeRuntimeEvent(event: WorkspaceRuntimeEvent): Uint8Array { let frame: RuntimeFrame; - if (event.name === "exit" || event.name === "result") frame = event; + if (event.name === "exit") frame = event; else if (typeof event.value === "string") { frame = { id: event.id, seq: event.seq, name: event.name, enc: "utf8", value: event.value }; } else { @@ -40,7 +39,7 @@ export function encodeRuntimeEvent(event: WorkspaceRuntimeEvent): } function decodeFrame(frame: RuntimeFrame): WorkspaceRuntimeEvent { - if (frame.name === "exit" || frame.name === "result") return frame; + if (frame.name === "exit") return frame; return { id: frame.id, seq: frame.seq, diff --git a/packages/computer/src/shell.test.ts b/packages/computer/src/shell.test.ts index 99f937b5..bcbd8fdb 100644 --- a/packages/computer/src/shell.test.ts +++ b/packages/computer/src/shell.test.ts @@ -1,25 +1,20 @@ -// Unit tests for WorkspaceShell. The harness shell.test.ts under +// Unit tests for CommandExecutor. The harness shell.test.ts under // src/test-harness covers the wire end-to-end against a real computerd // container; these tests run in-process with a fake WorkspaceRPC so -// the host-side facade (RPC forwarding, handle shape, encoding, -// result accumulation, push/pull bracket math) is exercised -// without needing Docker. +// the host-side executor (RPC forwarding, envelope shape, push/pull +// bracket math, reattach) is exercised without needing Docker. +// +// Encoding and result accumulation are the runtime's job now — the +// executor returns raw events and the runtime drains them — so those +// cases live in runtime.test.ts, not here. import type { ExecEvent, ShellRPC, SyncRPC, WorkspaceRPC } from "@cloudflare/computer-rpc"; import { describe, expect, it } from "vitest"; -import type { KillSignal } from "./shell.js"; -import { type Sync, WorkspaceShell } from "./shell.js"; +import { CommandExecutor, type KillSignal, type Sync } from "./shell.js"; -// Inert sync impl. The shell unit tests aren't exercising the -// push/pull bracket — they just need a Sync that returns 0 from -// both methods so the bracket plumbing is a no-op. Workspace.test.ts -// covers the bracket against a real workspace. -// -// pull() returns the dofs ApplyResult shape ({ applied, skipped }); -// tests that only care about counts use the `applied` helper below -// to build a synthetic result. Tests that want to exercise the -// skipped path build the shape inline. +// Inert sync impl. Tests that don't exercise the bracket use a Sync +// that returns 0 from both halves. Tests that do build their own. function applied(n: number) { return { applied: n, skipped: [] }; } @@ -35,14 +30,8 @@ function makeSync(): Sync { }; } -// --------------------------------------------------------------------------- -// Test fixture: a fully synthesised WorkspaceRPC. Each helper method on the -// returned object also exposes the call log on the `calls` field so tests -// can assert on what the facade forwarded. -// --------------------------------------------------------------------------- - interface ExecCall { - command: string; + source: string; id: string | undefined; cwd: string | undefined; timeoutMs: number | undefined; @@ -68,26 +57,16 @@ interface FakeRpc { } interface FakeRpcOptions { - // Events to push onto the stream returned by shell.exec / getExec. - // The runner's id is stamped onto each event before enqueue. events?: ExecEvent[]; - // Optional: shell.exec rejects with this error. throwOnExec?: Error; - // Optional: enqueue events, then error the stream with this. streamError?: Error; - // Optional: id the runner mints when the caller doesn't supply one. - // Defaults to "runner-minted-id". mintedId?: string; } function fakeRpc(options: FakeRpcOptions = {}): FakeRpc { - const events = options.events ?? [{ id: "_", seq: 1, name: "exit", value: 0 }]; + const events = options.events ?? [{ id: "_", seq: 1, name: "exit", code: 0 }]; const mintedId = options.mintedId ?? "runner-minted-id"; - const calls: FakeRpc["calls"] = { - exec: [], - getExec: [], - killExec: [], - }; + const calls: FakeRpc["calls"] = { exec: [], getExec: [], killExec: [] }; function makeStream(id: string): ReadableStream { return new ReadableStream({ @@ -129,7 +108,7 @@ function fakeRpc(options: FakeRpcOptions = {}): FakeRpc { const shell: ShellRPC = { async exec(input) { calls.exec.push({ - command: input.command, + source: input.source, id: input.id, cwd: input.cwd, timeoutMs: input.timeoutMs, @@ -151,79 +130,83 @@ function fakeRpc(options: FakeRpcOptions = {}): FakeRpc { return { rpc: { sync, shell }, calls }; } -// Convenience: a stream-event with the encoder bytes inlined. function stdout(seq: number, text: string): ExecEvent { return { id: "_", seq, name: "stdout", value: new TextEncoder().encode(text) }; } -function stderr(seq: number, text: string): ExecEvent { - return { id: "_", seq, name: "stderr", value: new TextEncoder().encode(text) }; -} function exit(seq: number, code: number): ExecEvent { - return { id: "_", seq, name: "exit", value: code }; + return { id: "_", seq, name: "exit", code: code }; } -// --------------------------------------------------------------------------- -// exec() — RPC forwarding -// --------------------------------------------------------------------------- +// Drain an execution's events to completion and settle its sync +// outcome, mirroring what the runtime does when it wraps the handle. +async function drain(execution: { + events: ReadableStream; + sync: { pushed: number; outcome: Promise }; +}) { + const seen: ExecEvent[] = []; + const reader = execution.events.getReader(); + try { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + seen.push(value); + } + } finally { + reader.releaseLock(); + } + const outcome = await execution.sync.outcome; + return { seen, pushed: execution.sync.pushed, outcome }; +} -describe("WorkspaceShell.exec — RPC forwarding", () => { - it("forwards the command verbatim", async () => { +describe("CommandExecutor.exec — RPC forwarding", () => { + it("forwards the source verbatim", async () => { const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); + const shell = new CommandExecutor(f.rpc.shell, makeSync()); await shell.exec("echo hi && exit 0"); expect(f.calls.exec).toHaveLength(1); - expect(f.calls.exec[0].command).toBe("echo hi && exit 0"); + expect(f.calls.exec[0].source).toBe("echo hi && exit 0"); }); it("forwards an explicit id", async () => { const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - await shell.exec("noop", { id: "stable-id" }); + await new CommandExecutor(f.rpc.shell, makeSync()).exec("noop", { id: "stable-id" }); expect(f.calls.exec[0].id).toBe("stable-id"); }); it("omits id from the RPC when the caller doesn't supply one", async () => { const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - await shell.exec("noop"); + await new CommandExecutor(f.rpc.shell, makeSync()).exec("noop"); expect(f.calls.exec[0].id).toBeUndefined(); }); it("forwards cwd", async () => { const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - await shell.exec("noop", { cwd: "/workspace/sub" }); + await new CommandExecutor(f.rpc.shell, makeSync()).exec("noop", { cwd: "/workspace/sub" }); expect(f.calls.exec[0].cwd).toBe("/workspace/sub"); }); it("forwards timeoutMs", async () => { const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - await shell.exec("noop", { timeoutMs: 1000 }); + await new CommandExecutor(f.rpc.shell, makeSync()).exec("noop", { timeoutMs: 1000 }); expect(f.calls.exec[0].timeoutMs).toBe(1000); }); it("forwards timeoutMs: 0 to disable the timeout", async () => { const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - await shell.exec("noop", { timeoutMs: 0 }); + await new CommandExecutor(f.rpc.shell, makeSync()).exec("noop", { timeoutMs: 0 }); expect(f.calls.exec[0].timeoutMs).toBe(0); }); it("leaves timeoutMs undefined when the caller omits it", async () => { const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - await shell.exec("noop"); + await new CommandExecutor(f.rpc.shell, makeSync()).exec("noop"); expect(f.calls.exec[0].timeoutMs).toBeUndefined(); }); it("uses the id the runner returned, not the caller-supplied one", async () => { - // The runner is authoritative — if it mints an id, the handle - // exposes it. The facade doesn't second-guess. const f = fakeRpc({ mintedId: "from-runner" }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop"); - expect(handle.id).toBe("from-runner"); + const execution = await new CommandExecutor(f.rpc.shell, makeSync()).exec("noop"); + expect(execution.id).toBe("from-runner"); }); it("propagates errors from shell.exec; the pre-spawn push ran, the post-drain pull did not", async () => { @@ -240,81 +223,28 @@ describe("WorkspaceShell.exec — RPC forwarding", () => { return applied(0); }, }; - const shell = new WorkspaceShell(f.rpc.shell, sync); - await expect(shell.exec("noop")).rejects.toThrow("EEXEC_BUSY"); + await expect(new CommandExecutor(f.rpc.shell, sync).exec("noop")).rejects.toThrow("EEXEC_BUSY"); expect(pushCalls).toBe(1); expect(pullCalls).toBe(0); }); }); -// --------------------------------------------------------------------------- -// exec() — handle shape -// --------------------------------------------------------------------------- - -describe("WorkspaceShell.exec — handle shape", () => { - it("returns a ReadableStream that can be consumed with getReader()", async () => { - const f = fakeRpc({ events: [stdout(1, "hi"), exit(2, 0)] }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop"); - expect(handle).toBeInstanceOf(ReadableStream); - const reader = handle.getReader(); - const first = await reader.read(); - expect(first.done).toBe(false); - reader.releaseLock(); - }); - - it("returns a stream that supports for-await iteration", async () => { +describe("CommandExecutor.exec — envelope events", () => { + it("streams the raw events through in order", async () => { const f = fakeRpc({ events: [stdout(1, "a"), stdout(2, "b"), exit(3, 0)] }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop"); - const seen: string[] = []; - for await (const event of handle) seen.push(event.name); - expect(seen).toEqual(["stdout", "stdout", "exit"]); + const execution = await new CommandExecutor(f.rpc.shell, makeSync()).exec("noop"); + const { seen } = await drain(execution); + expect(seen.map((e) => e.name)).toEqual(["stdout", "stdout", "exit"]); }); - it("runs the post-command pull after stream-only consumption", async () => { - const f = fakeRpc({ events: [exit(1, 0)] }); - let pulls = 0; - const sync: Sync = { - push: async () => 0, - pull: async () => { - pulls += 1; - return applied(0); - }, - }; - const handle = await new WorkspaceShell(f.rpc.shell, sync).exec("noop"); - for await (const _event of handle) { - // Drain the stream. - } - expect(pulls).toBe(1); - }); - - it("hides id / result / kill from Object.keys and JSON.stringify", async () => { - const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop"); - expect(Object.keys(handle)).not.toContain("id"); - expect(Object.keys(handle)).not.toContain("result"); - expect(Object.keys(handle)).not.toContain("kill"); - // JSON.stringify on a stream returns "{}" (no enumerable own - // properties), confirming the extras aren't traversed. - expect(JSON.stringify(handle)).toBe("{}"); - }); - - it("kill(signal) forwards the signal to killExec", async () => { - const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop", { id: "kid" }); - await handle.kill("SIGKILL"); - expect(f.calls.killExec).toEqual([{ id: "kid", signal: "SIGKILL" }]); - }); - - it("kill() with no signal forwards undefined (server defaults to SIGTERM)", async () => { - const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop", { id: "kid" }); - await handle.kill(); - expect(f.calls.killExec).toEqual([{ id: "kid", signal: undefined }]); + it("carries stdout bytes untouched (no host-side encoding)", async () => { + const f = fakeRpc({ events: [stdout(1, "hi"), exit(2, 0)] }); + const execution = await new CommandExecutor(f.rpc.shell, makeSync()).exec("noop"); + const { seen } = await drain(execution); + const first = seen[0]; + expect(first.name).toBe("stdout"); + expect(first.value).toBeInstanceOf(Uint8Array); + expect(new TextDecoder().decode(first.value as Uint8Array)).toBe("hi"); }); it("disposes the RPC envelope when the event stream errors", async () => { @@ -339,8 +269,8 @@ describe("WorkspaceShell.exec — handle shape", () => { async killExec() {}, async disposeExec() {}, }; - const handle = await new WorkspaceShell(shellRpc, makeSync()).exec("noop"); - await expect(handle.getReader().read()).rejects.toThrow("transport failed"); + const execution = await new CommandExecutor(shellRpc, makeSync()).exec("noop"); + await expect(execution.events.getReader().read()).rejects.toThrow("transport failed"); expect(disposed).toBe(1); }); @@ -362,229 +292,14 @@ describe("WorkspaceShell.exec — handle shape", () => { async killExec() {}, async disposeExec() {}, }; - const handle = await new WorkspaceShell(shellRpc, makeSync()).exec("noop"); - await handle.cancel(); + const execution = await new CommandExecutor(shellRpc, makeSync()).exec("noop"); + await execution.events.cancel(); expect(disposed).toBe(1); }); - - it("kill() still reaches a non-retained backend when tail replay is absent", async () => { - let kills = 0; - const shellRpc: ShellRPC = { - async exec(input) { - return { id: input.id ?? "kid", events: new ReadableStream() }; - }, - async getExec() { - const error = new Error("no retained execution") as Error & { code: string }; - error.code = "ENOENT"; - throw error; - }, - async killExec() { - kills += 1; - }, - async disposeExec() {}, - }; - const handle = await new WorkspaceShell(shellRpc, makeSync()).exec("noop", { id: "kid" }); - await handle.kill(); - expect(kills).toBe(1); - }); - - it("kill() remains a no-op when the execution already completed", async () => { - const shellRpc: ShellRPC = { - async exec(input) { - return { id: input.id ?? "kid", events: new ReadableStream() }; - }, - async getExec() { - return { - id: "kid", - events: new ReadableStream({ - start(controller) { - controller.close(); - }, - }), - }; - }, - async killExec() {}, - async disposeExec() {}, - }; - const handle = await new WorkspaceShell(shellRpc, makeSync()).exec("noop", { id: "kid" }); - await expect(handle.kill()).resolves.toBeUndefined(); - }); -}); - -// --------------------------------------------------------------------------- -// exec() — result accumulation -// --------------------------------------------------------------------------- - -describe("WorkspaceShell.exec — result accumulation", () => { - it("concatenates stdout chunks in arrival order (Uint8Array default)", async () => { - const f = fakeRpc({ - events: [stdout(1, "one"), stdout(2, "two"), stdout(3, "three"), exit(4, 0)], - }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop"); - const result = await handle.result(); - expect(result.stdout).toBeInstanceOf(Uint8Array); - expect(new TextDecoder().decode(result.stdout as Uint8Array)).toBe("onetwothree"); - }); - - it("keeps stdout and stderr separate", async () => { - const f = fakeRpc({ - events: [stdout(1, "out"), stderr(2, "err"), stdout(3, "out2"), exit(4, 0)], - }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop"); - const result = await handle.result(); - expect(new TextDecoder().decode(result.stdout as Uint8Array)).toBe("outout2"); - expect(new TextDecoder().decode(result.stderr as Uint8Array)).toBe("err"); - }); - - it("captures the exit code from the exit event", async () => { - const f = fakeRpc({ events: [exit(1, 42)] }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop"); - const result = await handle.result(); - expect(result.exitCode).toBe(42); - }); - - it("returns exitCode = -1 when the stream closes without an exit event", async () => { - // The runner shouldn't do this, but if the wire drops mid-flight - // the facade must still resolve so callers see something. -1 is - // the documented sentinel. - const f = fakeRpc({ events: [stdout(1, "partial")] }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop"); - const result = await handle.result(); - expect(result.exitCode).toBe(-1); - }); - - it("returns an empty Uint8Array when no output arrives", async () => { - const f = fakeRpc({ events: [exit(1, 0)] }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop"); - const result = await handle.result(); - expect(result.stdout).toBeInstanceOf(Uint8Array); - expect((result.stdout as Uint8Array).byteLength).toBe(0); - expect((result.stderr as Uint8Array).byteLength).toBe(0); - }); - - it("returns an empty string for utf8 encoding with no output", async () => { - const f = fakeRpc({ events: [exit(1, 0)] }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop", { encoding: "utf8" }); - const result = await handle.result(); - expect(result.stdout).toBe(""); - expect(result.stderr).toBe(""); - }); - - it("rejects result() when the wire stream errors mid-flight", async () => { - const f = fakeRpc({ - events: [stdout(1, "partial")], - streamError: new Error("ESHUTDOWN"), - }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop"); - await expect(handle.result()).rejects.toThrow("ESHUTDOWN"); - }); }); -// --------------------------------------------------------------------------- -// exec() — utf8 encoding -// --------------------------------------------------------------------------- - -describe("WorkspaceShell.exec — utf8 encoding", () => { - it("returns stdout / stderr as strings when encoding is 'utf8'", async () => { - const f = fakeRpc({ - events: [stdout(1, "hello "), stderr(2, "warn"), stdout(3, "world"), exit(4, 0)], - }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop", { encoding: "utf8" }); - const result = await handle.result(); - expect(result.stdout).toBe("hello world"); - expect(result.stderr).toBe("warn"); - }); - - it("decodes multi-byte UTF-8 split across chunks correctly", async () => { - // "🎉" is F0 9F 8E 89. Split mid-character: first three bytes, - // then the trailing byte. A naive decoder per chunk produces - // replacement characters; the streaming decoder must hold the - // partial sequence and emit the full code point. - const partyHat = new Uint8Array([0xf0, 0x9f, 0x8e, 0x89]); - const head = partyHat.subarray(0, 3); - const tail = partyHat.subarray(3); - const f = fakeRpc({ - events: [ - { id: "_", seq: 1, name: "stdout", value: head }, - { id: "_", seq: 2, name: "stdout", value: tail }, - exit(3, 0), - ], - }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop", { encoding: "utf8" }); - const result = await handle.result(); - expect(result.stdout).toBe("🎉"); - }); - - it("keeps decoder flush sequence numbers strictly monotonic", async () => { - const f = fakeRpc({ - events: [ - { id: "_", seq: 1, name: "stdout", value: new Uint8Array([0xf0, 0x9f]) }, - exit(2, 0), - ], - }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop", { encoding: "utf8" }); - const seen: Array<{ seq: number; name: string; value: unknown }> = []; - for await (const event of handle) { - seen.push({ seq: event.seq, name: event.name, value: event.value }); - } - - expect(seen).toEqual([ - { seq: 1, name: "stdout", value: "" }, - { seq: 1.5, name: "stdout", value: "�" }, - { seq: 2, name: "exit", value: 0 }, - ]); - }); - - it("keeps the stdout and stderr decoders independent", async () => { - // Interleave a partial code point on each stream. If the - // decoders share state, one stream's tail would land on the - // other's head and corrupt both. - const partyHat = new Uint8Array([0xf0, 0x9f, 0x8e, 0x89]); // 🎉 - const heart = new Uint8Array([0xe2, 0x9d, 0xa4]); // ❤ - const f = fakeRpc({ - events: [ - { id: "_", seq: 1, name: "stdout", value: partyHat.subarray(0, 2) }, - { id: "_", seq: 2, name: "stderr", value: heart.subarray(0, 2) }, - { id: "_", seq: 3, name: "stdout", value: partyHat.subarray(2) }, - { id: "_", seq: 4, name: "stderr", value: heart.subarray(2) }, - exit(5, 0), - ], - }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop", { encoding: "utf8" }); - const result = await handle.result(); - expect(result.stdout).toBe("🎉"); - expect(result.stderr).toBe("❤"); - }); - - it("preserves encoding when consuming the stream directly", async () => { - const f = fakeRpc({ events: [stdout(1, "stream-mode"), exit(2, 0)] }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.exec("noop", { encoding: "utf8" }); - const seen: unknown[] = []; - for await (const event of handle) { - if (event.name === "stdout") seen.push(event.value); - } - expect(seen).toEqual(["stream-mode"]); - }); -}); - -// --------------------------------------------------------------------------- -// exec() — push/pull bracket math -// --------------------------------------------------------------------------- - -describe("WorkspaceShell.exec — push/pull bracket", () => { - it("reports pushed and pulled from the Sync calls", async () => { +describe("CommandExecutor.exec — push/pull bracket", () => { + it("reports pushed up front and the pull outcome after drain", async () => { const f = fakeRpc({ events: [stdout(1, "hi"), exit(2, 0)] }); const sync: Sync = { async push() { @@ -594,55 +309,41 @@ describe("WorkspaceShell.exec — push/pull bracket", () => { return applied(7); }, }; - const shell = new WorkspaceShell(f.rpc.shell, sync); - const handle = await shell.exec("noop"); - const result = await handle.result(); - expect(result.exitCode).toBe(0); - expect(result.pushed).toBe(5); - expect(result.pulled).toBe(7); - expect(result.skipped).toEqual([]); - expect(result.sync).toEqual({ status: "complete", applied: 7, skipped: [] }); - }); - - it("reports a clean no-op pull as complete", async () => { - const f = fakeRpc({ events: [exit(1, 0)] }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const result = await (await shell.exec("true")).result(); - expect(result.sync).toEqual({ status: "complete", applied: 0, skipped: [] }); + const execution = await new CommandExecutor(f.rpc.shell, sync).exec("noop"); + expect(execution.sync.pushed).toBe(5); + const { outcome } = await drain(execution); + expect(outcome).toEqual({ + applied: 7, + skipped: [], + sync: { status: "complete", applied: 7, skipped: [] }, + }); }); it("surfaces skipped read-only entries from the post-drain pull", async () => { const f = fakeRpc({ events: [exit(1, 0)] }); + const skipped = [ + { + path: "/workspace/r2/touched.txt", + mountRoot: "/workspace/r2", + op: "write" as const, + reason: "read-only" as const, + }, + ]; const sync: Sync = { async push() { return 0; }, async pull() { - return { - applied: 2, - skipped: [ - { - path: "/workspace/r2/touched.txt", - mountRoot: "/workspace/r2", - op: "write", - reason: "read-only", - }, - ], - }; + return { applied: 2, skipped }; }, }; - const shell = new WorkspaceShell(f.rpc.shell, sync); - const handle = await shell.exec("noop"); - const result = await handle.result(); - expect(result.pulled).toBe(2); - expect(result.skipped).toEqual([ - { - path: "/workspace/r2/touched.txt", - mountRoot: "/workspace/r2", - op: "write", - reason: "read-only", - }, - ]); + const execution = await new CommandExecutor(f.rpc.shell, sync).exec("noop"); + const { outcome } = await drain(execution); + expect(outcome).toEqual({ + applied: 2, + skipped, + sync: { status: "complete", applied: 2, skipped }, + }); }); it("calls push() before spawn and pull() after drain, in that order", async () => { @@ -658,10 +359,9 @@ describe("WorkspaceShell.exec — push/pull bracket", () => { return applied(0); }, }; - const shell = new WorkspaceShell(f.rpc.shell, sync); - const handle = await shell.exec("noop"); + const execution = await new CommandExecutor(f.rpc.shell, sync).exec("noop"); expect(order).toEqual(["push"]); // push fired before exec returned - await handle.result(); + await drain(execution); expect(order).toEqual(["push", "pull"]); // pull fired after drain }); @@ -675,19 +375,15 @@ describe("WorkspaceShell.exec — push/pull bracket", () => { return applied(3); }, }; - const shell = new WorkspaceShell(f.rpc.shell, sync); - const handle = await shell.exec("noop"); - const result = await handle.result(); - expect(result.exitCode).toBe(0); - expect(result.pushed).toBe(0); + const execution = await new CommandExecutor(f.rpc.shell, sync).exec("noop"); + expect(execution.sync.pushed).toBe(0); // pull still fires — docs/05 says one failed half doesn't abort the other - expect(result.pulled).toBe(3); + const { outcome } = await drain(execution); + expect((outcome as { applied: number }).applied).toBe(3); }); it("reports a pending sync after a Durable Object storage reset", async () => { - const f = fakeRpc({ - events: [stdout(1, "command output"), stderr(2, "command warning"), exit(3, 23)], - }); + const f = fakeRpc({ events: [exit(1, 0)] }); const reset = "Internal error in Durable Object storage write caused object to be reset."; const sync: Sync = { async push() { @@ -697,15 +393,10 @@ describe("WorkspaceShell.exec — push/pull bracket", () => { throw new Error(reset); }, }; - const shell = new WorkspaceShell(f.rpc.shell, sync); - const handle = await shell.exec("noop", { encoding: "utf8" }); - const result = await handle.result(); - expect(result).toMatchObject({ - exitCode: 23, - stdout: "command output", - stderr: "command warning", - pushed: 2, - pulled: 0, + const execution = await new CommandExecutor(f.rpc.shell, sync).exec("noop"); + const { outcome } = await drain(execution); + expect(outcome).toEqual({ + applied: 0, skipped: [], sync: { status: "pending", applied: 0, skipped: [], error: reset }, }); @@ -722,91 +413,68 @@ describe("WorkspaceShell.exec — push/pull bracket", () => { throw new Error(`transport failed token=${secret} ${"x".repeat(700)}`); }, }; - const shell = new WorkspaceShell(f.rpc.shell, sync); - const result = await (await shell.exec("noop")).result(); - expect(result.sync.status).toBe("pending"); - if (result.sync.status !== "pending") throw new Error("expected pending sync"); - expect(result.sync.error.length).toBeLessThanOrEqual(512); - expect(result.sync.error).toContain("transport failed token=[REDACTED]"); - expect(result.sync.error).not.toContain(secret); + const execution = await new CommandExecutor(f.rpc.shell, sync).exec("noop"); + const { outcome } = await drain(execution); + const settled = outcome as { sync: { status: string; error: string } }; + expect(settled.sync.status).toBe("pending"); + expect(settled.sync.error.length).toBeLessThanOrEqual(512); + expect(settled.sync.error).toContain("transport failed token=[REDACTED]"); + expect(settled.sync.error).not.toContain(secret); }); +}); - it("reports a pending sync after an ordinary transport error", async () => { - const f = fakeRpc({ events: [exit(1, 0)] }); - const sync: Sync = { - async push() { - return 2; - }, - async pull() { - throw new Error("WebSocket closed before pull completed"); - }, - }; - const shell = new WorkspaceShell(f.rpc.shell, sync); - const handle = await shell.exec("noop"); - const result = await handle.result(); - expect(result.pushed).toBe(2); - expect(result.pulled).toBe(0); - expect(result.skipped).toEqual([]); - expect(result.sync).toEqual({ - status: "pending", - applied: 0, - skipped: [], - error: "WebSocket closed before pull completed", - }); +describe("CommandExecutor.kill / dispose", () => { + it("kill(signal) forwards the signal to killExec", async () => { + const f = fakeRpc(); + await new CommandExecutor(f.rpc.shell, makeSync()).kill("kid", "SIGKILL"); + expect(f.calls.killExec).toEqual([{ id: "kid", signal: "SIGKILL" }]); }); -}); -// --------------------------------------------------------------------------- -// get() — reattach -// --------------------------------------------------------------------------- + it("kill() with no signal forwards undefined (server defaults to SIGTERM)", async () => { + const f = fakeRpc(); + await new CommandExecutor(f.rpc.shell, makeSync()).kill("kid"); + expect(f.calls.killExec).toEqual([{ id: "kid", signal: undefined }]); + }); +}); -describe("WorkspaceShell.get — reattach", () => { +describe("CommandExecutor.get — reattach", () => { it("forwards id to getExec", async () => { const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - await shell.get("attach-id"); + await new CommandExecutor(f.rpc.shell, makeSync()).get("attach-id"); expect(f.calls.getExec[0].id).toBe("attach-id"); }); it("maps resume: 'full' to after: undefined", async () => { const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - await shell.get("id", { resume: "full" }); + await new CommandExecutor(f.rpc.shell, makeSync()).get("id", { resume: "full" }); expect(f.calls.getExec[0].after).toBeUndefined(); }); it("maps resume: 'tail' to after: 'tail'", async () => { const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - await shell.get("id", { resume: "tail" }); + await new CommandExecutor(f.rpc.shell, makeSync()).get("id", { resume: "tail" }); expect(f.calls.getExec[0].after).toBe("tail"); }); it("maps resume: to after: ", async () => { const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - await shell.get("id", { resume: 17 }); + await new CommandExecutor(f.rpc.shell, makeSync()).get("id", { resume: 17 }); expect(f.calls.getExec[0].after).toBe(17); }); it("omits after when resume is not supplied", async () => { const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - await shell.get("id"); + await new CommandExecutor(f.rpc.shell, makeSync()).get("id"); expect(f.calls.getExec[0].after).toBeUndefined(); }); - it("returns a handle whose id matches the requested id", async () => { + it("returns an envelope whose id matches the requested id", async () => { const f = fakeRpc(); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.get("replay-me"); - expect(handle.id).toBe("replay-me"); + const execution = await new CommandExecutor(f.rpc.shell, makeSync()).get("replay-me"); + expect(execution.id).toBe("replay-me"); }); it("skips the pre-exec push but still runs the post-drain pull", async () => { - // Reattach doesn't own the original push frame: pushed = 0. - // The post-drain pull still fires — anything computerd produced - // between reattach and drain lands locally. const f = fakeRpc({ events: [exit(1, 0)] }); let pushCalls = 0; let pullCalls = 0; @@ -820,83 +488,11 @@ describe("WorkspaceShell.get — reattach", () => { return applied(2); }, }; - const shell = new WorkspaceShell(f.rpc.shell, sync); - const handle = await shell.get("x", { resume: "full" }); - const result = await handle.result(); + const execution = await new CommandExecutor(f.rpc.shell, sync).get("x", { resume: "full" }); + expect(execution.sync.pushed).toBe(0); + const { outcome } = await drain(execution); expect(pushCalls).toBe(0); expect(pullCalls).toBe(1); - expect(result.pushed).toBe(0); - expect(result.pulled).toBe(2); - }); - - it("accumulates the replayed output the same way exec() does", async () => { - const f = fakeRpc({ events: [stdout(1, "replay"), exit(2, 5)] }); - const shell = new WorkspaceShell(f.rpc.shell, makeSync()); - const handle = await shell.get("id", { encoding: "utf8" }); - const result = await handle.result(); - expect(result.stdout).toBe("replay"); - expect(result.exitCode).toBe(5); - }); - - it("reattaches to a live run once the first handle is dropped", async () => { - // The runner allows one live subscriber per run, so dropping a - // handle has to give up its subscription. The handle keeps a - // second reader on the event stream to watch for the exit event - // (kill() awaits it), and that reader has to go too — otherwise - // the subscription outlives the handle and reattach is refused - // for the rest of the run. - const f = liveRpc(); - const shell = new WorkspaceShell(f.shell, makeSync()); - const started = await shell.exec("sleep 100", { id: "long-run" }); - await started.cancel(); - - const again = await shell.get("long-run", { encoding: "utf8", resume: "tail" }); - f.emit(stdout(1, "still here\n")); - f.emit(exit(2, 0)); - const result = await again.result(); - expect(result.stdout).toBe("still here\n"); + expect((outcome as { applied: number }).applied).toBe(2); }); }); - -// A ShellRPC that models the runner's live subscriber bookkeeping: one -// subscriber per run, and the slot only frees when that subscriber -// cancels. Events are pushed by the test through emit(), so the run -// stays live for as long as the test wants it to. -function liveRpc(): { - shell: ShellRPC; - emit: (event: ExecEvent) => void; -} { - let subscriber: ReadableStreamDefaultController | undefined; - const subscribe = (id: string): ReadableStream => { - if (subscriber !== undefined) { - throw new Error(`EEXEC_BUSY: exec ${id} already has a live subscriber`); - } - return new ReadableStream({ - start(c) { - subscriber = c; - }, - cancel() { - subscriber = undefined; - }, - }); - }; - return { - // The runner closes the subscriber's stream after the exit - // event; result() drains until close. - emit: (event) => { - subscriber?.enqueue(event); - if (event.name === "exit") subscriber?.close(); - }, - shell: { - async exec(input) { - const id = input.id ?? "runner-minted-id"; - return { id, events: subscribe(id) }; - }, - async getExec(input) { - return { id: input.id, events: subscribe(input.id) }; - }, - async killExec() {}, - async disposeExec() {}, - }, - }; -} diff --git a/packages/computer/src/shell.ts b/packages/computer/src/shell.ts index a199281c..eb2a267c 100644 --- a/packages/computer/src/shell.ts +++ b/packages/computer/src/shell.ts @@ -1,30 +1,24 @@ -// Host-side WorkspaceShell facade. +// Host-side command executor. // -// Wraps the ShellRPC half of a WorkspaceRPC stub. The docs/05 -// contract: one entry point — exec() — that returns a detached -// handle. Callers either await `result()` (run-and-wait) or -// consume the ReadableStream directly (run-and-stream), or drop -// the handle entirely (fire-and-forget). +// Wraps the ShellRPC half of a WorkspaceRPC stub and presents it as +// the unified execution surface the Workspace runtime consumes: +// exec() spawns a command, get() reattaches to one, and both return +// an envelope of raw events plus the docs/05 sync bracket stats. // -// Every exec() call brackets the spawn with the docs/05 sync -// frames: -// - pushOnce(db, rpc.sync) runs *before* the spawn so any -// host-side writes since the last push are visible to the -// command. -// - pullOnce(db, rpc.sync) runs *after* the stream drains (i.e. -// after the exit event), so anything the command produced is -// visible to subsequent Workspace.fs reads. -// The pushed / pulled counts land in ExecResult. +// Every exec() call brackets the spawn with the docs/05 sync frames: +// - the pre-exec push ships any host-side writes since the last +// push so the spawned command sees them. +// - the post-drain pull runs after the event stream reaches its +// end, so anything the command produced is visible to subsequent +// Workspace.fs reads. +// The pushed count is known when the envelope is built; the pull +// outcome settles once the events stream drains. The runtime applies +// encoding and accumulates the result from the raw events. // -// Pull fires after either result() or direct stream consumption drains -// the execution events. The stream does not close until that pull settles. -// -// get() (reattach) is intentionally not bracketed. Reattaching -// to an already-running exec doesn't represent a new push frame. -// The result() of a reattached handle reports pushed = 0 and the -// pulled count from a pull that runs after its own drain — best- -// effort, can be 0 if nothing landed in computerd between reattach and -// drain. +// get() (reattach) is intentionally not push-bracketed. Reattaching +// to an already-running exec doesn't represent a new push frame, so +// it reports pushed = 0; the post-drain pull still fires, scoped to +// whatever landed between reattach and drain. import type { ExecEvent, ShellRPC } from "@cloudflare/computer-rpc"; import type { ApplyResult, SkippedEntry } from "@cloudflare/dofs"; @@ -47,59 +41,25 @@ export type ExecSyncResult = | { status: "complete"; applied: number; skipped: SkippedEntry[] } | { status: "pending"; applied: number; skipped: SkippedEntry[]; error: string }; -export interface ExecResult { - exitCode: number; - stdout: Chunk; - stderr: Chunk; - // VFS sync stats from the docs/05 bracket. - // pushed — entries shipped by the pre-exec pushOnce. - // pulled — entries the post-drain pullOnce applied locally. - // skipped — entries the post-drain pullOnce did NOT apply - // because they targeted a read-only mount root. - // Empty when no read-only mounts are registered or - // the container stayed clear of them. - // pushed is observed before the stream is returned. The remaining - // fields describe the post-command pull when result() is used. - pushed: number; - pulled: number; - skipped: SkippedEntry[]; - // Structured post-command sync outcome. The legacy pulled and - // skipped fields remain available for existing callers. - sync: ExecSyncResult; -} - export type KillSignal = "SIGTERM" | "SIGKILL" | "SIGINT" | "SIGHUP"; -// ExecHandle is a ReadableStream with three -// extras tacked on. Implemented as the wire stream + extra own -// properties (id / result / kill) rather than a subclass for two -// reasons: -// -// 1. The wire stream comes back from capnweb already built; -// subclassing means a pump-through layer that copies every -// chunk for no behavioural gain. -// 2. pipeThrough (used for the utf8 transform) returns a plain -// ReadableStream, so the subclass identity gets lost on the -// first transform anyway. -export interface ExecHandle - extends ReadableStream> { - readonly id: string; - result(): Promise>; - kill(signal?: KillSignal): Promise; - [Symbol.dispose](): void; +// The envelope both exec() and get() return: an execution id, the +// raw (unencoded) event stream, and the sync bracket stats. The +// pushed count is known up front; `outcome` settles when `events` +// reaches its end, carrying the post-drain pull result. +export interface CommandExecution { + id: string; + events: ReadableStream; + sync: { pushed: number; outcome: Promise }; } -export interface ExecOptions { +export interface ExecOptions { // Stable id. If omitted the runner mints a UUID. Reusing an id // while a previous run is still active throws EEXEC_BUSY. id?: string; // Absolute path inside the container. Defaults to the // workspace root. cwd?: string; - // Encoding for stdout/stderr value payloads. Default is - // Uint8Array; "utf8" decodes per-chunk through a stream-mode - // TextDecoder so multi-byte boundaries survive. - encoding?: E; // Per-call timeout in milliseconds. Past this duration the // container sends SIGTERM (then SIGKILL after a short grace). // Omit to use the runner's default (typically 320_000). Pass 0 @@ -112,37 +72,29 @@ export interface ExecOptions { // Standard input fed to the command. Bytes, or a string encoded // as UTF-8. stdin?: Uint8Array | string; - // Backend selector. Omit to use the default backend (the first - // one passed to the Workspace constructor); pass the id of - // another configured backend to route this call there. - backend?: string; } -export interface GetExecOptions { - encoding?: E; +export interface GetExecOptions { // "tail" yields only events produced after this call. A // number resumes from that seq+1. Omit to receive every // event from the start of the run (replays the whole log). resume?: "tail" | "full" | number; - // Backend selector. Same shape as ExecOptions.backend; routes - // the get / reattach to the named backend. - backend?: string; } -// Push/pull bracket plumbing. WorkspaceShell doesn't know about +// Push/pull bracket plumbing. CommandExecutor doesn't know about // the local Database or the SyncRPC wire — the host wires both // behind a Sync object that exposes the entry counts. // Workspace itself satisfies this interface (push() / pull() are // public methods); tests pass a plain { push, pull } object. -// pull() returns the dofs ApplyResult so the shell can surface -// skipped read-only entries on ExecResult. +// pull() returns the dofs ApplyResult so the executor can surface +// skipped read-only entries on the sync outcome. export interface Sync { push(): Promise; pull(): Promise; onPullPending?(error: unknown): Promise; } -export class WorkspaceShell { +export class CommandExecutor { readonly #shell: ShellRPC; readonly #sync: Sync; readonly #observer: WorkspaceObserver; @@ -153,17 +105,12 @@ export class WorkspaceShell { this.#observer = observer; } - exec(command: string): Promise>; - exec(command: string, options: ExecOptions): Promise>; - exec(command: string, options: ExecOptions<"utf8">): Promise>; - async exec( - command: string, - options: ExecOptions = {}, - ): Promise> { - assertNotTemplate(command); - // Pre-exec push: ship anything the host wrote since the last - // push so the spawned command sees it. Failures non-fatal per - // docs/05 — the command still runs; pushed reports 0. + // Spawn a command. Pushes host-side writes first so the command + // sees them, then returns the raw event stream and the sync + // bracket stats. The push failure is non-fatal per docs/05 — the + // command still runs and pushed reports 0. + async exec(source: string, options: ExecOptions = {}): Promise { + assertNotTemplate(source); let pushed = 0; try { pushed = await this.#sync.push(); @@ -180,7 +127,7 @@ export class WorkspaceShell { }, () => this.#shell.exec({ - command, + source, id: options.id, cwd: options.cwd, timeoutMs: options.timeoutMs, @@ -194,38 +141,33 @@ export class WorkspaceShell { if (outcome.ok) span.setAttribute("workspace.runtime.id", outcome.value.id); }, ); - // Dispose the result envelope when the event stream finishes + // Dispose the RPC envelope when the event stream finishes // draining. Without this, capnweb's exports table holds onto - // the envelope for the life of the session — one entry per - // exec call — because we hand the inner stream off to the - // caller and can't `using` the envelope ourselves. - const events = disposeOnDone(envelope.events, () => maybeDispose(envelope)); - return wrapHandle(this.#shell, this.#sync, envelope.id, events, options.encoding, pushed); + // the envelope for the life of the session — one entry per exec + // call — because the inner stream is handed off to the caller + // and the envelope can't be bound with `using` here. + const drained = disposeOnDone(envelope.events, () => maybeDispose(envelope)); + const { stream, outcome } = withPostPull(drained, this.#sync); + return { id: envelope.id, events: stream, sync: { pushed, outcome } }; } - get(id: string): Promise>; - get(id: string, options: GetExecOptions): Promise>; - get(id: string, options: GetExecOptions<"utf8">): Promise>; - async get( - id: string, - options: GetExecOptions = {}, - ): Promise> { + // Reattach to an in-flight or recently-completed exec. Reattach + // does not own the original push frame, so pushed = 0; the + // post-drain pull still fires, scoped to whatever landed between + // reattach and drain. + async get(id: string, options: GetExecOptions = {}): Promise { const after = resumeToAfter(options.resume); const envelope = await this.#shell.getExec({ id, after }); - const events = disposeOnDone(envelope.events, () => maybeDispose(envelope)); - // Reattach doesn't own the original push frame: pushed = 0. - // The post-drain pull still fires, scoped to whatever lands - // between reattach and the next drain. - return wrapHandle(this.#shell, this.#sync, id, events, options.encoding, 0); + const drained = disposeOnDone(envelope.events, () => maybeDispose(envelope)); + const { stream, outcome } = withPostPull(drained, this.#sync); + return { id, events: stream, sync: { pushed: 0, outcome } }; } - kill(id: string, signal?: KillSignal, _options: { backend?: string } = {}): Promise { + kill(id: string, signal?: KillSignal): Promise { return this.#shell.killExec({ id, signal }); } - dispose(id: string): Promise; - dispose(id: string, options: { backend?: string }): Promise; - dispose(id: string, _options: { backend?: string } = {}): Promise { + dispose(id: string): Promise { return this.#shell.disposeExec({ id }); } } @@ -236,155 +178,22 @@ function resumeToAfter(resume: "tail" | "full" | number | undefined): number | " return resume; } -// Stitch the runtime extras (id, result, kill) onto a fresh -// ReadableStream that pipes from the wire stream and applies any -// encoding conversion in flight. -// -// The user stream remains the only reader so backpressure reaches the backend. -// kill() requests a signal; result() or stream completion observes the exit. -function wrapHandle( - shell: ShellRPC, - sync: Sync, - id: string, - wireEvents: ReadableStream, - encoding: E | undefined, - pushed: number, -): ExecHandle { - const postPull = withPostPull(pipeEvents(wireEvents, encoding), sync); - const stream = postPull.stream; - const handle = stream as ExecHandle; - let resultPromise: Promise> | undefined; - let resultReader: ReadableStreamDefaultReader> | undefined; - // configurable: true on result/kill lets the Workspace-level - // router redefine them to add cross-cutting concerns (transport - // failure invalidation on result(); future kill hooks). The id - // slot stays non-configurable — nothing should rewrite it. - Object.defineProperties(handle, { - id: { value: id, enumerable: false, writable: false, configurable: false }, - result: { - value: () => { - resultPromise ??= drainToResult(stream, encoding, pushed, postPull.outcome, (reader) => { - resultReader = reader; - }); - return resultPromise; - }, - enumerable: false, - writable: false, - configurable: true, - }, - kill: { - value: (signal?: KillSignal) => shell.killExec({ id, signal }), - enumerable: false, - writable: false, - configurable: true, - }, - [Symbol.dispose]: { - value: () => { - if (resultReader) void resultReader.cancel().catch(() => undefined); - else void stream.cancel().catch(() => undefined); - }, - }, - }); - return handle; -} - -function pipeEvents( - source: ReadableStream, - encoding: E | undefined, -): ReadableStream> { - if (encoding !== "utf8") { - // Identity pipe — the wire shape already matches. - return source as unknown as ReadableStream>; - } - // Per-stream TextDecoders preserve multi-byte boundaries - // across chunk splits. - const stdoutDec = new TextDecoder("utf-8", { fatal: false }); - const stderrDec = new TextDecoder("utf-8", { fatal: false }); - let stdoutMeta: { id: string; seq: number } | undefined; - let stderrMeta: { id: string; seq: number } | undefined; - let lastSeq = 0; - const enqueue = ( - controller: TransformStreamDefaultController>, - event: WorkspaceExecEvent, - ) => { - lastSeq = event.seq; - controller.enqueue(event); - }; - const flushPending = ( - controller: TransformStreamDefaultController>, - beforeSeq?: number, - ) => { - const pending: Array<{ - id: string; - seq: number; - name: "stdout" | "stderr"; - value: Chunk; - }> = []; - const stdout = stdoutDec.decode(); - const stderr = stderrDec.decode(); - if (stdout && stdoutMeta) { - pending.push({ ...stdoutMeta, name: "stdout", value: stdout as Chunk }); - } - if (stderr && stderrMeta) { - pending.push({ ...stderrMeta, name: "stderr", value: stderr as Chunk }); - } - pending.sort((a, b) => a.seq - b.seq); - const span = beforeSeq !== undefined && beforeSeq > lastSeq ? beforeSeq - lastSeq : 1; - for (let index = 0; index < pending.length; index++) { - const event = pending[index]; - enqueue(controller, { - ...event, - seq: lastSeq + (span * (index + 1)) / (pending.length + 1), - }); - } - stdoutMeta = undefined; - stderrMeta = undefined; - }; - return source.pipeThrough( - new TransformStream>({ - transform(event, controller) { - if (event.name === "stdout") { - stdoutMeta = { id: event.id, seq: event.seq }; - enqueue(controller, { - id: event.id, - seq: event.seq, - name: "stdout", - value: stdoutDec.decode(event.value, { stream: true }) as Chunk, - }); - } else if (event.name === "stderr") { - stderrMeta = { id: event.id, seq: event.seq }; - enqueue(controller, { - id: event.id, - seq: event.seq, - name: "stderr", - value: stderrDec.decode(event.value, { stream: true }) as Chunk, - }); - } else { - flushPending(controller, event.seq); - enqueue(controller, event as WorkspaceExecEvent); - } - }, - flush: flushPending, - }), - ); -} - -interface PostPullOutcome { +export interface PostPullOutcome { applied: number; skipped: SkippedEntry[]; sync: ExecSyncResult; } -function withPostPull( - source: ReadableStream>, +export function withPostPull( + source: ReadableStream, sync: Sync, -): { stream: ReadableStream>; outcome: Promise } { +): { stream: ReadableStream; outcome: Promise } { const reader = source.getReader(); let resolveOutcome!: (outcome: PostPullOutcome) => void; const outcome = new Promise((resolve) => { resolveOutcome = resolve; }); - const stream = new ReadableStream>( + const stream = new ReadableStream( { async pull(controller) { try { @@ -447,66 +256,9 @@ async function runPostPull(sync: Sync): Promise { } } -async function drainToResult( - stream: ReadableStream>, - encoding: E | undefined, - pushed: number, - postPull: Promise, - setReader: (reader: ReadableStreamDefaultReader> | undefined) => void, -): Promise> { - const reader = stream.getReader(); - setReader(reader); - const stdoutParts: Array> = []; - const stderrParts: Array> = []; - let exitCode = -1; - try { - while (true) { - const { value, done } = await reader.read(); - if (done) break; - if (value.name === "stdout") stdoutParts.push(value.value); - else if (value.name === "stderr") stderrParts.push(value.value); - else exitCode = value.value; - } - } finally { - reader.releaseLock(); - setReader(undefined); - } - const pulled = await postPull; - return { - exitCode, - stdout: joinParts(stdoutParts, encoding), - stderr: joinParts(stderrParts, encoding), - pushed, - pulled: pulled.applied, - skipped: pulled.skipped, - sync: pulled.sync, - }; -} - -function joinParts( - parts: Array>, - encoding: E | undefined, -): Chunk { - if (parts.length === 0) { - return (encoding === "utf8" ? "" : new Uint8Array(0)) as Chunk; - } - if (typeof parts[0] === "string") { - return (parts as string[]).join("") as Chunk; - } - const arrays = parts as Uint8Array[]; - const total = arrays.reduce((acc, a) => acc + a.byteLength, 0); - const out = new Uint8Array(total); - let offset = 0; - for (const a of arrays) { - out.set(a, offset); - offset += a.byteLength; - } - return out as Chunk; -} - // Wrap `stream` so its capnweb envelope is released exactly once on clean // completion, source failure, or consumer cancellation. -function disposeOnDone(stream: ReadableStream, onDone: () => void): ReadableStream { +export function disposeOnDone(stream: ReadableStream, onDone: () => void): ReadableStream { const reader = stream.getReader(); let finished = false; const finish = () => { @@ -552,7 +304,7 @@ function disposeOnDone(stream: ReadableStream, onDone: () => void): Readab // Best-effort dispose of a capnweb result envelope. Real envelopes // expose [Symbol.dispose]; test fakes return plain objects, so the // symbol may be absent. -function maybeDispose(value: unknown): void { +export function maybeDispose(value: unknown): void { const d = (value as { [Symbol.dispose]?: () => void } | null | undefined)?.[Symbol.dispose]; if (typeof d === "function") d.call(value); } diff --git a/packages/computer/src/stub.test.ts b/packages/computer/src/stub.test.ts index 507acb28..8d7dad13 100644 --- a/packages/computer/src/stub.test.ts +++ b/packages/computer/src/stub.test.ts @@ -248,7 +248,7 @@ describe("WorkspaceStub", () => { id: "e-1", events: new ReadableStream({ start(c) { - c.enqueue({ id: "e-1", seq: 1, name: "exit", value: 0 }); + c.enqueue({ id: "e-1", seq: 1, name: "exit", code: 0 }); c.close(); }, }), @@ -279,7 +279,7 @@ describe("WorkspaceStub", () => { id: "other-id", events: new ReadableStream({ start(c) { - c.enqueue({ id: "other-id", seq: 1, name: "exit", value: 0 }); + c.enqueue({ id: "other-id", seq: 1, name: "exit", code: 0 }); c.close(); }, }), @@ -317,7 +317,7 @@ describe("WorkspaceStub", () => { id: `e-${execCalls}`, events: new ReadableStream({ start(c) { - c.enqueue({ id: `e-${execCalls}`, seq: 1, name: "exit", value: 0 }); + c.enqueue({ id: `e-${execCalls}`, seq: 1, name: "exit", code: 0 }); c.close(); }, }), @@ -486,7 +486,7 @@ describe("WorkspaceStub", () => { events: new ReadableStream({ start(c) { c.enqueue({ id: `e-${execCalls}`, seq: 1, name: "stdout", value: new Uint8Array() }); - c.enqueue({ id: `e-${execCalls}`, seq: 2, name: "exit", value: 0 }); + c.enqueue({ id: `e-${execCalls}`, seq: 2, name: "exit", code: 0 }); c.close(); }, }), @@ -524,7 +524,7 @@ describe("WorkspaceStub", () => { events: new ReadableStream({ start(c) { c.enqueue({ id: "e-1", seq: 1, name: "stdout", value: payload }); - c.enqueue({ id: "e-1", seq: 2, name: "exit", value: 0 }); + c.enqueue({ id: "e-1", seq: 2, name: "exit", code: 0 }); c.close(); }, }), @@ -538,19 +538,27 @@ describe("WorkspaceStub", () => { async (ws) => { const stub = ws.stub(); const handle = await stub.runtime.exec("noop"); - const events: Array<{ name: string; value: unknown }> = []; + type Collected = + | { name: "stdout" | "stderr"; value: Uint8Array } + | { name: "exit"; code: number }; + const events: Collected[] = []; const decoded = decodeRuntimeEvents(handle.stream()); const reader = decoded.getReader(); while (true) { const { value, done } = await reader.read(); if (done) break; - events.push({ name: value.name, value: value.value }); + events.push( + value.name === "exit" + ? { name: "exit", code: value.code } + : { name: value.name, value: value.value as Uint8Array }, + ); } reader.releaseLock(); expect(events).toHaveLength(2); - expect(events[0].name).toBe("stdout"); - expect(Array.from(events[0].value as Uint8Array)).toEqual(Array.from(payload)); - expect(events[1]).toEqual({ name: "exit", value: 0 }); + const first = events[0]; + if (first.name !== "stdout") throw new Error("expected stdout first"); + expect(Array.from(first.value)).toEqual(Array.from(payload)); + expect(events[1]).toEqual({ name: "exit", code: 0 }); }, { backend: backend({ shell: shellRpc }) }, ); @@ -574,7 +582,7 @@ describe("WorkspaceStub", () => { }); return; } - c.enqueue({ id: "e-1", seq: 10, name: "exit", value: 0 }); + c.enqueue({ id: "e-1", seq: 10, name: "exit", code: 0 }); c.close(); }, }), @@ -613,7 +621,7 @@ describe("WorkspaceStub", () => { id: "e-1", events: new ReadableStream({ start(c) { - c.enqueue({ id: "e-1", seq: 1, name: "exit", value: 0 }); + c.enqueue({ id: "e-1", seq: 1, name: "exit", code: 0 }); c.close(); }, }), diff --git a/packages/computer/src/tools/ai.test.ts b/packages/computer/src/tools/ai.test.ts index 9ea35cdb..bb60d70c 100644 --- a/packages/computer/src/tools/ai.test.ts +++ b/packages/computer/src/tools/ai.test.ts @@ -1,6 +1,6 @@ import { SQLiteTestStorage } from "@cloudflare/dofs/testing"; import { describe, expect, it } from "vitest"; -import type { ExecHandle, ExecResult } from "../shell.js"; +import type { WorkspaceRuntimeExecHandle, WorkspaceRuntimeResult } from "../runtime/types.js"; import { Workspace } from "../workspace.js"; import { createAITools, @@ -293,7 +293,7 @@ describe("createAITools exec tool", () => { runtime: { async exec(command: string, options: { cwd?: string; encoding: "utf8"; backend?: string }) { calls.push({ command, cwd: options.cwd, backend: options.backend }); - const result: ExecResult<"utf8"> = { + const result: WorkspaceRuntimeResult<"utf8"> = { exitCode: 2, stdout: "abcdef", stderr: "uvwxyz", @@ -301,7 +301,7 @@ describe("createAITools exec tool", () => { pulled: 0, skipped: [], }; - return { result: async () => result } as ExecHandle<"utf8">; + return { result: async () => result } as unknown as WorkspaceRuntimeExecHandle<"utf8">; }, }, }; @@ -334,7 +334,7 @@ describe("createAITools exec tool", () => { const workspace = { runtime: { async exec() { - const result: ExecResult<"utf8"> = { + const result: WorkspaceRuntimeResult<"utf8"> = { exitCode: 0, stdout: "a🙂b", stderr: "🙂🙂", @@ -342,7 +342,7 @@ describe("createAITools exec tool", () => { pulled: 0, skipped: [], }; - return { result: async () => result } as ExecHandle<"utf8">; + return { result: async () => result } as unknown as WorkspaceRuntimeExecHandle<"utf8">; }, }, }; @@ -367,7 +367,7 @@ describe("createAITools exec tool", () => { runtime: { async exec(command: string, options: { encoding: "utf8"; backend?: string }) { calls.push({ command, backend: options.backend }); - const result: ExecResult<"utf8"> = { + const result: WorkspaceRuntimeResult<"utf8"> = { exitCode: 0, stdout: "ok", stderr: "", @@ -375,7 +375,7 @@ describe("createAITools exec tool", () => { pulled: 0, skipped: [], }; - return { result: async () => result } as ExecHandle<"utf8">; + return { result: async () => result } as unknown as WorkspaceRuntimeExecHandle<"utf8">; }, }, }; diff --git a/packages/computer/src/workspace.test.ts b/packages/computer/src/workspace.test.ts index 467d7fee..95d3dc05 100644 --- a/packages/computer/src/workspace.test.ts +++ b/packages/computer/src/workspace.test.ts @@ -147,18 +147,18 @@ function makeBackend( // Backend with a wired shell.exec that pings a counter on every // call. exec resolves immediately with an empty event stream so -// the WorkspaceShell exec bracket settles right away. Used by the -// multi-backend selection tests. +// the command executor's exec bracket settles right away. Used by +// the multi-backend selection tests. function execBackend(id: string, onExec: (command: string) => void): WorkspaceBackend { const shell: import("@cloudflare/computer-rpc").ShellRPC = { async exec(input) { - onExec(input.command); + onExec(input.source); const execId = input.id ?? `${id}-${Math.random().toString(36).slice(2)}`; return { id: execId, events: new ReadableStream({ start(c) { - c.enqueue({ id: execId, seq: 1, name: "exit", value: 0 }); + c.enqueue({ id: execId, seq: 1, name: "exit", code: 0 }); c.close(); }, }), @@ -182,8 +182,9 @@ function execBackend(id: string, onExec: (command: string) => void): WorkspaceBa } // Drain an ExecHandle (or its result()-aware wrapper) to settle -// the WorkspaceShell push/pull bracket. The selection tests don't -// care about the values — only that exec ran on the right backend. +// the command executor's push/pull bracket. The selection tests +// don't care about the values — only that exec ran on the right +// backend. async function drainExec(handle: { result(): Promise }): Promise { await handle.result(); } @@ -261,7 +262,7 @@ describe("Workspace backend selection", () => { id, events: new ReadableStream({ start(controller) { - controller.enqueue({ id, seq: 1, name: "exit", value: 0 }); + controller.enqueue({ id, seq: 1, name: "exit", code: 0 }); controller.close(); }, }), @@ -293,7 +294,7 @@ describe("Workspace backend selection", () => { id, events: new ReadableStream({ start(controller) { - controller.enqueue({ id, seq: 1, name: "exit", value: 0 }); + controller.enqueue({ id, seq: 1, name: "exit", code: 0 }); controller.close(); }, }), @@ -329,7 +330,7 @@ describe("Workspace backend selection", () => { events: new ReadableStream({ start(controller) { controller.enqueue({ id, seq: 1, name: "stdout", value: new Uint8Array([0xe2]) }); - controller.enqueue({ id, seq: 2, name: "exit", value: 0 }); + controller.enqueue({ id, seq: 2, name: "exit", code: 0 }); controller.close(); }, }), @@ -374,7 +375,7 @@ describe("Workspace backend selection", () => { return { id, events: events() }; }, async killExec() { - exit = { id, seq: 1, name: "exit", value: 143 }; + exit = { id, seq: 1, name: "exit", code: 143 }; for (const controller of controllers) { controller.enqueue(exit); controller.close(); @@ -417,8 +418,7 @@ describe("Workspace backend selection", () => { name: "stdout", value: new Uint8Array([0xe2]), }); - controller.enqueue({ id: "module-exec", seq: 3, name: "result", value: 42 }); - controller.enqueue({ id: "module-exec", seq: 4, name: "exit", value: 0 }); + controller.enqueue({ id: "module-exec", seq: 3, name: "exit", code: 0, result: 42 }); controller.close(); }, }); @@ -474,7 +474,7 @@ describe("Workspace backend selection", () => { name: "stdout", value: new Uint8Array([0xf0, 0x9f]), }); - controller.enqueue({ id: "module-exec", seq: 2, name: "exit", value: 0 }); + controller.enqueue({ id: "module-exec", seq: 2, name: "exit", code: 0 }); controller.close(); }, }), @@ -491,15 +491,22 @@ describe("Workspace backend selection", () => { }; const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); const execution = await ws.runtime.exec("export default 42", { encoding: "utf8" }); - const seen: Array<{ seq: number; name: string; value: unknown }> = []; + type Collected = + | { seq: number; name: "stdout" | "stderr"; value: unknown } + | { seq: number; name: "exit"; code: number }; + const seen: Collected[] = []; for await (const event of execution) { - seen.push({ seq: event.seq, name: event.name, value: event.value }); + seen.push( + event.name === "exit" + ? { seq: event.seq, name: "exit", code: event.code } + : { seq: event.seq, name: event.name, value: event.value }, + ); } expect(seen).toEqual([ { seq: 1, name: "stdout", value: "" }, { seq: 1.5, name: "stdout", value: "�" }, - { seq: 2, name: "exit", value: 0 }, + { seq: 2, name: "exit", code: 0 }, ]); }); diff --git a/packages/computer/src/workspace.ts b/packages/computer/src/workspace.ts index e75ada0f..ac4b9e90 100644 --- a/packages/computer/src/workspace.ts +++ b/packages/computer/src/workspace.ts @@ -38,8 +38,9 @@ import { type WorkspaceModuleBackend, type WorkspaceModuleBackendHandle, type WorkspaceRegisteredBackend, + type WorkspaceRuntimeEvent, } from "./runtime/types.js"; -import { WorkspaceShell } from "./shell.js"; +import { CommandExecutor } from "./shell.js"; import { WorkspaceStub } from "./stub.js"; import { isWorkspaceTransportFailure } from "./transport-failure.js"; @@ -94,10 +95,6 @@ export interface WorkspaceOptions { // to Date.now. Override for deterministic tests. now?: () => number; - // Attach detached module execution to the Durable Object event lifetime. - // Pass `ctx.waitUntil.bind(ctx)` when using module backends in production. - waitUntil?: (promise: Promise) => void; - // Identifier for this workspace / session. Forwarded to mount // factories via MountContext.sessionId. Optional; defaults to "". sessionId?: string; @@ -216,10 +213,8 @@ export class Workspace { readonly #registeredBackendIds: Set; readonly #callableBackendIds: Set; readonly #defaultBackendId: string | undefined; - readonly #defaultCommandBackendId: string | undefined; readonly #observer: WorkspaceObserver; readonly #now: () => number; - readonly #waitUntil: ((promise: Promise) => void) | undefined; readonly #retryScheduler: SyncRetryScheduler | undefined; readonly #retryInitialDelayMs: number; readonly #retryMaxDelayMs: number; @@ -242,9 +237,13 @@ export class Workspace { // In-flight connect promises keyed by backend id, so concurrent // callers for the same backend share one connect pass. readonly #connecting = new Map>(); - // Per-backend WorkspaceShell facades. Constructed alongside each + // Per-backend CommandExecutor facades. Constructed alongside each // handle; reused for the life of the handle. - readonly #shells = new Map(); + readonly #shells = new Map(); + // Cached command adapters presenting a CommandExecutor as the + // unified backend handle. Cleared alongside #shells so an adapter + // never outlives the shell it wraps. + readonly #commandHandles = new Map(); readonly #moduleHandles = new Map(); readonly #connectingModuleHandles = new Map>(); #connectionGeneration = 0; @@ -270,7 +269,6 @@ export class Workspace { constructor(options: WorkspaceOptions) { this.#now = options.now ?? Date.now; - this.#waitUntil = options.waitUntil; this.#retryScheduler = options.retryScheduler; this.#retryInitialDelayMs = positiveRetryOption( options.retry?.initialDelayMs, @@ -301,13 +299,6 @@ export class Workspace { initializeSchema(this.#db, this.#now); this.#fs = new WorkspaceFilesystem(this.#db, { now: this.#now }); const registered = (options.backends ?? []).slice(); - if (registered.some((backend) => isModuleBackend(backend) && backend.requiresWaitUntil)) { - if (!options.waitUntil) { - throw new Error( - "Workspace module backend requires waitUntil; pass ctx.waitUntil.bind(ctx).", - ); - } - } this.#backends = registered.filter( (backend): backend is WorkspaceBackend => !isModuleBackend(backend), ); @@ -330,7 +321,6 @@ export class Workspace { this.#registeredBackendIds.add(backend.id); } this.#defaultBackendId = registered[0]?.id; - this.#defaultCommandBackendId = this.#backends[0]?.id; this.#observer = options.observer ?? noopObserver; this.#mounts = buildMountRegistry(options.mounts, { sessionId: options.sessionId, @@ -438,10 +428,8 @@ export class Workspace { get runtime(): WorkspaceRuntime { if (!this.#runtime) { this.#runtime = new WorkspaceRuntime({ - commandBackendIds: new Set(this.#backendsById.keys()), callableBackendIds: this.#callableBackendIds, - shell: () => this.#routedShell(), - moduleHandle: (id) => this.#moduleHandleFor(id), + backendHandle: (id) => this.#backendHandleFor(id), resolveBackendId: (id) => this.#resolveBackendId(id) ?? "", }); } @@ -537,7 +525,7 @@ export class Workspace { // push() ships everything the host has written since the last // push to that backend; pull() applies everything the backend // has produced since the last pull. Both are explicit — the - // package doesn't run a background loop. WorkspaceShell.exec + // package doesn't run a background loop. CommandExecutor.exec // brackets each call automatically against the backend it // selects; reach for push() / pull() directly only when an // FS-only flow needs the bracket without an exec. @@ -686,6 +674,7 @@ export class Workspace { if (this.#handles.get(id) !== handle) return false; this.#handles.delete(id); this.#shells.delete(id); + this.#commandHandles.delete(id); return true; } @@ -756,20 +745,6 @@ export class Workspace { return target; } - #resolveCommandBackendId(id: string | undefined): string | undefined { - const target = id ?? this.#defaultCommandBackendId; - if (target === undefined) return undefined; - if (!this.#registeredBackendIds.has(target)) { - throw new Error(`Workspace: no backend with id ${JSON.stringify(target)}`); - } - if (!this.#backendsById.has(target)) { - throw new Error( - `Workspace backend ${JSON.stringify(target)} does not accept shell commands.`, - ); - } - return target; - } - async close(): Promise { // Close every cached handle in parallel. Drop caches before // awaiting so a subsequent ready() / exec sees an empty slate @@ -779,6 +754,7 @@ export class Workspace { const moduleHandles = [...this.#moduleHandles.values()]; this.#handles.clear(); this.#shells.clear(); + this.#commandHandles.clear(); this.#connecting.clear(); this.#moduleHandles.clear(); this.#connectingModuleHandles.clear(); @@ -786,7 +762,7 @@ export class Workspace { await Promise.all( [...handles, ...moduleHandles].map(async (h) => { try { - await h.close(); + await h.close?.(); } catch { // close() is best-effort; a transport that's already // gone shouldn't take the workspace down with it. @@ -795,6 +771,87 @@ export class Workspace { ); } + // Unified backend handle used by the runtime. Module backends + // return their native handle; command backends are presented + // through the same interface by an adapter over their + // CommandExecutor, so the runtime has a single execution path. + async #backendHandleFor(id: string): Promise { + if (this.#moduleBackendsById.has(id)) return this.#moduleHandleFor(id); + return this.#commandHandleFor(id); + } + + // Command adapters are cached per backend so the module and + // command paths are symmetric. The cache is cleared alongside + // #shells whenever a handle is invalidated, so an adapter never + // outlives the shell it closed over. + async #commandHandleFor(id: string): Promise { + const cached = this.#commandHandles.get(id); + if (cached) return cached; + const { shell, handle } = await this.#shellFor(id); + const onError = (error: unknown) => this.#onShellError(id, handle, error); + const adapter: WorkspaceModuleBackendHandle = { + exec: async (input) => { + let envelope: Awaited>; + try { + envelope = await shell.exec(input.source, { + id: input.id, + cwd: input.cwd, + timeoutMs: input.timeoutMs, + env: input.env, + stdin: input.stdin, + }); + } catch (error) { + onError(error); + throw error; + } + return { + id: envelope.id, + events: watchStreamForTransportError( + envelope.events, + onError, + ) as ReadableStream, + sync: envelope.sync, + }; + }, + getExec: async ({ id: execId, after }) => { + const resume = after === undefined ? "full" : after; + let envelope: Awaited>; + try { + envelope = await shell.get(execId, { resume }); + } catch (error) { + onError(error); + throw error; + } + return { + id: envelope.id, + events: watchStreamForTransportError( + envelope.events, + onError, + ) as ReadableStream, + sync: envelope.sync, + }; + }, + killExec: async ({ id: execId, signal }) => { + try { + await shell.kill(execId, signal); + } catch (error) { + onError(error); + throw error; + } + }, + disposeExec: async ({ id: execId }) => { + try { + await shell.dispose(execId); + } catch (error) { + onError(error); + throw error; + } + }, + }; + this.#commandHandles.set(id, adapter); + return adapter; + } + #moduleHandleFor(id: string): Promise { const cached = this.#moduleHandles.get(id); if (cached) return Promise.resolve(cached); @@ -815,7 +872,6 @@ export class Workspace { () => backend.connect({ db: this.#db, - waitUntil: this.#waitUntil, fs: this.#fs, git: this.#gitFactory ? this.git : DISABLED_GIT_CLIENT, artifacts: this.#artifacts, @@ -823,7 +879,7 @@ export class Workspace { ) .then(async (handle) => { if (generation !== this.#connectionGeneration) { - await handle.close().catch(() => undefined); + await handle.close?.().catch(() => undefined); throw new Error(`Workspace closed while backend ${JSON.stringify(id)} was connecting.`); } this.#moduleHandles.set(id, handle); @@ -860,7 +916,6 @@ export class Workspace { () => backend.connect({ db: this.#db, - waitUntil: this.#waitUntil, fs: this.#fs, git: this.#gitFactory ? this.git : DISABLED_GIT_CLIENT, artifacts: this.#artifacts, @@ -899,6 +954,7 @@ export class Workspace { if (this.#handles.get(id) === handle) { this.#handles.delete(id); this.#shells.delete(id); + this.#commandHandles.delete(id); } }); } @@ -912,18 +968,18 @@ export class Workspace { return promise; } - // Per-backend WorkspaceShell, constructed on demand and cached + // Per-backend CommandExecutor, constructed on demand and cached // for the life of the handle. Returns both the shell and the // BackendHandle it was built against so the caller can hold the // handle reference for a later identity check; #invalidateHandle // clears both caches together, so a shell pulled from #shells is // always paired with the live handle for that id at the moment // of the lookup. - async #shellFor(id: string): Promise<{ shell: WorkspaceShell; handle: BackendHandle }> { + async #shellFor(id: string): Promise<{ shell: CommandExecutor; handle: BackendHandle }> { const handle = await this.#handleFor(id); const cached = this.#shells.get(id); if (cached !== undefined) return { shell: cached, handle }; - const shell = new WorkspaceShell( + const shell = new CommandExecutor( handle.rpc.shell, { push: () => this.push(id), @@ -936,19 +992,6 @@ export class Workspace { return { shell, handle }; } - // Routed shell facade. Each method picks the right backend per - // call (default, or the one named through ExecOptions.backend) - // and forwards to that backend's WorkspaceShell. - #routedShell(): WorkspaceShell { - const router = new WorkspaceShellRouter( - this.#defaultCommandBackendId ?? "", - (id) => this.#shellFor(id), - (id) => this.#resolveCommandBackendId(id) ?? "", - (id, handle, error) => this.#onShellError(id, handle, error), - ); - return router as unknown as WorkspaceShell; - } - // Invalidate the cached handle for `id` when a shell-routed RPC // fails with a known transport error. Compares the caller's // captured handle against the live cache entry so a late-failing @@ -960,6 +1003,36 @@ export class Workspace { } } +// Pass an execution event stream through unchanged, but classify any +// error that tears it down. A transport-classified mid-stream failure +// invalidates the cached backend handle so the next call reconnects, +// matching the invalidation the shell router used to install around a +// command handle's result(). +function watchStreamForTransportError( + events: ReadableStream, + onError: (error: unknown) => void, +): ReadableStream { + const reader = events.getReader(); + return new ReadableStream({ + async pull(controller) { + try { + const { value, done } = await reader.read(); + if (done) { + controller.close(); + return; + } + controller.enqueue(value); + } catch (error) { + onError(error); + controller.error(error); + } + }, + async cancel(reason) { + await reader.cancel(reason); + }, + }); +} + function positiveRetryOption(value: number | undefined, fallback: number, name: string): number { const resolved = value ?? fallback; if (!Number.isSafeInteger(resolved) || resolved <= 0) { @@ -989,140 +1062,6 @@ function createDisabledArtifactsClient(): ArtifactClient { } as ArtifactClient; } -// Selector wrapper that satisfies the WorkspaceShell surface but -// resolves the underlying ShellRPC per call. ExecOptions and -// GetExecOptions both gain an optional `backend` field; when -// present the router routes the call to that backend's -// WorkspaceShell, otherwise it routes to the default. -// -// Implemented as a non-extending class with the same method -// names so the routed object slots into every callsite that -// expects a WorkspaceShell without having to thread the union -// type through. -class WorkspaceShellRouter { - readonly #defaultId: string; - readonly #shellFor: (id: string) => Promise<{ shell: WorkspaceShell; handle: BackendHandle }>; - readonly #resolveId: (id: string | undefined) => string; - readonly #onError: (id: string, handle: BackendHandle, error: unknown) => void; - - constructor( - defaultId: string, - shellFor: (id: string) => Promise<{ shell: WorkspaceShell; handle: BackendHandle }>, - resolveId: (id: string | undefined) => string, - onError: (id: string, handle: BackendHandle, error: unknown) => void, - ) { - this.#defaultId = defaultId; - this.#shellFor = shellFor; - this.#resolveId = resolveId; - this.#onError = onError; - } - - async exec(command: string, options: { backend?: string } & Record = {}) { - const id = this.#resolveId(options.backend) || this.#defaultId; - // Capture the BackendHandle here, at dispatch time. A late - // failure from this command's stream must invalidate THIS - // handle, not whatever the cache holds when the rejection - // eventually fires; a concurrent reconnect may have already - // swapped in a newer handle that we must not clobber. - const { shell, handle: dispatchHandle } = await this.#shellFor(id); - const { backend: _backend, ...rest } = options; - let execHandle: unknown; - try { - execHandle = await (shell.exec as unknown as (c: string, o: typeof rest) => Promise)( - command, - rest, - ); - } catch (error) { - this.#onError(id, dispatchHandle, error); - throw error; - } - return this.#wrapHandle(id, dispatchHandle, execHandle); - } - - async get(id: string, options: { backend?: string } & Record = {}) { - const backendId = this.#resolveId(options.backend) || this.#defaultId; - const { shell, handle: dispatchHandle } = await this.#shellFor(backendId); - const { backend: _backend, ...rest } = options; - let execHandle: unknown; - try { - execHandle = await (shell.get as unknown as (e: string, o: typeof rest) => Promise)( - id, - rest, - ); - } catch (error) { - this.#onError(backendId, dispatchHandle, error); - throw error; - } - return this.#wrapHandle(backendId, dispatchHandle, execHandle); - } - - async kill( - id: string, - signal?: import("./shell.js").KillSignal, - options: { backend?: string } = {}, - ): Promise { - const backendId = this.#resolveId(options.backend) || this.#defaultId; - const { shell, handle } = await this.#shellFor(backendId); - try { - await shell.kill(id, signal); - } catch (error) { - this.#onError(backendId, handle, error); - throw error; - } - } - - async dispose(id: string, options: { backend?: string } = {}): Promise { - const backendId = this.#resolveId(options.backend) || this.#defaultId; - const { shell, handle } = await this.#shellFor(backendId); - try { - await shell.dispose(id); - } catch (error) { - this.#onError(backendId, handle, error); - throw error; - } - } - - // Wrap an ExecHandle so a transport-classified rejection from - // result() invalidates the cached backend handle. The dispatch- - // time catch above only fires when shell.exec()/get() rejects - // immediately; in practice a long-running command loses its - // transport mid-stream and the rejection surfaces through - // result() draining the event stream. - // - // The handle reference captured here is the one that was active - // when the exec was dispatched. By the time result() rejects, a - // concurrent reconnect may have replaced the cached entry for - // this id with a newer handle; invalidation in #onShellError - // checks identity against the dispatch-time handle so the newer - // entry survives. - // - // WorkspaceShell installs .result via defineProperty; we set it - // configurable: true so this slot can be redefined. The handle's - // stream identity is preserved — callers that consume the - // ReadableStream directly are unaffected; only result() routes - // through the invalidation path. - #wrapHandle(id: string, dispatchHandle: BackendHandle, execHandle: unknown): unknown { - const original = execHandle as { result?: unknown }; - if (typeof original.result !== "function") return execHandle; - const onError = this.#onError; - const originalResult = original.result.bind(execHandle) as () => Promise; - Object.defineProperty(execHandle, "result", { - value: async () => { - try { - return await originalResult(); - } catch (error) { - onError(id, dispatchHandle, error); - throw error; - } - }, - enumerable: false, - writable: false, - configurable: true, - }); - return execHandle; - } -} - export function createThinkCompatibility( fs: ThinkWorkspaceFilesystem, ): ThinkWorkspaceCompatibility { diff --git a/packages/computerd/src/exec/log.ts b/packages/computerd/src/exec/log.ts index 7583f6a5..6a722898 100644 --- a/packages/computerd/src/exec/log.ts +++ b/packages/computerd/src/exec/log.ts @@ -212,7 +212,7 @@ function materialise(id: string, row: EventRow): ExecEvent { // node:sqlite returns BLOB as Uint8Array; we need the bytes // view to read the int32. const view = new DataView(row.value.buffer, row.value.byteOffset, row.value.byteLength); - return { id, seq: row.seq, name: "exit", value: view.getInt32(0, true) }; + return { id, seq: row.seq, name: "exit", code: view.getInt32(0, true) }; } const name = row.kind === KIND_STDOUT ? "stdout" : "stderr"; return { id, seq: row.seq, name, value: row.value }; diff --git a/packages/computerd/src/exec/runner.fuse.test.ts b/packages/computerd/src/exec/runner.fuse.test.ts index 9a7fa5b5..da5d2e8c 100644 --- a/packages/computerd/src/exec/runner.fuse.test.ts +++ b/packages/computerd/src/exec/runner.fuse.test.ts @@ -119,7 +119,7 @@ describeIfReal("Runner shell.exec under real FUSE", () => { const result = await Promise.race([ client.shell .exec({ - command: "echo hello && pwd", + source: "echo hello && pwd", cwd: "/workspace", timeoutMs: 5_000, }) @@ -149,7 +149,7 @@ describeIfReal("Runner shell.exec under real FUSE", () => { .join(""); const exit = events.find((e) => e.name === "exit"); expect(stdout).toBe("hello\n/workspace\n"); - expect(exit?.value).toBe(0); + expect(exit?.code).toBe(0); } finally { await client.close(); } diff --git a/packages/computerd/src/exec/runner.test.ts b/packages/computerd/src/exec/runner.test.ts index 97be8ec9..4258017b 100644 --- a/packages/computerd/src/exec/runner.test.ts +++ b/packages/computerd/src/exec/runner.test.ts @@ -7,7 +7,7 @@ import { Runner } from "./runner.js"; type ExecEvent = | { id: string; seq: number; name: "stdout"; value: Uint8Array } | { id: string; seq: number; name: "stderr"; value: Uint8Array } - | { id: string; seq: number; name: "exit"; value: number }; + | { id: string; seq: number; name: "exit"; code: number }; function fixture(options: Record = {}): { runner: InstanceType; @@ -66,7 +66,7 @@ test("exec captures stdout and propagates exit code", async () => { const exit = events.find((e) => e.name === "exit"); expect(stdout).toBe("hello\n"); expect(stderr).toBe("world\n"); - expect(exit?.value).toBe(3); + expect(exit?.code).toBe(3); // seq is monotonic per-id starting at 1. const seqs = events.map((e) => e.seq); for (let i = 1; i < seqs.length; i++) { @@ -114,7 +114,7 @@ test("feeds per-execution stdin to the child and closes it", async () => { .join(""); const exit = events.find((event) => event.name === "exit"); expect(stdout).toBe("piped-input"); - expect(exit?.value).toBe(0); + expect(exit?.code).toBe(0); } finally { dispose(); } @@ -180,7 +180,7 @@ test("kill() terminates a running exec", async () => { const exit = events.find((e) => e.name === "exit"); expect(exit !== undefined).toBeTruthy(); // SIGTERM → 143 per the mapping in runner.ts. - expect(exit?.value).toBe(143); + expect(exit?.code).toBe(143); } finally { dispose(); } @@ -196,7 +196,7 @@ test("exec times out at timeoutMs and exits 143", async () => { const exit = events.find((e) => e.name === "exit"); expect(exit !== undefined).toBeTruthy(); // SIGTERM → 143 per mapExitCode. - expect(exit?.value).toBe(143); + expect(exit?.code).toBe(143); } finally { dispose(); } @@ -209,7 +209,7 @@ test("exec uses defaultTimeoutMs from the runner when no per-call value", async const events = await drain(handle.events); const exit = events.find((e) => e.name === "exit"); expect(exit !== undefined).toBeTruthy(); - expect(exit?.value).toBe(143); + expect(exit?.code).toBe(143); } finally { dispose(); } @@ -235,7 +235,7 @@ test("timeoutMs: 0 disables the timeout", async () => { const handle = runner.exec("echo hi", { id: "noto", timeoutMs: 0 }); const events = await drain(handle.events); const exit = events.find((e) => e.name === "exit"); - expect(exit?.value).toBe(0); + expect(exit?.code).toBe(0); } finally { dispose(); } @@ -361,7 +361,7 @@ test("exec(cwd) does not pass cwd to spawn; threads it through the shell", async .join(""); const exit = events.find((e) => e.name === "exit"); expect(stdout.trim()).toBe("/tmp"); - expect(exit?.value).toBe(0); + expect(exit?.code).toBe(0); } finally { dispose(); } @@ -384,7 +384,7 @@ test("exec(cwd) with a missing path synthesises a spawn-failed shape", async () const exit = events.find((e) => e.name === "exit"); expect(stderr).toMatch(/^spawn failed: /); expect(stderr).toMatch(/no such path|ENOENT/i); - expect(exit?.value).toBe(-1); + expect(exit?.code).toBe(-1); } finally { dispose(); } @@ -410,7 +410,7 @@ test("exec(cwd) quotes path segments with spaces and single quotes", async () => .join(""); const exit = events.find((e) => e.name === "exit"); expect(stdout.trim()).toBe(tricky); - expect(exit?.value).toBe(0); + expect(exit?.code).toBe(0); } finally { dispose(); } diff --git a/packages/computerd/src/exec/runner.ts b/packages/computerd/src/exec/runner.ts index 76ff977e..347fd227 100644 --- a/packages/computerd/src/exec/runner.ts +++ b/packages/computerd/src/exec/runner.ts @@ -230,7 +230,7 @@ export class Runner { this.scheduleSweep(); return; } - record.subscriber?.enqueue({ id, seq, name: "exit", value: exitCode }); + record.subscriber?.enqueue({ id, seq, name: "exit", code: exitCode }); record.subscriber?.close(); record.subscriber = undefined; this.scheduleSweep(); @@ -466,7 +466,7 @@ export class Runner { const events = new ReadableStream({ start(controller) { controller.enqueue({ id, seq: stderrSeq, name: "stderr", value }); - controller.enqueue({ id, seq: exitSeq, name: "exit", value: -1 }); + controller.enqueue({ id, seq: exitSeq, name: "exit", code: -1 }); controller.close(); }, }); diff --git a/packages/computerd/src/exec/types.ts b/packages/computerd/src/exec/types.ts index 647ee0a7..68fc5cbf 100644 --- a/packages/computerd/src/exec/types.ts +++ b/packages/computerd/src/exec/types.ts @@ -20,7 +20,7 @@ export type HeartbeatValue = { export type ExecEvent = | { id: string; seq: number; name: "stdout"; value: Uint8Array } | { id: string; seq: number; name: "stderr"; value: Uint8Array } - | { id: string; seq: number; name: "exit"; value: number } + | { id: string; seq: number; name: "exit"; code: number } | { id: string; seq: number; name: "heartbeat"; value: HeartbeatValue }; export interface ExecOptions { diff --git a/packages/rpc/src/interface.ts b/packages/rpc/src/interface.ts index 7eb5bca0..e12c3293 100644 --- a/packages/rpc/src/interface.ts +++ b/packages/rpc/src/interface.ts @@ -96,9 +96,13 @@ export interface ShellRPC { // Capnweb streams handle backpressure end-to-end; consumer-side // slowness propagates to the spawned process via the kernel pipe. exec(input: { - command: string; + source: string; cwd?: string; id?: string; + // Structured value handed to a callable backend. Command + // backends ignore it; the caller only sends it to backends that + // declare themselves callable. + input?: unknown; // Per-call timeout in milliseconds. Past this duration the // container sends SIGTERM (then SIGKILL after a short grace). // 0 disables the timeout. Omit to use the runner's default @@ -146,7 +150,7 @@ export interface WorkspaceRPC { export type ExecEvent = | { id: string; seq: number; name: "stdout"; value: Uint8Array } | { id: string; seq: number; name: "stderr"; value: Uint8Array } - | { id: string; seq: number; name: "exit"; value: number }; + | { id: string; seq: number; name: "exit"; code: number; result?: unknown }; // Error codes carried over the wire. The client adapter rethrows as // WorkspaceError preserving `code`, so application code can branch diff --git a/packages/rpc/src/server.ts b/packages/rpc/src/server.ts index b07915e3..9d0f23b8 100644 --- a/packages/rpc/src/server.ts +++ b/packages/rpc/src/server.ts @@ -237,9 +237,10 @@ class ShellRPCServer extends RpcTarget implements ShellRPC { } async exec(input: { - command: string; + source: string; cwd?: string; id?: string; + input?: unknown; timeoutMs?: number; env?: Record; stdin?: Uint8Array; @@ -247,7 +248,7 @@ class ShellRPCServer extends RpcTarget implements ShellRPC { id: string; events: ReadableStream; }> { - return this.runner.exec(input.command, { + return this.runner.exec(input.source, { id: input.id, cwd: input.cwd, timeoutMs: input.timeoutMs, diff --git a/packages/rpc/tests/shell-and-composite.test.ts b/packages/rpc/tests/shell-and-composite.test.ts index bc964df5..002c8e5e 100644 --- a/packages/rpc/tests/shell-and-composite.test.ts +++ b/packages/rpc/tests/shell-and-composite.test.ts @@ -75,7 +75,7 @@ function makeFakeRunner(): FakeRunner { name: "stdout", value: new TextEncoder().encode(`ran:${command}\n`), }, - { id, seq: 2, name: "exit", value: 0 }, + { id, seq: 2, name: "exit", code: 0 }, ], }; records.set(id, rec); @@ -158,12 +158,12 @@ describe("ShellRPC over a real WebSocket", () => { // `client.exec(...)` lands on it directly. createWorkspaceClient // proxies all property access through; capnweb routes by name. // biome-ignore lint/suspicious/noExplicitAny: client targets ShellRPC for this harness, not WorkspaceRPC - const handle = await (client as any).exec({ command: "echo hi" }); + const handle = await (client as any).exec({ source: "echo hi" }); const events = await drainExec(handle.events); expect(events).toHaveLength(2); expect(events[0]?.name).toBe("stdout"); expect(new TextDecoder().decode(events[0]?.value as Uint8Array)).toBe("ran:echo hi\n"); - expect(events[1]).toMatchObject({ name: "exit", value: 0 }); + expect(events[1]).toMatchObject({ name: "exit", code: 0 }); // Server-side: the runner saw the call. const rec = harness.runner.records.get(handle.id); @@ -178,7 +178,7 @@ describe("ShellRPC over a real WebSocket", () => { const client = createWorkspaceClient({ url: harness.url }); try { // biome-ignore lint/suspicious/noExplicitAny: see above - const handle = await (client as any).exec({ command: "sleep", id: "fixed" }); + const handle = await (client as any).exec({ source: "sleep", id: "fixed" }); await drainExec(handle.events); // biome-ignore lint/suspicious/noExplicitAny: see above @@ -198,7 +198,7 @@ describe("ShellRPC over a real WebSocket", () => { const client = createWorkspaceClient({ url: harness.url }); try { // biome-ignore lint/suspicious/noExplicitAny: see above - const first = await (client as any).exec({ command: "first", id: "repeat" }); + const first = await (client as any).exec({ source: "first", id: "repeat" }); await drainExec(first.events); // biome-ignore lint/suspicious/noExplicitAny: see above @@ -266,10 +266,10 @@ describe("Composite WorkspaceRPC (sync + shell on one session)", () => { expect(typeof wm.currentRev).toBe("number"); expect(wm.currentRev).toBeGreaterThanOrEqual(0); - const handle = await client.shell.exec({ command: "ls" }); + const handle = await client.shell.exec({ source: "ls" }); const events = await drainExec(handle.events); expect(events).toHaveLength(2); - expect(events[1]).toMatchObject({ name: "exit", value: 0 }); + expect(events[1]).toMatchObject({ name: "exit", code: 0 }); // Sanity check: server side records both interactions. expect(harness.runner.records.get(handle.id)?.command).toBe("ls"); @@ -287,7 +287,7 @@ describe("Composite WorkspaceRPC (sync + shell on one session)", () => { try { expect(await client.sync.readEntry("/none")).toBeNull(); - const handle = await client.shell.exec({ command: "noop" }); + const handle = await client.shell.exec({ source: "noop" }); await drainExec(handle.events); expect(harness.runner.records.size).toBe(1); @@ -304,7 +304,7 @@ describe("Composite WorkspaceRPC (sync + shell on one session)", () => { harness = await startCompositeHarness(); const client = createWorkspaceClient({ url: harness.url }); try { - const handle = await client.shell.exec({ command: "x" }); + const handle = await client.shell.exec({ source: "x" }); await drainExec(handle.events); const wm = await client.sync.watermarks(); expect(typeof wm.currentRev).toBe("number"); diff --git a/script/computerd-stub-soak.mjs b/script/computerd-stub-soak.mjs index 54b640fc..2bfedc34 100755 --- a/script/computerd-stub-soak.mjs +++ b/script/computerd-stub-soak.mjs @@ -283,7 +283,7 @@ async function main() { // pattern as fetchChanges. console.error(`[soak] ${EXEC_CALLS} exec calls (no disposal)…`); for (let i = 0; i < EXEC_CALLS; i++) { - const result = await stub.shell.exec({ command: "true" }); + const result = await stub.shell.exec({ source: "true" }); const reader = result.events.getReader(); try { while (true) { @@ -324,7 +324,7 @@ async function main() { console.error(`[soak] ${EXEC_CALLS} exec calls (with disposal)…`); for (let i = 0; i < EXEC_CALLS; i++) { - const result = await stub.shell.exec({ command: "true" }); + const result = await stub.shell.exec({ source: "true" }); const reader = result.events.getReader(); try { while (true) { diff --git a/script/exec-tests b/script/exec-tests index 24533a02..05217693 100755 --- a/script/exec-tests +++ b/script/exec-tests @@ -80,7 +80,7 @@ const assert = (cond, msg) => { const client = createWorkspaceClient({ url: `ws://localhost:${port}/ws` }); try { // 1. echo + exit code. - const h = await client.shell.exec({ command: "echo hello && exit 7" }); + const h = await client.shell.exec({ source: "echo hello && exit 7" }); const dec = new TextDecoder(); const events = []; const reader = h.events.getReader(); @@ -93,17 +93,17 @@ try { .map(e => dec.decode(e.value)).join(""); const exit = events.find(e => e.name === "exit"); assert(stdout === "hello\n", `unexpected stdout: ${JSON.stringify(stdout)}`); - assert(exit && exit.value === 7, `unexpected exit: ${JSON.stringify(exit)}`); + assert(exit && exit.code === 7, `unexpected exit: ${JSON.stringify(exit)}`); console.log(" PASS exec captures stdout + exit code"); // 1b. EEXEC_BUSY surfaces with the right code over the wire. // The runner refuses a second exec on a live id; the host // catches a plain Error with err.code intact (capnweb copies // own enumerable props on Error subclasses). - const live = await client.shell.exec({ command: "sleep 30", id: "busy" }); + const live = await client.shell.exec({ source: "sleep 30", id: "busy" }); let busyErr; try { - await client.shell.exec({ command: "echo nope", id: "busy" }); + await client.shell.exec({ source: "echo nope", id: "busy" }); } catch (err) { busyErr = err; } @@ -120,14 +120,14 @@ try { console.log(" PASS EEXEC_BUSY propagates with err.code"); // 2. kill a long-running process. - const long = await client.shell.exec({ command: "sleep 30", id: "killme" }); + const long = await client.shell.exec({ source: "sleep 30", id: "killme" }); await client.shell.killExec({ id: "killme", signal: "SIGTERM" }); const longReader = long.events.getReader(); let killExit; while (true) { const { value, done } = await longReader.read(); if (done) break; - if (value.name === "exit") killExit = value.value; + if (value.name === "exit") killExit = value.code; } // SIGTERM → 143 per Runner mapping. assert(killExit === 143, `kill exit code was ${killExit}`); @@ -161,7 +161,7 @@ try { // overflowed the cap raises ELOG_TRUNCATED. EXEC_LOG_MAX_BYTES=256 // on the container makes the cap tiny so even an echo trips it. const big = await client.shell.exec({ - command: "head -c 2048 /dev/urandom | base64", + source: "head -c 2048 /dev/urandom | base64", id: "big", }); // Drain the live stream first — eviction does not gate live. @@ -214,7 +214,7 @@ try { // flow-control window stalls for long — a real test of pipe // pause needs a capnweb upgrade or a different observable. const chatty = await client.shell.exec({ - command: "head -c 8192 /dev/urandom | base64", + source: "head -c 8192 /dev/urandom | base64", id: "chatty", }); let totalBytes = 0;