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 217f0c69..8ea7a673 100644 --- a/packages/computer/README.md +++ b/packages/computer/README.md @@ -34,11 +34,11 @@ Three backends ship today on tree-shakeable subpaths: `node:fs/promises`, and trusted `ws:git` / `ws:artifacts` modules. See [`docs/17_isolate_javascript.md`](../../docs/17_isolate_javascript.md). -The worker-JavaScript backend runs after `runtime.exec()` returns. Pass -`waitUntil: ctx.waitUntil.bind(ctx)` to `Workspace` so completion remains -attached to the Durable Object event. The backend refuses to connect without -this lifecycle hook. It admits one execution at a time by default and bounds -completed execution retention by time and count. +The worker-JavaScript backend runs after `runtime.exec()` returns. The run +keeps advancing while its event stream is consumed, which holds the Durable +Object resident; a handle that is never read can be evicted once the object +goes idle. It admits one execution at a time by default and bounds completed +execution retention by time and count. A backend can declare `sync: "none"` on the handle it returns to opt out of the push/pull bracket entirely — the worker backend 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..2bfd22f0 100644 --- a/packages/computer/src/backends/worker-javascript/frames.test.ts +++ b/packages/computer/src/backends/worker-javascript/frames.test.ts @@ -36,16 +36,21 @@ 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","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","value":0,"result":{"a":[1,2,null]}}`); + expect(frame).toEqual({ name: "exit", value: 0, result: { a: [1, 2, null] } }); + }); + + it("decodes an exit frame carrying a null result", () => { + const frame = parseRuntimeFrame(`{"name":"exit","value":0,"result":null}`); + expect(frame).toEqual({ name: "exit", value: 0, result: null }); + }); + it("rejects invalid JSON", () => { expect(() => parseRuntimeFrame("not json")).toThrow(); }); diff --git a/packages/computer/src/backends/worker-javascript/frames.ts b/packages/computer/src/backends/worker-javascript/frames.ts index d924a6f5..4821b875 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"; value: number; result?: WorkspaceRuntimeValue }; export function parseRuntimeFrame(line: string): RuntimeFrame { let record: Record; @@ -20,13 +19,17 @@ 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)) { throw new Error("WorkerJavaScriptBackend received a malformed exit frame"); } + if ("result" in record) { + return { + name, + value: record.value as number, + result: record.result as WorkspaceRuntimeValue, + }; + } return { name, value: record.value 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..a2d1b7ec 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,8 +26,7 @@ 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", value: 0, result: value })); } catch (error) { const message = error instanceof Error ? error.message : String(error); frames.push(JSON.stringify({ name: "stderr", b64: btoa(`${message}\n`) })); @@ -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" }); diff --git a/packages/computer/src/backends/worker-javascript/worker-javascript.ts b/packages/computer/src/backends/worker-javascript/worker-javascript.ts index d0bdf43a..2005fd76 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); } } @@ -415,7 +409,6 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { onComplete: () => this.#finalize(record), onError: (message) => this.#finalize(record, message), }); - this.#host.waitUntil?.(record.control.completion); } catch (error) { record.control?.cancel(); await record.control?.completion.catch(() => undefined); @@ -612,11 +605,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; + if ("result" in frame) { + record.result = frame.result; + record.hasResult = true; + } } } @@ -676,22 +670,17 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { 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", + value: exitCode, + result: record.result as WorkspaceRuntimeValue, + } + : { id: record.id, seq: record.events.length + 1, name: "exit", value: exitCode }; + this.#finish(record, exitCode === 0 ? "completed" : "failed", [exit]); } #stream(record: ExecutionRecord, after?: number | "tail") { @@ -889,7 +878,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 ? { value: event.value, result: event.result } : { value: event.value }; + return new TextEncoder().encode(JSON.stringify(payload)); } function decodeEvent( @@ -899,10 +890,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 && "value" in decoded) { + const record = decoded as { value: unknown; result?: unknown }; + if ("result" in record) { + assertRuntimeValue(record.result); + return { id, seq, name: "exit", value: Number(record.value), result: record.result }; + } + return { id, seq, name: "exit", value: Number(record.value) }; + } + return { id, seq, name: "exit", value: Number(decoded) }; } function startJavaScriptExecution(options: { @@ -1158,8 +1155,7 @@ 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", value: 0, result: value }); } catch (error) { const message = error instanceof Error ? error.message : String(error); enqueue({ name: "stderr", b64: toBase64(truncate(message) + "\\n") }); 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..755154d1 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) { @@ -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..a3ea33a5 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, diff --git a/packages/computer/src/observe-integration.test.ts b/packages/computer/src/observe-integration.test.ts index 5e6a5ab0..83d41281 100644 --- a/packages/computer/src/observe-integration.test.ts +++ b/packages/computer/src/observe-integration.test.ts @@ -242,7 +242,7 @@ 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({ 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/runtime/runtime.test.ts b/packages/computer/src/runtime/runtime.test.ts index cc27ba26..b7eaa234 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,154 @@ 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", value: 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", value: 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", value: 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", value: code }])); + const result = await (await runtime.exec("noop")).result(); + expect(result.status).toBe("cancelled"); + expect(result.exitCode).toBe(code); + } + }); +}); + +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", value: 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", value: 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", value: 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", value: 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 +187,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..2e849705 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,10 +14,8 @@ 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; } @@ -27,6 +26,14 @@ export class WorkspaceRuntime { this.#options = options; } + // Whether the named backend accepts a structured `input` value and + // returns a structured result. Consumers such as the exec tool ask + // this to know whether a backend is callable without the caller + // having to declare it a second time. + isCallable(id: string): boolean { + return this.#options.callableBackendIds.has(id); + } + exec(source: string): Promise>; exec( source: string, @@ -42,36 +49,12 @@ export class WorkspaceRuntime { ): Promise> { if (options.id !== undefined) assertExecutionId(options.id); const backend = this.#backend(options.backend); - if (options.input !== undefined && !this.#options.callableBackendIds.has(backend)) { + if (options.input !== undefined && !this.isCallable(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); - } - - 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 +64,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 +90,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 +99,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 +126,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 +133,7 @@ function wrapModuleHandle( source: ReadableStream, encoding: E | undefined, resultMayUseSource = true, + sync?: ModuleExecutionEnvelope["sync"], ): WorkspaceRuntimeExecHandle { let claimed: "result" | "stream" | undefined; let sourceCancelled = false; @@ -314,10 +192,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; }, @@ -414,6 +293,7 @@ async function drainModuleResult( events: ReadableStream, encoding: E | undefined, setReader: (reader: ReadableStreamDefaultReader | undefined) => void, + sync?: ModuleExecutionEnvelope["sync"], ): Promise> { const stdout: Uint8Array[] = []; const stderr: Uint8Array[] = []; @@ -428,23 +308,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.value; + 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..9562106d 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"; value: 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 { @@ -165,7 +173,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..73ec91b6 --- /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", value: 0, result: { a: [1, 2, null] } }), + ), + ); + expect(events).toEqual([ + { id: "e-1", seq: 2, name: "exit", value: 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", value: 1 })), + ); + expect(events).toEqual([{ id: "e-1", seq: 1, name: "exit", value: 1 }]); + }); +}); diff --git a/packages/computer/src/runtime/wire.ts b/packages/computer/src/runtime/wire.ts index fac275bf..93f40a03 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"; value: 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..e68529ab 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 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 }; } -// --------------------------------------------------------------------------- -// 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..d3d88bb3 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,46 +41,16 @@ 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 { @@ -129,20 +93,20 @@ export interface GetExecOptions { 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 +117,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 +139,7 @@ export class WorkspaceShell { }, () => this.#shell.exec({ - command, + source, id: options.id, cwd: options.cwd, timeoutMs: options.timeoutMs, @@ -194,38 +153,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,146 +190,13 @@ 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( +export function withPostPull( source: ReadableStream>, sync: Sync, ): { stream: ReadableStream>; outcome: Promise } { @@ -447,66 +268,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 +316,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/tools/ai.test.ts b/packages/computer/src/tools/ai.test.ts index 9ea35cdb..1a0cb7ab 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, @@ -17,7 +17,43 @@ async function executeTool(tool: unknown, input: unknown): Promise { const execute = (tool as { execute?: (input: unknown, options: typeof toolOptions) => unknown }) .execute; if (!execute) throw new Error("tool has no execute function"); - return await execute(input, toolOptions); + const output = await execute(input, toolOptions); + if (output && typeof output === "object" && Symbol.asyncIterator in output) { + let last: unknown; + for await (const chunk of output as AsyncIterable) last = chunk; + return last; + } + return output; +} + +async function collectTool(tool: unknown, input: unknown): Promise { + const execute = (tool as { execute?: (input: unknown, options: typeof toolOptions) => unknown }) + .execute; + if (!execute) throw new Error("tool has no execute function"); + const output = await execute(input, toolOptions); + if (!output || typeof output !== "object" || !(Symbol.asyncIterator in output)) { + return [output]; + } + const chunks: unknown[] = []; + for await (const chunk of output as AsyncIterable) chunks.push(chunk); + return chunks; +} + +type ExecStreamEvent = + | { name: "stdout"; value: string } + | { name: "stderr"; value: string } + | { name: "result"; value: unknown } + | { name: "exit"; value: number; result?: unknown }; + +function streamingHandle(events: ExecStreamEvent[]) { + return { + async *[Symbol.asyncIterator]() { + for (const event of events) yield event; + }, + result: async () => { + throw new Error("result() must not be called on a streamed handle"); + }, + }; } function toolDescription(tool: unknown): string { @@ -293,7 +329,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 +337,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 +370,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 +378,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 +403,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 +411,7 @@ describe("createAITools exec tool", () => { pulled: 0, skipped: [], }; - return { result: async () => result } as ExecHandle<"utf8">; + return { result: async () => result } as unknown as WorkspaceRuntimeExecHandle<"utf8">; }, }, }; @@ -481,6 +517,366 @@ describe("createAITools exec tool", () => { }); }); +describe("createAITools callable exec", () => { + it("forwards env and input to the runtime and returns the result value", async () => { + const calls: Array<{ + command: string; + env: Record | undefined; + input: unknown; + backend: string | undefined; + }> = []; + const workspace = { + runtime: { + async exec( + command: string, + options: { + cwd?: string; + encoding: "utf8"; + backend?: string; + env?: Record; + input?: unknown; + }, + ) { + calls.push({ + command, + env: options.env, + input: options.input, + backend: options.backend, + }); + return { + result: async () => ({ + exitCode: 0, + stdout: "ran", + stderr: "", + value: { doubled: 84 }, + }), + }; + }, + isCallable: (id: string) => id === "js", + }, + }; + const tools = createAITools({ + workspace, + shell: { + defaultBackend: "js", + backends: { + js: { description: "JavaScript module runtime" }, + }, + }, + }); + + await expect( + executeTool(tools.exec, { + command: "export default (input) => ({ doubled: input.value * 2 })", + env: { API_KEY: "secret" }, + input: { value: 42 }, + }), + ).resolves.toEqual({ + command: "export default (input) => ({ doubled: input.value * 2 })", + cwd: null, + backend: "js", + exitCode: 0, + stdout: "ran", + stderr: "", + result: { doubled: 84 }, + }); + expect(calls).toEqual([ + { + command: "export default (input) => ({ doubled: input.value * 2 })", + env: { API_KEY: "secret" }, + input: { value: 42 }, + backend: "js", + }, + ]); + }); + + it("omits the result field when the backend returns no value", async () => { + const workspace = { + runtime: { + async exec() { + return { + result: async () => ({ exitCode: 0, stdout: "ok", stderr: "" }), + }; + }, + isCallable: (id: string) => id === "js", + }, + }; + const tools = createAITools({ + workspace, + shell: { + defaultBackend: "js", + backends: { js: { description: "JavaScript module runtime" } }, + }, + }); + + const output = (await executeTool(tools.exec, { command: "noop" })) as Record; + expect(output).not.toHaveProperty("result"); + expect(output).toMatchObject({ backend: "js", exitCode: 0, stdout: "ok" }); + }); + + it("errors quickly without calling the backend when input targets a non-callable backend", async () => { + let called = false; + const workspace = { + runtime: { + async exec() { + called = true; + return { result: async () => ({ exitCode: 0, stdout: "", stderr: "" }) }; + }, + isCallable: (id: string) => id === "js", + }, + }; + const tools = createAITools({ + workspace, + shell: { + defaultBackend: "shell", + backends: { + shell: { description: "fast shell" }, + js: { description: "JavaScript module runtime" }, + }, + }, + }); + + await expect( + executeTool(tools.exec, { command: "echo hi", input: { value: 1 }, backend: "shell" }), + ).resolves.toEqual({ + command: "echo hi", + cwd: null, + backend: "shell", + error: 'Backend "shell" is not callable; it does not accept structured input.', + }); + expect(called).toBe(false); + }); + + it("allows env on non-callable backends", async () => { + const calls: Array<{ env: Record | undefined; input: unknown }> = []; + const workspace = { + runtime: { + async exec(_command: string, options: { env?: Record; input?: unknown }) { + calls.push({ env: options.env, input: options.input }); + return { result: async () => ({ exitCode: 0, stdout: "", stderr: "" }) }; + }, + }, + }; + const tools = createAITools({ + workspace, + shell: { + defaultBackend: "shell", + backends: { shell: { description: "fast shell" } }, + }, + }); + + await expect( + executeTool(tools.exec, { command: "env", env: { FOO: "bar" } }), + ).resolves.toMatchObject({ backend: "shell", exitCode: 0 }); + expect(calls).toEqual([{ env: { FOO: "bar" }, input: undefined }]); + }); + + it("describes callable backends in the tool description", () => { + const workspace = { + runtime: { + async exec() { + throw new Error("not used"); + }, + isCallable: (id: string) => id === "js", + }, + }; + const tools = createAITools({ + workspace, + shell: { + defaultBackend: "js", + backends: { js: { description: "JavaScript module runtime" } }, + }, + }); + + expect(toolDescription(tools.exec)).toContain("callable"); + }); +}); + +describe("createAITools exec streaming", () => { + it("streams stdout and stderr chunks and yields a final aggregate", async () => { + const workspace = { + runtime: { + async exec() { + return streamingHandle([ + { name: "stdout", value: "one\n" }, + { name: "stderr", value: "warn\n" }, + { name: "stdout", value: "two\n" }, + { name: "exit", value: 0 }, + ]); + }, + }, + }; + const tools = createAITools({ + workspace, + shell: { + defaultBackend: "shell", + backends: { shell: { description: "fast shell" } }, + }, + }); + + await expect(collectTool(tools.exec, { command: "run" })).resolves.toEqual([ + { command: "run", cwd: null, backend: "shell", exitCode: null, stdout: "one\n", stderr: "" }, + { + command: "run", + cwd: null, + backend: "shell", + exitCode: null, + stdout: "one\n", + stderr: "warn\n", + }, + { + command: "run", + cwd: null, + backend: "shell", + exitCode: null, + stdout: "one\ntwo\n", + stderr: "warn\n", + }, + { + command: "run", + cwd: null, + backend: "shell", + exitCode: 0, + stdout: "one\ntwo\n", + stderr: "warn\n", + }, + ]); + }); + + it("streams a callable backend's result value on the final chunk", async () => { + const workspace = { + runtime: { + async exec() { + return streamingHandle([ + { name: "stdout", value: "working\n" }, + { name: "result", value: { ok: true } }, + { name: "exit", value: 0 }, + ]); + }, + isCallable: (id: string) => id === "js", + }, + }; + const tools = createAITools({ + workspace, + shell: { + defaultBackend: "js", + backends: { js: { description: "JavaScript module runtime" } }, + }, + }); + + const chunks = await collectTool(tools.exec, { + command: "export default () => ({ ok: true })", + input: {}, + }); + expect(chunks).toHaveLength(2); + expect(chunks.at(-1)).toEqual({ + command: "export default () => ({ ok: true })", + cwd: null, + backend: "js", + exitCode: 0, + stdout: "working\n", + stderr: "", + result: { ok: true }, + }); + }); + + it("reads a callable backend's result folded onto the exit event", async () => { + const workspace = { + runtime: { + async exec() { + return streamingHandle([ + { name: "stdout", value: "working\n" }, + { name: "exit", value: 0, result: { ok: true } }, + ]); + }, + isCallable: (id: string) => id === "js", + }, + }; + const tools = createAITools({ + workspace, + shell: { + defaultBackend: "js", + backends: { js: { description: "JavaScript module runtime" } }, + }, + }); + + const chunks = await collectTool(tools.exec, { + command: "export default () => ({ ok: true })", + input: {}, + }); + expect(chunks).toHaveLength(2); + expect(chunks.at(-1)).toEqual({ + command: "export default () => ({ ok: true })", + cwd: null, + backend: "js", + exitCode: 0, + stdout: "working\n", + stderr: "", + result: { ok: true }, + }); + }); + + it("truncates streamed output on UTF-8 byte boundaries", async () => { + const workspace = { + runtime: { + async exec() { + return streamingHandle([ + { name: "stdout", value: "a\u{1f642}b" }, + { name: "exit", value: 0 }, + ]); + }, + }, + }; + const tools = createAITools({ + workspace, + shell: { + defaultBackend: "shell", + backends: { shell: { description: "fast shell" } }, + maxBytes: 5, + }, + }); + + const chunks = await collectTool(tools.exec, { command: "echo emoji" }); + expect(chunks.at(-1)).toMatchObject({ + exitCode: 0, + stdout: "a\u{1f642}\n\n[truncated, 1 more bytes]", + }); + }); + + it("yields a structured error when the stream fails mid-run", async () => { + const workspace = { + runtime: { + async exec() { + return { + async *[Symbol.asyncIterator]() { + yield { name: "stdout", value: "partial\n" } as ExecStreamEvent; + throw new Error("stream broke"); + }, + result: async () => { + throw new Error("result() must not be called on a streamed handle"); + }, + }; + }, + }, + }; + const tools = createAITools({ + workspace, + shell: { + defaultBackend: "shell", + backends: { shell: { description: "fast shell" } }, + }, + }); + + const chunks = await collectTool(tools.exec, { command: "run" }); + expect(chunks.at(-1)).toEqual({ + command: "run", + cwd: null, + backend: "shell", + error: "stream broke", + }); + }); +}); + describe("createAITools publish tool", () => { it("adds publish by default when assets are configured", async () => { const calls: Array<{ path: string; expiresAfter: number; prefix?: string }> = []; diff --git a/packages/computer/src/tools/exec.ts b/packages/computer/src/tools/exec.ts index efbb93c1..8c27e0d3 100644 --- a/packages/computer/src/tools/exec.ts +++ b/packages/computer/src/tools/exec.ts @@ -1,18 +1,50 @@ import { type Tool, tool } from "ai"; import { z } from "zod"; +// One event drained from a running execution. stdout / stderr carry +// output chunks as they arrive; exit carries the process exit code and, +// for a callable backend, the structured return value on `result`. The +// value settles at the same instant as the exit code, so it rides the +// same terminal event rather than a separate one. +// +// The standalone `result` event is the shape the runtime wire still +// emits today. The tool accepts it so it keeps working until the wire is +// consolidated, but callers should read the value from the exit event. +export type ExecStreamEvent = + | { name: "stdout"; value: string } + | { name: "stderr"; value: string } + | { name: "result"; value: unknown } + | { name: "exit"; value: number; result?: unknown }; + +// A detached execution handle. The tool streams stdout / stderr +// chunks by iterating the handle when it is async-iterable, and +// falls back to draining result() when it is not. +export interface ExecRuntimeHandle extends Partial> { + result(): Promise<{ + exitCode: number; + stdout: string; + stderr: string; + value?: unknown; + }>; +} + export interface ExecWorkspaceLike { runtime: { exec( command: string, - options: { cwd?: string; encoding: "utf8"; backend?: string }, - ): Promise<{ - result(): Promise<{ - exitCode: number; - stdout: string; - stderr: string; - }>; - }>; + options: { + cwd?: string; + encoding: "utf8"; + backend?: string; + env?: Record; + input?: unknown; + }, + ): Promise; + // Whether a backend accepts a structured `input` value and returns + // a structured result. The tool asks this to know which backends + // are callable; the runtime derives it from each backend's + // `callable` flag. Omit when no backend is callable. + isCallable?(id: string): boolean; }; } @@ -29,9 +61,32 @@ export interface ExecToolOptions { const DEFAULT_MAX_BYTES = 64 * 1024; -export function createExecTool( - options: ExecToolOptions, -): Tool<{ command: string; cwd?: string; backend?: string }> { +// Progressive snapshot emitted while a command streams (exitCode +// null until the run ends), and the terminal snapshot once the exit +// code lands. `result` appears only when a callable backend returned +// a value; `error` replaces the run fields when the exec fails. +export type ExecToolOutput = + | { + command: string; + cwd: string | null; + backend: string; + exitCode: number | null; + stdout: string; + stderr: string; + result?: unknown; + } + | { command: string; cwd: string | null; backend: string; error: string }; + +export function createExecTool(options: ExecToolOptions): Tool< + { + command: string; + cwd?: string; + backend?: string; + env?: Record; + input?: unknown; + }, + ExecToolOutput +> { const maxBytes = options.maxBytes ?? DEFAULT_MAX_BYTES; const backendIds = Object.keys(options.backends); if (backendIds.length === 0) { @@ -43,9 +98,21 @@ export function createExecTool( ); } + const isCallable = options.workspace.runtime.isCallable?.bind(options.workspace.runtime); + const callableBackendIds = new Set(backendIds.filter((id) => isCallable?.(id) === true)); const backendGuidance = backendIds - .map((id) => `- ${JSON.stringify(id)}: ${options.backends[id].description}`) + .map((id) => { + const suffix = callableBackendIds.has(id) ? " (callable)" : ""; + return `- ${JSON.stringify(id)}${suffix}: ${options.backends[id].description}`; + }) .join("\n"); + const callableGuidance = + callableBackendIds.size > 0 + ? [ + "", + `Callable backends (${[...callableBackendIds].map((id) => JSON.stringify(id)).join(", ")}) run \`command\` as module source rather than a shell command. Pass \`input\` to hand the module a structured value, and read the module's returned value back from the \`result\` field. Other backends reject \`input\`.`, + ].join("\n") + : ""; const description = [ "Run a shell command in the workspace. The workspace exposes multiple backends, each with different capabilities.", "Pick the cheapest backend that can run the command; fall back to a heavier one only when the lighter backend's command set doesn't cover what you need.", @@ -55,6 +122,7 @@ export function createExecTool( "", `Default backend: ${JSON.stringify(options.defaultBackend)}. Try this first for any command you're not sure about; if it fails with a "command not found" or a similar capability error, retry on a backend whose description covers the missing tool.`, "Use for builds, test runs, typechecks, formatters, and git plumbing. Prefer the dedicated read, write, and edit tools for file operations. Long output is truncated to keep tool replies small.", + callableGuidance, ].join("\n"); const backendSchema = z @@ -71,39 +139,120 @@ export function createExecTool( return tool({ description, inputSchema: z.object({ - command: z.string().describe("Shell command, e.g. 'npm test' or 'git diff HEAD'."), + command: z + .string() + .describe( + "Shell command, e.g. 'npm test' or 'git diff HEAD'. For a callable backend this is the module source to run.", + ), cwd: z.string().optional().describe("Working directory. Defaults to the workspace root."), backend: backendSchema, + env: z + .record(z.string(), z.string()) + .optional() + .describe( + "Environment variables for this run only. Values override the backend's base environment without affecting later runs.", + ), + input: z + .unknown() + .optional() + .describe( + "Structured value handed to a callable backend's module. Only callable backends accept it; other backends reject it.", + ), }), - execute: async ({ command, cwd, backend }) => { + execute: async function* ({ command, cwd, backend, env, input }) { const selectedBackend = backend ?? options.defaultBackend; + const base = { command, cwd: cwd ?? null, backend: selectedBackend }; + if (input !== undefined && !callableBackendIds.has(selectedBackend)) { + yield { + ...base, + error: `Backend ${JSON.stringify(selectedBackend)} is not callable; it does not accept structured input.`, + }; + return; + } + let handle: ExecRuntimeHandle; try { - const handle = await options.workspace.runtime.exec(command, { + handle = await options.workspace.runtime.exec(command, { cwd, encoding: "utf8", backend: selectedBackend, + env, + input, }); + } catch (err) { + yield { ...base, error: errorMessage(err) }; + return; + } + + // Stream stdout / stderr chunks as they arrive when the handle + // is iterable. Each chunk yields a fresh snapshot with the + // running output so the model sees progress before the run + // ends; the exit event settles the terminal snapshot. + if (typeof handle[Symbol.asyncIterator] === "function") { + let stdout = ""; + let stderr = ""; + let exitCode: number | null = null; + let value: unknown; + let hasValue = false; + try { + for await (const event of handle as AsyncIterable) { + if (event.name === "stdout") stdout += event.value; + else if (event.name === "stderr") stderr += event.value; + else if (event.name === "result") { + value = event.value; + hasValue = true; + continue; + } else { + exitCode = event.value; + if ("result" in event) { + value = event.result; + hasValue = true; + } + continue; + } + // A stdout / stderr chunk arrived: emit a running snapshot + // so the model sees output before the run ends. + yield { + ...base, + exitCode: null, + stdout: truncate(stdout, maxBytes), + stderr: truncate(stderr, maxBytes), + }; + } + } catch (err) { + yield { ...base, error: errorMessage(err) }; + return; + } + yield { + ...base, + exitCode, + stdout: truncate(stdout, maxBytes), + stderr: truncate(stderr, maxBytes), + ...(hasValue ? { result: value } : {}), + }; + return; + } + + // Non-streaming handle: drain the aggregate result. + try { const result = await handle.result(); - return { - command, - cwd: cwd ?? null, - backend: selectedBackend, + yield { + ...base, exitCode: result.exitCode, stdout: truncate(result.stdout, maxBytes), stderr: truncate(result.stderr, maxBytes), + ...(result.value === undefined ? {} : { result: result.value }), }; } catch (err) { - return { - command, - cwd: cwd ?? null, - backend: selectedBackend, - error: err instanceof Error ? err.message : String(err), - }; + yield { ...base, error: errorMessage(err) }; } }, }); } +function errorMessage(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} + const encoder = new TextEncoder(); function truncate(value: string, maxBytes: number): string { diff --git a/packages/computer/src/tools/index.ts b/packages/computer/src/tools/index.ts index 45dc297c..b2686a55 100644 --- a/packages/computer/src/tools/index.ts +++ b/packages/computer/src/tools/index.ts @@ -1,5 +1,12 @@ export { type CreateAIToolsOptions, createAITools } from "./ai.js"; -export { createExecTool, type ExecBackendDescription, type ExecToolOptions } from "./exec.js"; +export { + createExecTool, + type ExecBackendDescription, + type ExecRuntimeHandle, + type ExecStreamEvent, + type ExecToolOptions, + type ExecToolOutput, +} from "./exec.js"; export { createEditTool, type EditToolOptions } from "./fs/edit.js"; export { createListTool, type ListToolOptions } from "./fs/list.js"; export { createReadTool, type ReadToolOptions } from "./fs/read.js"; diff --git a/packages/computer/src/workspace.test.ts b/packages/computer/src/workspace.test.ts index 467d7fee..b3236f04 100644 --- a/packages/computer/src/workspace.test.ts +++ b/packages/computer/src/workspace.test.ts @@ -417,8 +417,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", value: 0, result: 42 }); controller.close(); }, }); diff --git a/packages/computer/src/workspace.ts b/packages/computer/src/workspace.ts index e75ada0f..bc283fc9 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,9 @@ 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(); readonly #moduleHandles = new Map(); readonly #connectingModuleHandles = new Map>(); #connectionGeneration = 0; @@ -270,7 +265,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 +295,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 +317,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 +424,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 +521,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. @@ -756,20 +740,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 @@ -795,6 +765,83 @@ 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 + // its 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); + } + + async #commandHandleFor(id: string): Promise { + const { shell, handle } = await this.#shellFor(id); + const onError = (error: unknown) => this.#onShellError(id, handle, error); + return { + 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; + } + }, + close: async () => { + // The backend handle owns the transport; closing happens + // through the Workspace's own close path. + }, + }; + } + #moduleHandleFor(id: string): Promise { const cached = this.#moduleHandles.get(id); if (cached) return Promise.resolve(cached); @@ -815,7 +862,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, @@ -860,7 +906,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, @@ -912,18 +957,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 +981,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 +992,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 +1051,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/rpc/src/interface.ts b/packages/rpc/src/interface.ts index 7eb5bca0..f897ca72 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"; value: 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..956d9886 100644 --- a/packages/rpc/tests/shell-and-composite.test.ts +++ b/packages/rpc/tests/shell-and-composite.test.ts @@ -158,7 +158,7 @@ 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"); @@ -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,7 +266,7 @@ 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 }); @@ -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");