computer, rpc, computerd: harden JavaScript finalize and add command stdin

Treat a JavaScript execution whose output stream closes without an exit
frame as a failure rather than a silent exit 0 with no result: that
state means a dropped frame write or a crashed isolate. Serialize the
runner's frame writes on a chain the task awaits before closing, so a
rejected write is caught instead of surfacing as an unhandled rejection.

Thread standard input through the command path so container and worker
shell backends receive it: the shell facade, the shell RPC contract and
its server, the container runner spawn, and the worker shell entrypoint
into just-bash. Reject structured input on the command path as well.

Cover the new behavior: a no-exit-frame run settling as failed, stdin
forwarded to command backends and to just-bash, stdin fed to a real
spawned child, and structured input rejected on a command backend.

Author: aron-cf

packages/computer/src/backends/worker-javascript/worker-javascript.test.ts

@@ -638,6 +638,57 @@ describe("WorkerJavaScriptBackend", () => {
     await handle.close();
   });
 
+  it("settles as failed when the output stream closes without an exit frame", async () => {
+    const db = new Database(new SQLiteTestStorage());
+    initializeSchema(db, () => 0);
+    const fs = new WorkspaceFilesystem(db);
+    await fs.mkdir("/workspace", { recursive: true });
+    const encoder = new TextEncoder();
+    const backend = new WorkerJavaScriptBackend({
+      loader: {
+        load() {
+          return {
+            getEntrypoint() {
+              return {
+                async evaluate(
+                  _input: unknown,
+                  host: { attachOutput(readable: ReadableStream<Uint8Array>): Promise<void> },
+                ) {
+                  // Emit stdout, then close the stream with no result or
+                  // exit frame, mimicking a dropped terminal write.
+                  const readable = new ReadableStream<Uint8Array>({
+                    start(controller) {
+                      controller.enqueue(
+                        encoder.encode(
+                          `${JSON.stringify({ name: "stdout", b64: btoa("partial\n") })}\n`,
+                        ),
+                      );
+                      controller.close();
+                    },
+                  });
+                  await host.attachOutput(readable);
+                },
+              };
+            },
+          };
+        },
+      },
+    });
+    const handle = await backend.connect({
+      db,
+      fs,
+      git: undefined as never,
+      artifacts: undefined as never,
+    });
+    const execution = await handle.exec({ id: "no-exit", source: "export default 1" });
+    const events = [];
+    for await (const event of execution.events) events.push(event);
+    const exit = events.find((event) => event.name === "exit");
+    expect(exit).toMatchObject({ name: "exit", value: 1 });
+    expect(events.some((event) => event.name === "result")).toBe(false);
+    await handle.close();
+  });
+
   it("aborts cooperative trusted-module calls at their deadline", async () => {
     const db = new Database(new SQLiteTestStorage());
     initializeSchema(db, () => 0);

packages/computer/src/backends/worker-javascript/worker-javascript.ts

@@ -659,7 +659,23 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle {
       ]);
       return;
     }
-    const exitCode = record.exitCode ?? 0;
+    // No exit frame means the runner never reported a terminal state:
+    // the output stream closed early, a frame write was dropped, or the
+    // isolate crashed. Settle as a failure rather than a silent exit 0
+    // with no result.
+    if (record.exitCode === undefined) {
+      this.#finish(record, "failed", [
+        {
+          id: record.id,
+          seq: record.events.length + 1,
+          name: "stderr",
+          value: new TextEncoder().encode("Execution ended without reporting a result.\n"),
+        },
+        { id: record.id, seq: record.events.length + 2, name: "exit", value: 1 },
+      ]);
+      return;
+    }
+    const exitCode = record.exitCode;
     const terminal: WorkspaceRuntimeEvent[] = [];
     if (exitCode === 0 && record.hasResult) {
       terminal.push({
@@ -1054,10 +1070,14 @@ function runtimeWorkerModule(entryName: string, maxStdioBytes: number) {
           }
           return btoa(binary);
         };
+        // Serialize writes on a chain so a rejected write (the host
+        // cancelled or the stream errored) is caught here rather than
+        // surfacing as an unhandled rejection inside the isolate. The
+        // task awaits writeChain before closing.
+        let writeChain = Promise.resolve();
         const enqueue = (frame) => {
-          try {
-            writer.write(encoder.encode(JSON.stringify(frame) + "\\n"));
-          } catch {}
+          const bytes = encoder.encode(JSON.stringify(frame) + "\\n");
+          writeChain = writeChain.then(() => writer.write(bytes)).catch(() => {});
         };
         let stdioBytes = 0;
         let stdioTruncated = false;
@@ -1145,6 +1165,7 @@ function runtimeWorkerModule(entryName: string, maxStdioBytes: number) {
           enqueue({ name: "stderr", b64: toBase64(truncate(message) + "\\n") });
           enqueue({ name: "exit", value: 1 });
         }
+        await writeChain;
         try {
           await writer.close();
         } catch {}

packages/computer/src/backends/worker-shell/entrypoint.test.ts

@@ -33,7 +33,12 @@ class TestShellWorker extends ShellWorker {
     env: E,
     bashFactory: (
       command: string,
-      options: { cwd?: string; env?: Record<string, string>; signal?: AbortSignal },
+      options: {
+        cwd?: string;
+        env?: Record<string, string>;
+        stdin?: Uint8Array;
+        signal?: AbortSignal;
+      },
     ) => Promise<{ stdout: string; stderr: string; exitCode: number }>,
   ): TestShellWorker {
     const w = new TestShellWorker(undefined as never, env as never);
@@ -203,6 +208,17 @@ describe("ShellWorker", () => {
     expect(observedEnv).toEqual({ TOKEN: "secret", EMPTY: "" });
   });
 
+  it("forwards per-execution stdin bytes to Bash", async () => {
+    let observedStdin: Uint8Array | undefined;
+    const worker = TestShellWorker.withFakeBash(fakeEnv(), async (_command, options) => {
+      observedStdin = options.stdin;
+      return { stdout: "", stderr: "", exitCode: 0 };
+    });
+    const bytes = new TextEncoder().encode("piped");
+    await drain((await worker.exec({ command: "cat", stdin: bytes })).events);
+    expect(observedStdin).toEqual(bytes);
+  });
+
   it("getExec without a prior exec throws ENOENT", async () => {
     const worker = new TestShellWorker(undefined as never, fakeEnv() as never);
     await expect(worker.getExec({ id: "missing" })).rejects.toMatchObject({ code: "ENOENT" });

packages/computer/src/backends/worker-shell/entrypoint.ts

@@ -31,6 +31,7 @@ export interface ExecInput {
   id?: string;
   timeoutMs?: number;
   env?: Record<string, string>;
+  stdin?: Uint8Array;
 }
 
 export interface ShellWorkerOptions {
@@ -107,6 +108,7 @@ export class ShellWorker<
         options: {
           cwd?: string;
           env?: Record<string, string>;
+          stdin?: Uint8Array;
           signal?: AbortSignal;
           customCommands: CustomCommand[];
         },
@@ -174,6 +176,7 @@ export class ShellWorker<
         result = await this.bashFactoryOverride(input.command, {
           cwd,
           env: input.env,
+          stdin: input.stdin,
           signal: controller.signal,
           customCommands,
         });
@@ -195,7 +198,14 @@ export class ShellWorker<
           defenseInDepth: { enabled: false },
           executionLimits: { maxOutputSize: MAX_OUTPUT_BYTES },
         });
-        result = await bash.exec(input.command, { cwd, env: input.env, signal: controller.signal });
+        result = await bash.exec(input.command, {
+          cwd,
+          env: input.env,
+          ...(input.stdin !== undefined
+            ? { stdin: latin1FromBytes(input.stdin), stdinKind: "bytes" as const }
+            : {}),
+          signal: controller.signal,
+        });
       }
     } catch (error) {
       const message = error instanceof Error ? error.message : String(error);
@@ -263,6 +273,16 @@ function framedStream(events: WireEvent[]): ReadableStream<Uint8Array> {
   });
 }
 
+// just-bash's `stdin` with `stdinKind: "bytes"` carries each byte as one
+// latin1 char. Pack the caller's bytes into that shape.
+function latin1FromBytes(bytes: Uint8Array): string {
+  let result = "";
+  for (let index = 0; index < bytes.length; index += 1) {
+    result += String.fromCharCode(bytes[index]);
+  }
+  return result;
+}
+
 function createShellError(code: string, message: string): Error & { code: string } {
   const error = new Error(message) as Error & { code: string };
   error.name = "ShellWorkerError";

packages/computer/src/backends/worker-shell/worker-shell.ts

@@ -39,6 +39,7 @@ export interface WorkerShellFetcher {
     id?: string;
     timeoutMs?: number;
     env?: Record<string, string>;
+    stdin?: Uint8Array;
   }): Promise<{
     id: string;
     events: ReadableStream<Uint8Array>;
@@ -175,6 +176,7 @@ export class WorkerShellBackend implements WorkspaceBackend {
           id: input.id,
           timeoutMs: input.timeoutMs,
           env: input.env,
+          stdin: input.stdin,
         });
         return { id: envelope.id, events: decodeFramedEvents(envelope.events) };
       },

packages/computer/src/runtime/runtime.ts

@@ -66,6 +66,7 @@ export class WorkspaceRuntime {
         id: options.id,
         timeoutMs: options.timeoutMs,
         env: options.env,
+        stdin: options.stdin,
       });
       return wrapCommandHandle(handle, backend);
     }

packages/computer/src/shell.ts

@@ -109,6 +109,9 @@ export interface ExecOptions<E extends ExecEncoding = undefined> {
   // override the backend's base environment without changing later
   // executions.
   env?: Record<string, string>;
+  // Standard input fed to the command. Bytes, or a string encoded
+  // as UTF-8.
+  stdin?: Uint8Array | string;
   // Backend selector. Omit to use the default backend (the first
   // one passed to the Workspace constructor); pass the id of
   // another configured backend to route this call there.
@@ -182,6 +185,10 @@ export class WorkspaceShell {
           cwd: options.cwd,
           timeoutMs: options.timeoutMs,
           env: options.env,
+          stdin:
+            typeof options.stdin === "string"
+              ? new TextEncoder().encode(options.stdin)
+              : options.stdin,
         }),
       (span, outcome) => {
         if (outcome.ok) span.setAttribute("workspace.runtime.id", outcome.value.id);

packages/computer/src/workspace.test.ts

@@ -243,6 +243,46 @@ describe("Workspace backend selection", () => {
     expect(result).toMatchObject({ status: "completed", exitCode: 0 });
   });
 
+  it("rejects structured input for a non-callable command backend", async () => {
+    const backend = execBackend("command", () => {});
+    const ws = new Workspace({ storage: makeStorage(), backends: [backend] });
+    await expect(ws.runtime.exec("true", { backend: "command", input: { a: 1 } })).rejects.toThrow(
+      /not callable/,
+    );
+  });
+
+  it("forwards per-execution stdin to command backends", async () => {
+    let receivedStdin: Uint8Array | undefined;
+    const shell: import("@cloudflare/computer-rpc").ShellRPC = {
+      async exec(input) {
+        receivedStdin = input.stdin;
+        const id = input.id ?? "stdin-command";
+        return {
+          id,
+          events: new ReadableStream({
+            start(controller) {
+              controller.enqueue({ id, seq: 1, name: "exit", value: 0 });
+              controller.close();
+            },
+          }),
+        };
+      },
+      getExec: () => Promise.reject(new Error("not used")),
+      killExec: () => Promise.reject(new Error("not used")),
+      disposeExec: async () => undefined,
+    };
+    const backend: WorkspaceBackend = {
+      id: "command",
+      type: "fake",
+      async connect() {
+        return { rpc: { sync: fakeRpc(), shell }, sync: "none", close: async () => undefined };
+      },
+    };
+    const ws = new Workspace({ storage: makeStorage(), backends: [backend] });
+    await drainExec(await ws.runtime.exec("cat", { backend: "command", stdin: "piped" }));
+    expect(receivedStdin).toEqual(new TextEncoder().encode("piped"));
+  });
+
   it("forwards per-execution environment variables to command backends", async () => {
     let receivedEnv: Record<string, string> | undefined;
     const shell: import("@cloudflare/computer-rpc").ShellRPC = {

packages/computerd/src/exec/runner.test.ts

@@ -103,6 +103,23 @@ test("per-execution env overrides the base env without leaking to later commands
   }
 });
 
+test("feeds per-execution stdin to the child and closes it", async () => {
+  const { runner, dispose } = fixture();
+  try {
+    const handle = runner.exec("cat", { stdin: new TextEncoder().encode("piped-input") });
+    const events = await drain(handle.events);
+    const stdout = events
+      .filter((event) => event.name === "stdout")
+      .map((event) => decode(event.value as Uint8Array))
+      .join("");
+    const exit = events.find((event) => event.name === "exit");
+    expect(stdout).toBe("piped-input");
+    expect(exit?.value).toBe(0);
+  } finally {
+    dispose();
+  }
+});
+
 test("reusing a live id throws EEXEC_BUSY", async () => {
   const { runner, dispose } = fixture();
   try {

packages/computerd/src/exec/runner.ts

@@ -144,8 +144,15 @@ export class Runner {
     const wrapped = cwd !== undefined ? `cd ${shellQuote(cwd)} && ${command}` : command;
     const child = spawn("/bin/sh", ["-c", wrapped], {
       env,
-      stdio: ["ignore", "pipe", "pipe"],
+      stdio: [options.stdin === undefined ? "ignore" : "pipe", "pipe", "pipe"],
     });
+    if (options.stdin !== undefined && child.stdin) {
+      // Feed the caller's bytes then close so the child sees EOF.
+      // Ignore write/EPIPE errors: a command that never reads stdin
+      // (or exits first) must not fail the run.
+      child.stdin.on("error", () => {});
+      child.stdin.end(Buffer.from(options.stdin));
+    }
     const log = createLog(this.db, id, {
       maxBytes: this.opts.logMaxBytes,
       now: this.opts.now,