Exec and process output (NDJSON)

Exec streaming, retained process output and setup logs use different frame unions. Read Content-Type before decoding: a keyed replay is JSON identity even if the original request asked for NDJSON. Do not run a command again to recreate lost output.

Identity and offsets

First acceptance of streaming POST /exec sets CE-Process-Id before the body and emits a started frame. Synchronous exec, process creation and keyed replay do not promise that header. Process-output GET and setup logs never emit started. Save proc_id and per-stream byte offsets before reconnecting.

Offsets are the byte position at the end of a stdout or stderr chunk. Base64 represents bytes that are not valid UTF-8. Resume with both saved offsets; never use character counts. Retained output is at most 10,000,000 raw bytes per process across both streams, evicting oldest complete chunks without killing or blocking the process. It expires one hour after exit; synchronous truncation remains a separate one-megabyte-per-stream limit.

Missing output is a visible condition

An offset before retained data returns 409 output_gap with both earliest offsets. An offset beyond produced output returns 400. If a consumer falls behind after headers, the server emits a gap frame then closes without an exit frame. Expired output returns 410 output_expired. The client must explicitly choose whether to skip lost bytes; silently joining across a gap corrupts the transcript.

Setup and completion

Setup merges stderr into stdout before assigning its combined-stream offset. Followed output sends a heartbeat every 15 seconds and exit when finished. follow=false is a finite current-output read; it does not fabricate successful exit for an unfinished setup. Background execution may survive disconnect and token expiry, but lifecycle changes and process deadlines remain independent. Close only the output iterator to reconnect; killing the process is a separate explicit action.

Complete retained-output programs

Set CE_COMPUTER_ID to a running computer reserved for this example. Both programs start one bounded process with a fresh idempotency key, replay that same start to confirm identity, consume separate byte offsets and explicitly kill the retained handle in cleanup. A transport reconnect reads from consumed offsets; a gap or expired output fails instead of silently restarting the command. The guest command uses sh and its printf builtin. The returned bytes include both NUL and a non-UTF-8 byte.

Python

import base64
import os
import uuid
from computeengines import Client, OutputExpired, OutputGap, StreamDisconnected

with Client() as client:
    computer = client.computers.get(os.environ["CE_COMPUTER_ID"])
    key = str(uuid.uuid4())
    command = ["sh", "-c", "printf 'hello\\000\\377'; printf 'notice' >&2"]
    started = computer.processes.start(command, timeout_seconds=30, idempotency_key=key)
    try:
        # Replaying the same start key identifies the same process, not new work.
        replay = computer.processes.start(command, timeout_seconds=30, idempotency_key=key)
        if replay.proc_id != started.proc_id:
            raise RuntimeError("Process replay changed identity")
        process = computer.processes.get(started.proc_id)
        offsets = {"stdout_offset": 0, "stderr_offset": 0}
        output = {"stdout": bytearray(), "stderr": bytearray()}
        finished = False
        for attempt in range(3):
            stream = process.iter_output(**offsets)
            try:
                for frame in stream:
                    if frame.type in output:
                        data = base64.b64decode(frame.data, validate=True) if frame.encoding == "base64" else frame.data.encode()
                        output[frame.type].extend(data)
                    offsets = {"stdout_offset": stream.stdout_offset, "stderr_offset": stream.stderr_offset}
                    if frame.type == "exit":
                        if frame.exit_code != 0 or frame.timed_out:
                            raise RuntimeError("Guest process failed")
                        finished = True
                if finished:
                    break
            except StreamDisconnected:
                # Reopen retained output from consumed offsets, never start again.
                offsets = {"stdout_offset": stream.stdout_offset, "stderr_offset": stream.stderr_offset}
                if attempt == 2:
                    raise
            except (OutputGap, OutputExpired):
                # Missing bytes require a caller decision; do not reset to zero.
                raise
            finally:
                stream.close()
        if not finished or output != {"stdout": b"hello\x00\xff", "stderr": b"notice"}:
            raise RuntimeError("Output was incomplete or different")
        print("Retained bytes and replay identity verified")
    finally:
        computer.processes.kill(started.proc_id)

TypeScript

import { randomUUID } from "node:crypto";
import { Buffer } from "node:buffer";
import { Client, StreamDisconnected } from "@computeengines/agent-computers";

const id = process.env.CE_COMPUTER_ID;
if (!id) throw new Error("Set CE_COMPUTER_ID");
const client = new Client();
try {
  const computer = await client.computers.get(id);
  const command = ["sh", "-c", "printf 'hello\\000\\377'; printf 'notice' >&2"];
  const options = { timeout_seconds: 30, idempotency_key: randomUUID() };
  const started = await computer.processes.start(command, options);
  try {
    if ((await computer.processes.start(command, options)).proc_id !== started.proc_id) throw new Error("Replay identity changed");
    const handle = await computer.processes.get(started.proc_id);
    let offsets = { stdout_offset: 0, stderr_offset: 0 };
    /** @type {Buffer[]} */
    const stdout = [];
    /** @type {Buffer[]} */
    const stderr = [];
    let finished = false;
    for (let attempt = 0; attempt < 3; attempt++) {
      const stream = handle.iterOutput(offsets);
      try {
        for await (const frame of stream) {
          if (frame.type === "stdout" || frame.type === "stderr") {
            const bytes = Buffer.from(frame.data, frame.encoding === "base64" ? "base64" : "utf8");
            (frame.type === "stdout" ? stdout : stderr).push(bytes);
          }
          offsets = { stdout_offset: stream.stdout_offset, stderr_offset: stream.stderr_offset };
          if (frame.type === "exit") {
            if (frame.exit_code !== 0 || frame.timed_out) throw new Error("Guest process failed");
            finished = true;
          }
        }
        if (finished) break;
      } catch (error) {
        offsets = { stdout_offset: stream.stdout_offset, stderr_offset: stream.stderr_offset };
        // OutputGap/OutputExpired and other errors are not reconnect permission.
        if (!(error instanceof StreamDisconnected) || attempt === 2) throw error;
      } finally { await stream.close(); }
    }
    if (!finished || !Buffer.concat(stdout).equals(Buffer.from([104, 101, 108, 108, 111, 0, 255]))
        || Buffer.concat(stderr).toString() !== "notice") throw new Error("Incomplete retained output");
    console.log("Retained bytes and replay identity verified");
  } finally { await computer.processes.kill(started.proc_id); }
} finally { client.close(); }

Dory is in development; these pages describe the planned service and may change. Questions: support@computeengines.com.