rea/src/process/OwnedCommand.ts

import type { Readable, Writable } from "node:stream";
import { pipeline } from "node:stream/promises";

import type {
  OwnedProviderProcessSpawnOptions,
  ProviderProcessSnapshot,
  ProviderProcessStopResult,
  SpawnedOwnedProviderProcess,
} from "./ProviderProcess.js";
import {
  ProviderProcessSupervisor,
  spawnOwnedProviderProcess,
} from "./ProviderProcess.js";
import {
  cleanupOwnedProcessGroup,
  type OwnedProcessGroup,
  type ProcessCleanupFailure,
} from "./ProcessOwnership.js";
import { ABORTED, waitForAbortable } from "./AbortablePromise.js";
import {
  ProviderStartupDeadline,
  waitForAbortableDelay,
} from "./ProviderDeadline.js";

/** A short-lived command failed its lifecycle or complete-output contract. */
export class OwnedCommandFailure extends Error {
  readonly cleanupResult: ProcessCleanupFailure | undefined;
  readonly collectionFailure: boolean;

  constructor(
    readonly reason: "cancelled" | "timeout" | "output-limit" | "process",
    message: string,
    readonly snapshot: ProviderProcessSnapshot | null = null,
    readonly cleanupFailure: string | null = null,
    options?: ErrorOptions & {
      readonly cleanupResult?: ProcessCleanupFailure;
      readonly collectionFailure?: boolean;
    },
    readonly resources: readonly string[] = [],
    readonly cleanupOwner?: ProviderProcessSupervisor,
  ) {
    super(message, options);
    this.cleanupResult = options?.cleanupResult;
    this.collectionFailure = options?.collectionFailure ?? false;
  }
}

/** Report recorded launch identities; these labels do not authorize signaling. */
export const ownedCommandResources = (
  ownership: OwnedProcessGroup,
  directory?: string,
): readonly string[] => [
  ownership.runId,
  ...(directory === undefined ? [] : [directory]),
  `pid:${ownership.leaderPid}`,
  `process-group:${ownership.processGroupId}`,
];

/** Output handling for one owned command. */
interface OwnedCommandLimits {
  /** Retained stderr bytes, plus stdout bytes unless a sink consumes stdout. */
  readonly diagnosticBytes: number;
  /** Accept an ordinary nonzero exit after output closes without other process errors. */
  readonly acceptNonZeroExit?: boolean;
  /**
   * Receives stdout instead of the retained snapshot, for output parsed as it
   * arrives. The command completes only after the sink finishes. The sink
   * records its own rejections and keeps accepting, so the command's outcome
   * stays authoritative; a failing stream is a collection failure.
   */
  readonly stdout?: Writable;
}

/** Execute a single owned process with independent cleanup and bounded diagnostic retention. */
export const runOwnedCommand = async (
  spawn: OwnedProviderProcessSpawnOptions,
  limits: OwnedCommandLimits & { readonly timeoutMs: number },
  options: {
    readonly signal?: AbortSignal | undefined;
    /** Synchronously accept unresolved cleanup ownership; execution still returns its original outcome. Must not throw. */
    readonly retainCleanup?: (
      owner: ProviderProcessSupervisor,
      result: Extract<ProviderProcessStopResult, { status: "incomplete" }>,
    ) => void;
    readonly launcher?: (
      input: OwnedProviderProcessSpawnOptions,
    ) => Promise<SpawnedOwnedProviderProcess>;
  } = {},
): Promise<ProviderProcessSnapshot> => {
  const callerSignal = options.signal ?? spawn.signal;
  if (callerSignal?.aborted)
    throw new OwnedCommandFailure(
      "cancelled",
      "Command cancelled before launch.",
    );
  const deadline = new ProviderStartupDeadline(limits.timeoutMs, callerSignal);
  try {
    return await collectOwnedCommand(
      spawn,
      limits,
      options.launcher ?? spawnOwnedProviderProcess,
      deadline,
      callerSignal,
      options.retainCleanup,
    );
  } finally {
    deadline.dispose();
  }
};

const collectOwnedCommand = async (
  spawn: OwnedProviderProcessSpawnOptions,
  limits: OwnedCommandLimits,
  launcher: (
    input: OwnedProviderProcessSpawnOptions,
  ) => Promise<SpawnedOwnedProviderProcess>,
  deadline: ProviderStartupDeadline,
  callerSignal?: AbortSignal,
  retainCleanup?: (
    owner: ProviderProcessSupervisor,
    result: Extract<ProviderProcessStopResult, { status: "incomplete" }>,
  ) => void,
): Promise<ProviderProcessSnapshot> => {
  const signal = deadline.signal;
  let launched: SpawnedOwnedProviderProcess;
  try {
    launched = await launcher({
      ...spawn,
      signal,
    });
  } catch (cause: unknown) {
    const interruption = deadline.interruption;
    if (interruption !== undefined)
      throw new OwnedCommandFailure(
        interruption,
        interruption === "cancelled"
          ? "Command cancelled during ownership preparation."
          : "Command deadline elapsed during ownership preparation.",
        null,
        null,
        { cause },
      );
    throw cause;
  }
  let exceeded = false;
  let processFailure: string | undefined;
  const supervisor = new ProviderProcessSupervisor(
    {
      ...launched,
      ownsProcessLifetime: true,
      cleanup:
        launched.cleanup ??
        (() => cleanupOwnedProcessGroup(launched.ownership)),
    },
    {
      captureStdout: limits.stdout === undefined,
      maxDiagnosticBytes: limits.diagnosticBytes,
      onDiagnostic: (event) => {
        if (event.type === "output" && event.truncated === true)
          exceeded = true;
        if (event.type === "error") processFailure = event.message;
      },
    },
  );
  // A command that fails or is interrupted destroys its sink with its stdout.
  const abandoned = new AbortController();
  const consumed = consumeStdout(
    launched.process.stdout,
    limits.stdout,
    AbortSignal.any([signal, abandoned.signal]),
  );
  void consumed.then((failed) => {
    processFailure ??= failed;
  });
  let failure: OwnedCommandFailure | undefined;
  try {
    while (!(await supervisor.waitForOutputClose(10))) {
      const interruption = deadline.interruption;
      if (interruption !== undefined)
        throw new OwnedCommandFailure(
          interruption,
          interruption === "cancelled"
            ? "Command cancelled."
            : "Command deadline elapsed.",
        );
      if (exceeded)
        throw new OwnedCommandFailure(
          "output-limit",
          "Command diagnostic output exceeded its complete-output budget.",
        );
      if (processFailure !== undefined)
        throw new OwnedCommandFailure("process", processFailure, null, null, {
          collectionFailure: true,
        });
      await waitForAbortableDelay(25, signal);
    }
    // Output has closed, but the sink may still be parsing what it buffered.
    const sinkFailure = await waitForAbortable(consumed, signal);
    if (sinkFailure !== ABORTED) processFailure ??= sinkFailure;
    const snapshot = supervisor.snapshot();
    const interruption = deadline.interruption;
    if (interruption !== undefined)
      throw new OwnedCommandFailure(
        interruption,
        interruption === "cancelled"
          ? "Command cancelled."
          : "Command deadline elapsed.",
        snapshot,
      );
    if (exceeded)
      throw new OwnedCommandFailure(
        "output-limit",
        "Command diagnostic output exceeded its complete-output budget.",
        snapshot,
      );
    const acceptedNonzeroExit =
      limits.acceptNonZeroExit === true &&
      snapshot.exitCode !== null &&
      snapshot.exitCode !== undefined &&
      snapshot.exitCode !== 0 &&
      snapshot.signal === null;
    if (
      processFailure !== undefined ||
      ((snapshot.exitCode !== 0 || snapshot.signal !== null) &&
        !acceptedNonzeroExit)
    )
      throw new OwnedCommandFailure(
        "process",
        processFailure ??
          "Command failed; inspect its exit status and diagnostics.",
        snapshot,
        null,
        { collectionFailure: processFailure !== undefined },
      );
  } catch (cause: unknown) {
    failure =
      cause instanceof OwnedCommandFailure
        ? cause
        : new OwnedCommandFailure(
            "process",
            "Command collection failed.",
            null,
            null,
            { cause },
          );
  }
  // A finished sink ignores this; an unfinished one is destroyed.
  abandoned.abort();
  const stopped = await supervisor.stop();
  const snapshot = supervisor.snapshot();
  if (stopped.status === "incomplete" && retainCleanup !== undefined)
    retainCleanup(supervisor, stopped);
  else if (stopped.status === "incomplete")
    throw new OwnedCommandFailure(
      failure?.reason ?? "process",
      failure?.message ?? "Command cleanup failed.",
      snapshot,
      stopped.reason,
      {
        ...(failure === undefined ? {} : { cause: failure }),
        ...(stopped.cleanupResult === undefined
          ? {}
          : { cleanupResult: stopped.cleanupResult }),
        ...(failure?.collectionFailure === true
          ? { collectionFailure: true }
          : {}),
      },
      ownedCommandResources(launched.ownership, spawn.cwd),
      supervisor,
    );
  if (failure !== undefined)
    throw new OwnedCommandFailure(
      failure.reason,
      failure.message,
      snapshot,
      failure.cleanupFailure,
      {
        cause: failure,
        ...(failure.cleanupResult === undefined
          ? {}
          : { cleanupResult: failure.cleanupResult }),
        ...(failure.collectionFailure ? { collectionFailure: true } : {}),
      },
      failure.resources,
    );
  if (callerSignal?.aborted)
    throw new OwnedCommandFailure("cancelled", "Command cancelled.", snapshot);
  return snapshot;
};

/**
 * Pipe stdout into its sink until `signal` aborts, which destroys both; settles
 * with a failure message, never rejects.
 */
const consumeStdout = (
  stdout: Readable | null,
  sink: Writable | undefined,
  signal: AbortSignal,
): Promise<string | undefined> => {
  if (sink === undefined) return Promise.resolve(undefined);
  if (stdout === null) {
    sink.destroy();
    return Promise.resolve("stdout stream is unavailable");
  }
  return pipeline(stdout, sink, { signal }).then(
    () => undefined,
    (cause: unknown) =>
      `stdout stream failed: ${cause instanceof Error ? cause.message : String(cause)}`,
  );
};