Function pipeEvents

packages/computer/src/shell.ts:291–370

(
  source: ReadableStream<ExecEvent>,
  encoding: E | undefined,
)

}

function pipeEvents<E extends ExecEncoding>(
  source: ReadableStream<ExecEvent>,
  encoding: E | undefined,
): ReadableStream<WorkspaceExecEvent<E>> {
  if (encoding !== "utf8") {
    // Identity pipe — the wire shape already matches.
    return source as unknown as ReadableStream<WorkspaceExecEvent<E>>;
  }
  // 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<WorkspaceExecEvent<E>>,
    event: WorkspaceExecEvent<E>,
  ) => {
    lastSeq = event.seq;
    controller.enqueue(event);
  };
  const flushPending = (
    controller: TransformStreamDefaultController<WorkspaceExecEvent<E>>,
    beforeSeq?: number,
  ) => {
    const pending: Array<{
      id: string;
      seq: number;
      name: "stdout" | "stderr";
      value: Chunk<E>;
    }> = [];
    const stdout = stdoutDec.decode();
    const stderr = stderrDec.decode();
    if (stdout && stdoutMeta) {
      pending.push({ ...stdoutMeta, name: "stdout", value: stdout as Chunk<E> });
    }
    if (stderr && stderrMeta) {
      pending.push({ ...stderrMeta, name: "stderr", value: stderr as Chunk<E> });
    }
    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<ExecEvent, WorkspaceExecEvent<E>>({
      transform(event, controller) {
        if (event.name === "stdout") {
          stdoutMeta = { id: event.id, seq: event.seq };
          enqueue(controller, {