Function wrapHandle

packages/computer/src/shell.ts:245–289

(
  shell: ShellRPC,
  sync: Sync,
  id: string,
  wireEvents: ReadableStream<ExecEvent>,
  encoding: E | undefined,
  pushed: number,
)

// 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<E extends ExecEncoding>(
  shell: ShellRPC,
  sync: Sync,
  id: string,
  wireEvents: ReadableStream<ExecEvent>,
  encoding: E | undefined,
  pushed: number,
): ExecHandle<E> {
  const postPull = withPostPull(pipeEvents<E>(wireEvents, encoding), sync);
  const stream = postPull.stream;
  const handle = stream as ExecHandle<E>;
  let resultPromise: Promise<ExecResult<E>> | undefined;
  let resultReader: ReadableStreamDefaultReader<WorkspaceExecEvent<E>> | 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<E>(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<E extends ExecEncoding>(
  source: ReadableStream<ExecEvent>,