Method startPipe

packages/runtime/src/shell-run-manager.ts:726–803

(
    input: ShellRunBashInput,
    shellRunId: string,
    timeoutMs: number | undefined,
    forwardLive: boolean,
    slotReservation: ShellRunSlotReservation,
    sessionEpoch: number,
    onLiveAdmission: ((live: LiveShellRun) => void) | undefined,
  )

  }

  private async startPipe(
    input: ShellRunBashInput,
    shellRunId: string,
    timeoutMs: number | undefined,
    forwardLive: boolean,
    slotReservation: ShellRunSlotReservation,
    sessionEpoch: number,
    onLiveAdmission: ((live: LiveShellRun) => void) | undefined,
  ): Promise<LivePipeShellRun> {
    const collector = new PipeTailCollector(this.maxRetainedChars);
    const pending: Array<(live: LivePipeShellRun) => void> = [];
    let live: LivePipeShellRun | undefined;
    let startingRecord: ShellRunRecord | undefined;
    let spawnAttempted = false;
    const dispatch = (callback: (target: LivePipeShellRun) => void): void => {
      if (live) callback(live);
      else pending.push(callback);
    };
    try {
      const plan = input.argv
        ? {
            file: requireProgram(input.argv),
            args: [...input.argv.slice(1)],
            useShellOption: false,
          }
        : buildShellSpawnPlan(
            input.shell ?? defaultShellPlan(),
            input.command,
            input.env ?? process.env,
          );
      startingRecord = await this.createStartingRecord(
        input,
        shellRunId,
        timeoutMs,
        collector.snapshot(),
      );
      this.assertStartupAllowed(input.sessionId, sessionEpoch, input.abortSignal);
      spawnAttempted = true;
      const driver = new PipeProcessDriver({
        plan,
        cwd: input.cwd,
        ...((plan.env ?? input.env) ? { env: plan.env ?? input.env } : {}),
        ...(input.fdInputs ? { fdInputs: input.fdInputs } : {}),
        outputDrainMs: this.pipeOutputDrainMs,
        onData: (stream, data) => dispatch((target) => this.onPipeData(target, stream, data)),
        onRootExit: () => dispatch((target) => this.onNativeRootExit(target)),
        onExit: (exit) =>
          dispatch((target) => this.onDriverExit(target, { mode: 'pipes', value: exit })),
        onFailure: (error) => dispatch((target) => this.handleIntegrityFailure(target, error)),
      });
      live = {
        ...this.createLiveBase(input, startingRecord, 'pipes', timeoutMs, slotReservation),
        mode: 'pipes',
        driver,
        collector,
        pendingFlushChars: 0,
        forwardLive,
        liveEmitted: { stdout: 0, stderr: 0 },