Stream JavaScript execution output live over a framed stream

docs/05_runtime_interface.md

@@ -31,6 +31,8 @@ interface WorkspaceRuntimeExecOptions {
   encoding?: "utf8";
   input?: WorkspaceRuntimeValue;
   timeoutMs?: number;
+  env?: Record<string, string>;
+  stdin?: Uint8Array | string;
 }
 
 interface WorkspaceRuntimeExecHandle extends ReadableStream<WorkspaceRuntimeEvent> {
@@ -42,7 +44,7 @@ interface WorkspaceRuntimeExecHandle extends ReadableStream<WorkspaceRuntimeEven
 }
 ```
 
-`input` is accepted by structured module backends and rejected by command backends. `cwd` is the command working directory or the base for durable relative module imports. A handle is single-consumer: call `result()` or consume its event stream, not both. Repeated `result()` calls return the same promise. `backend` records the resolved backend needed for later reattachment.
+`input` is accepted by callable backends and rejected by the rest; it carries a structured value that the callable backend returns a structured value for. `env` is accepted everywhere: command backends inherit it for the spawned command, and the JavaScript module backend exposes it through `process.env`. Its values apply to that execution only and do not change later executions. `stdin` is the caller-supplied standard input, accepted by backends that model it (the JavaScript module backend reads it through `process.stdin`). `cwd` is the command working directory or the base for durable relative module imports. A handle is single-consumer: call `result()` or consume its event stream, not both. Repeated `result()` calls return the same promise. `backend` records the resolved backend needed for later reattachment.
 
 ## Results
 

docs/17_isolate_javascript.md

@@ -83,7 +83,7 @@ Workspace parses the graph before loading the Worker, confines every durable pat
 
 The backend admits one execution at a time by default. A concurrent start fails with `EEXEC_BUSY` instead of creating an unbounded number of Dynamic Workers. Set `maxConcurrentExecutions` only after measuring the Durable Object and Worker Loader limits for the deployment.
 
-Each execution also bounds log events, active event subscribers, directory entries per read, concurrent and total capability calls, and cumulative capability request and response bytes. The corresponding `maxLogEvents`, `maxExecutionSubscribers`, `maxDirectoryEntries`, and `max*Capability*` options may be lowered for public workloads. Directory reads apply their limit in SQLite before materializing rows. Requests are checked inside the isolate before Workers RPC and again by the host.
+Each execution also bounds combined stdout and stderr output, active event subscribers, directory entries per read, concurrent and total capability calls, and cumulative capability request and response bytes. The corresponding `maxStdioBytes`, `maxExecutionSubscribers`, `maxDirectoryEntries`, and `max*Capability*` options may be lowered for public workloads. Directory reads apply their limit in SQLite before materializing rows. Requests are checked inside the isolate before Workers RPC and again by the host.
 
 Completed execution records remain available for replay for five minutes by default. The backend also keeps at most 100 completed records. Configure these bounds with `retentionMs` and `maxRetainedExecutions`. Completed records leave the in-memory active set immediately; replay reads them from SQLite.
 
@@ -99,7 +99,7 @@ Each execution installs a small `node:process` shim so ordinary module code can
 
 `process.stdin` is a non-interactive async-iterable over the caller-supplied `stdin` bytes. The caller passes `stdin` as a `Uint8Array` or string on the exec options; `for await` yields the bytes once and then ends, and there is no blocking read for further input because an evaluate-once execution has no session to wait on. `isTTY` is `false`. The supplied input is bounded by `maxStdinBytes`; exceeding it fails the run with a clear error.
 
-`process.stdout` and `process.stderr` are writable streams whose writes are captured as standard output and standard error. `console.log` and `console.info` route to standard output, `console.warn` and `console.error` route to standard error, and the captured output is bounded. `process.argv`, `process.cwd()`, and `process.platform` return inert values: `cwd()` reflects the execution's working directory, while `argv` and `platform` carry fixed placeholders rather than describing the host process.
+`process.stdout` and `process.stderr` are writable streams whose writes flow to the live output described under Isolation and lifecycle. `console.log` and `console.info` route to standard output, `console.warn` and `console.error` route to standard error, and both share the single `maxStdioBytes` ceiling. `process.argv`, `process.cwd()`, and `process.platform` return inert values: `cwd()` reflects the execution's working directory, while `argv` and `platform` carry fixed placeholders rather than describing the host process.
 
 ```ts
 const handle = await workspace.runtime.exec(
@@ -185,12 +185,12 @@ Each execution receives a fresh Dynamic Worker with:
 - a host wall-clock deadline;
 - `globalOutbound: null` by default;
 - finite, acyclic JSON-compatible input and structured result validation;
-- configurable source/module graph, input, result, stdin, captured-log, file/capability request, and response byte limits (`maxSourceBytes`, `maxInputBytes`, `maxResultBytes`, `maxStdinBytes`, `maxLogBytes`, and `maxCapabilityBytes`);
+- configurable source/module graph, input, result, stdin, stdio, file/capability request, and response byte limits (`maxSourceBytes`, `maxInputBytes`, `maxResultBytes`, `maxStdinBytes`, `maxStdioBytes`, and `maxCapabilityBytes`);
 - explicit entrypoint and Worker disposal;
 - host-owned cancellation;
 - retained events and result rows in the Workspace database.
 
-Console output is bounded but currently buffered in the Dynamic Worker and published when evaluation settles; the execution event stream provides replay/lifecycle semantics rather than live JavaScript console streaming. Completed writes are durable immediately. Failure or cancellation does not roll back filesystem effects already completed.
+Standard output and standard error stream live. The Dynamic Worker hands the readable end of its output stream to the host through the `attachOutput` bridge call, and the host drains it frame by frame while user code is still running, appending each chunk to the execution event stream as it arrives rather than buffering the run and publishing at the end. The structured result and the exit event settle once the output stream closes, so the terminal events always follow the last output. Output remains bounded by `maxStdioBytes` across both streams. Completed writes are durable immediately. Failure or cancellation does not roll back filesystem effects already completed.
 
 ## Trusted integrations
 

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

@@ -0,0 +1,101 @@
+import { describe, expect, it } from "vitest";
+
+import { decodeRuntimeFrames, parseRuntimeFrame, type RuntimeFrame } from "./frames.js";
+
+function streamOf(...chunks: string[]): ReadableStream<Uint8Array> {
+  const encoder = new TextEncoder();
+  return new ReadableStream<Uint8Array>({
+    start(controller) {
+      for (const chunk of chunks) controller.enqueue(encoder.encode(chunk));
+      controller.close();
+    },
+  });
+}
+
+async function collect(stream: ReadableStream<RuntimeFrame>): Promise<RuntimeFrame[]> {
+  const frames: RuntimeFrame[] = [];
+  const reader = stream.getReader();
+  while (true) {
+    const next = await reader.read();
+    if (next.done) break;
+    frames.push(next.value);
+  }
+  return frames;
+}
+
+const b64 = (text: string) => btoa(text);
+
+describe("parseRuntimeFrame", () => {
+  it("decodes a base64 stdout frame into raw bytes", () => {
+    const frame = parseRuntimeFrame(`{"name":"stdout","b64":"${b64("hello")}"}`);
+    expect(frame).toEqual({ name: "stdout", value: new TextEncoder().encode("hello") });
+  });
+
+  it("decodes a base64 stderr frame into raw bytes", () => {
+    const frame = parseRuntimeFrame(`{"name":"stderr","b64":"${b64("oops")}"}`);
+    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("rejects invalid JSON", () => {
+    expect(() => parseRuntimeFrame("not json")).toThrow();
+  });
+
+  it("rejects an unknown frame name", () => {
+    expect(() => parseRuntimeFrame(`{"name":"other","value":1}`)).toThrow();
+  });
+
+  it("rejects a malformed stdout frame missing its payload", () => {
+    expect(() => parseRuntimeFrame(`{"name":"stdout"}`)).toThrow();
+  });
+
+  it("rejects an exit frame whose value is not an integer", () => {
+    expect(() => parseRuntimeFrame(`{"name":"exit","value":"x"}`)).toThrow();
+  });
+});
+
+describe("decodeRuntimeFrames", () => {
+  it("decodes newline-delimited frames arriving in one chunk", async () => {
+    const frames = await collect(
+      decodeRuntimeFrames(
+        streamOf(`{"name":"stdout","b64":"${b64("hi")}"}\n{"name":"exit","value":0}\n`),
+      ),
+    );
+    expect(frames).toEqual([
+      { name: "stdout", value: new TextEncoder().encode("hi") },
+      { name: "exit", value: 0 },
+    ]);
+  });
+
+  it("reassembles a frame split across chunk boundaries", async () => {
+    const line = `{"name":"stdout","b64":"${b64("split")}"}\n`;
+    const mid = Math.floor(line.length / 2);
+    const frames = await collect(
+      decodeRuntimeFrames(streamOf(line.slice(0, mid), line.slice(mid))),
+    );
+    expect(frames).toEqual([{ name: "stdout", value: new TextEncoder().encode("split") }]);
+  });
+
+  it("emits a trailing frame that arrives without a final newline", async () => {
+    const frames = await collect(decodeRuntimeFrames(streamOf(`{"name":"exit","value":1}`)));
+    expect(frames).toEqual([{ name: "exit", value: 1 }]);
+  });
+
+  it("skips blank lines between frames", async () => {
+    const frames = await collect(decodeRuntimeFrames(streamOf(`{"name":"exit","value":0}\n\n`)));
+    expect(frames).toEqual([{ name: "exit", value: 0 }]);
+  });
+
+  it("errors the stream on a malformed frame", async () => {
+    await expect(collect(decodeRuntimeFrames(streamOf(`garbage\n`)))).rejects.toThrow();
+  });
+});

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

@@ -0,0 +1,74 @@
+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 };
+
+export function parseRuntimeFrame(line: string): RuntimeFrame {
+  let record: Record<string, unknown>;
+  try {
+    record = JSON.parse(line) as Record<string, unknown>;
+  } catch {
+    throw new Error("WorkerJavaScriptBackend received an invalid execution frame");
+  }
+  const name = record.name;
+  if (name === "stdout" || name === "stderr") {
+    if (typeof record.b64 !== "string") {
+      throw new Error("WorkerJavaScriptBackend received a malformed output frame");
+    }
+    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");
+    }
+    return { name, value: record.value as number };
+  }
+  throw new Error("WorkerJavaScriptBackend received an unknown execution frame");
+}
+
+export function decodeRuntimeFrames(
+  source: ReadableStream<Uint8Array>,
+): ReadableStream<RuntimeFrame> {
+  const decoder = new TextDecoder();
+  let buffer = "";
+  const drain = (controller: TransformStreamDefaultController<RuntimeFrame>, final: boolean) => {
+    let nl = buffer.indexOf("\n");
+    while (nl !== -1) {
+      const line = buffer.slice(0, nl);
+      buffer = buffer.slice(nl + 1);
+      if (line.length > 0) controller.enqueue(parseRuntimeFrame(line));
+      nl = buffer.indexOf("\n");
+    }
+    if (final && buffer.length > 0) {
+      controller.enqueue(parseRuntimeFrame(buffer));
+      buffer = "";
+    }
+  };
+  return source.pipeThrough(
+    new TransformStream<Uint8Array, RuntimeFrame>({
+      transform(chunk, controller) {
+        buffer += decoder.decode(chunk, { stream: true });
+        drain(controller, false);
+      },
+      flush(controller) {
+        buffer += decoder.decode();
+        drain(controller, true);
+      },
+    }),
+  );
+}
+
+function decodeBase64(value: string): Uint8Array {
+  const binary = atob(value);
+  const bytes = new Uint8Array(binary.length);
+  for (let index = 0; index < binary.length; index += 1) {
+    bytes[index] = binary.charCodeAt(index);
+  }
+  return bytes;
+}