char/avril

(wip) why'd you have to go and make things so complicaaaaated?

git clone https://git.t4t.associates/char/avril

Charlotte Somgenerate std typedefs from typescript impl38d7f6c

main
40.6 KiB1339 linesraw
1import {
2  ArrayPrototypePush as push,
3  Error,
4  FinalizationRegistry,
5  ObjectCreate,
6  ObjectDefineProperty as define,
7  ObjectFreeze as freeze,
8  ObjectGetOwnPropertyDescriptor,
9  RangeError,
10  ReflectApply as apply,
11  ReflectOwnKeys,
12  Set,
13  SetPrototypeAdd as retain,
14  SetPrototypeDelete as release,
15  String,
16  StringPrototypeSlice as slice,
17  SymbolIterator,
18  SymbolToStringTag,
19  TypedArrayPrototypeGetBuffer as buffer,
20  TypedArrayPrototypeGetByteLength as byteLength,
21  TypedArrayPrototypeGetByteOffset as byteOffset,
22  TypedArrayPrototypeGetTag as tag,
23  TypeError,
24  Uint8Array,
25  Uint8ArrayPrototypeSet as setBytes,
26  uncurryThis,
27  WeakMap,
28  WeakMapPrototypeSet as set,
29} from "ext:core/primordials";
30import {
31  markPromiseHandled,
32  promiseAll,
33  promiseResolve,
34  promiseThen as then,
35  safePromise,
36} from "ext:core/promises";
37import {
38  op_process_cancel_read,
39  op_process_end,
40  op_process_keepalive,
41  op_process_kill,
42  op_process_read,
43  op_process_ref,
44  op_process_release,
45  op_process_resize,
46  op_process_run_sync,
47  op_process_spawn,
48  op_process_stop,
49  op_process_wait,
50  op_process_write,
51} from "ext:core/ops";
52import {
53  array,
54  dictionary,
55  isObject,
56  required,
57  state,
58  usv,
59} from "ext:web/idl.js";
60import { type AbortSignal, addAbort, signals } from "ext:web/events.js";
61import { registerUncloneable } from "ext:web/serialization.js";
62import type { ReadableStreamDefaultReader } from "ext:webstreams/streams.js";
63import {
64  enqueue,
65  finish,
66  read,
67  ReadableStream,
68  readerCancel,
69  release as releaseReader,
70  streamLocked,
71  streamReader,
72  WritableStream,
73} from "ext:webstreams/streams.js";
74import { TextDecoder, TextEncoder } from "ext:encoding/main.js";
75import { args, path, processHandle } from "ext:file/main.js";
76import { addSignalListener, removeSignalListener } from "./signals.js";
77import { errorCode, failure, result } from "ext:file/errors.js";
78import type { URL } from "ext:url/main.js";
79import type { ProcessOptions, RunSyncOptions, SpawnOptions } from "avril:std";
80import type {
81  NativeOptions,
82  NativeStdio,
83  ProcessBytes,
84  ProcessOutputValue,
85  ProcessStatusValue,
86} from "./types.js";
87
88const Number = globalThis.Number,
89  safeInteger = Number.isSafeInteger,
90  min = Math.min;
91const decode = uncurryThis(TextDecoder.prototype.decode),
92  encode = uncurryThis(TextEncoder.prototype.encode),
93  encodingOf = uncurryThis(
94    ObjectGetOwnPropertyDescriptor(TextDecoder.prototype, "encoding")!.get!,
95  );
96const encoder = new TextEncoder(), utf8 = new TextDecoder();
97const parseJSON = JSON.parse,
98  indexOf = uncurryThis(String.prototype.indexOf);
99const writerOf = uncurryThis(WritableStream.prototype.getWriter),
100  write = uncurryThis(WritableStreamDefaultWriter.prototype.write),
101  closeWriter = uncurryThis(WritableStreamDefaultWriter.prototype.close),
102  releaseWriter = uncurryThis(
103    WritableStreamDefaultWriter.prototype.releaseLock,
104  );
105const asyncDispose: typeof Symbol.asyncDispose =
106  globalThis.Symbol.asyncDispose;
107const children = new WeakMap<ChildProcess, ProcessRecord>(),
108  live = new Set<ChildProcess>(),
109  collecting = new Set<ChildProcess>(),
110  secret = ObjectCreate(null) as Secret;
111registerUncloneable(children);
112const register = uncurryThis(FinalizationRegistry.prototype.register),
113  unregister = uncurryThis(FinalizationRegistry.prototype.unregister);
114const finalizer = new FinalizationRegistry((id: number) =>
115  op_process_release(id)
116);
117
118type ProcessRecord<T = ProcessBytes | string> = {
119  __proto__?: null;
120  id: number;
121  pid: number;
122  reads: Record<number, ReadSource | undefined>;
123  writes: Record<number, WriteSource | undefined>;
124  stdio: Record<number, ProcessStream>;
125  terminal: ProcessTerminal | null;
126  collectLimit: number;
127  decoder: TextDecoder | null;
128  decode: (bytes: ProcessBytes) => T;
129  encoding: string;
130  check: boolean;
131  automaticInput: boolean;
132  collected: boolean;
133  disposed: boolean;
134  finished: boolean;
135  referenced: boolean;
136  inputTask: Promise<void> | undefined;
137  inputFailed: boolean;
138  inputError: unknown;
139  abortReason: unknown;
140  status: Promise<ProcessStatusValue>;
141  completed: Promise<ProcessStatusValue>;
142};
143type ProcessStream =
144  | ReadableStream<ProcessBytes>
145  | WritableStream<ProcessBytes>;
146type ReadSource = {
147  __proto__?: null;
148  owner: ChildProcess | null;
149  used: boolean;
150  stream: ReadableStream<ProcessBytes>;
151};
152type WriteSource = {
153  __proto__?: null;
154  owner: ChildProcess | null;
155  stream: WritableStream<ProcessBytes>;
156};
157type ProcessTerminal = {
158  readable: ReadableStream<ProcessBytes>;
159  writable: WritableStream<ProcessBytes>;
160  resize(columns: number, rows: number): void;
161};
162type CollectionBudget = {
163  __proto__?: null;
164  limit: number;
165  used: number;
166  exceeded: boolean;
167};
168type Collection = {
169  __proto__?: null;
170  status: ProcessStatusValue;
171  stdout: ProcessBytes | null;
172  stderr: ProcessBytes | null;
173  cause: unknown;
174};
175type LaunchedStream = {
176  __proto__?: null;
177  record: ProcessRecord;
178  reader: ReadableStreamDefaultReader<ProcessBytes>;
179  settled: Promise<Collection>;
180};
181interface ProcessPromise<T = ProcessBytes | string>
182  extends Promise<ProcessOutputValue<T>> {
183  bytes(): Promise<ProcessBytes>;
184  text(): Promise<string>;
185  json(): Promise<unknown>;
186  lines(): AsyncIterableIterator<string>;
187}
188type Prepared<T = ProcessBytes | string> = {
189  __proto__?: null;
190  native: NativeOptions;
191  input: ProcessBytes | ReadableStream<ProcessBytes> | undefined;
192  stream: boolean;
193  check: boolean;
194  decoder: TextDecoder | null;
195  decode: (bytes: ProcessBytes) => T;
196  signal: unknown;
197  maxCollectedBytes: number;
198};
199
200function own(object: object, key: PropertyKey, value: unknown): void {
201  define(object, key, {
202    __proto__: null,
203    value,
204    writable: true,
205    configurable: true,
206    enumerable: true,
207  } as PropertyDescriptor);
208}
209class ProcessError extends Error {
210  declare static readonly [Symbol.hasInstance]: (value: unknown) => value is ProcessError;
211  declare readonly result: ProcessOutputValue | null;
212
213  constructor(
214    message: string,
215    output: ProcessOutputValue | null = null,
216    cause: unknown = undefined,
217  ) {
218    super(message);
219    own(this, "name", "ProcessError");
220    own(this, "result", output);
221    if (cause !== undefined) own(this, "cause", cause);
222  }
223}
224class OutputLimitError extends ProcessError {
225  declare static readonly [Symbol.hasInstance]: (value: unknown) => value is OutputLimitError;
226  constructor(output: ProcessOutputValue) {
227    super("Subprocess output exceeded its configured limit", output);
228    own(this, "name", "OutputLimitError");
229  }
230}
231class ProcessTimeoutError extends ProcessError {
232  declare static readonly [Symbol.hasInstance]: (value: unknown) => value is ProcessTimeoutError;
233  constructor(output: ProcessOutputValue) {
234    super("Subprocess timed out", output);
235    own(this, "name", "ProcessTimeoutError");
236  }
237}
238class ProcessAbortError extends ProcessError {
239  declare static readonly [Symbol.hasInstance]: (value: unknown) => value is ProcessAbortError;
240  constructor(output: ProcessOutputValue | null, cause?: unknown) {
241    super("Subprocess was aborted", output, cause);
242    own(this, "name", "ProcessAbortError");
243  }
244}
245function checked<T extends Uint8Array | string>(
246  output: ProcessOutputValue<T>,
247  check: boolean,
248  cause?: unknown,
249): ProcessOutputValue<T> {
250  if (output.reason === "output-limit") throw new OutputLimitError(output);
251  if (output.reason === "timeout") throw new ProcessTimeoutError(output);
252  if (output.reason === "abort") throw new ProcessAbortError(output, cause);
253  if (output.reason !== null || (check && !output.success)) {
254    throw new ProcessError(
255      output.error ||
256        (output.reason === null
257          ? `Subprocess exited with ${output.signal || output.code}`
258          : `Subprocess ${output.reason}`),
259      output,
260      cause,
261    );
262  }
263  return output;
264}
265function keys(
266  value: Record<PropertyKey, unknown>,
267  allowed: readonly PropertyKey[],
268): void {
269  const names = ReflectOwnKeys(value);
270  for (let i = 0; i < names.length; i++) {
271    let found = false;
272    for (let j = 0; j < allowed.length; j++) {
273      if (names[i] === allowed[j]) found = true;
274    }
275    if (!found) throw new TypeError("Unknown subprocess option");
276  }
277}
278function integer<F extends number | null | undefined>(
279  value: unknown,
280  fallback: F,
281  minimum = 0,
282  maximum = 2 ** 53 - 1,
283): number | Extract<F, null | undefined> {
284  if (value === undefined) {
285    return fallback as number | Extract<F, null | undefined>;
286  }
287  if (
288    typeof value !== "number" || !safeInteger(value) || value < minimum ||
289    value > maximum
290  ) throw new RangeError("Subprocess option is outside its integer range");
291  return value;
292}
293function boolean(value: unknown, fallback: boolean): boolean {
294  if (value === undefined) return fallback;
295  if (typeof value !== "boolean") {
296    throw new TypeError("Expected a boolean subprocess option");
297  }
298  return value;
299}
300function bytes(value: unknown, text = false): ProcessBytes {
301  if (text && typeof value === "string") return encode(encoder, value);
302  if (tag(value) !== "Uint8Array") {
303    throw new TypeError("Subprocess input requires string or Uint8Array bytes");
304  }
305  const copy = new Uint8Array(byteLength(value as Uint8Array<ArrayBufferLike>));
306  setBytes(copy, value as Uint8Array<ArrayBufferLike>);
307  return copy;
308}
309function sequence(value: unknown): string[] {
310  const output = array<string>();
311  if (value === undefined) return output;
312  if (!isObject(value)) {
313    throw new TypeError("Arguments must be an iterable of strings");
314  }
315  const method = value[SymbolIterator];
316  if (typeof method !== "function") {
317    throw new TypeError("Arguments must be iterable");
318  }
319  const iterator = apply(method, value, []) as { next: unknown },
320    next = iterator.next;
321  if (!isObject(iterator) || typeof next !== "function") {
322    throw new TypeError("Invalid argument iterator");
323  }
324  while (true) {
325    const step = apply(next, iterator, []);
326    if (!isObject(step)) {
327      throw new TypeError("Invalid argument iterator result");
328    }
329    if (step.done) return output;
330    push(output, usv(step.value));
331  }
332}
333function stdio(
334  value: unknown,
335  fallback: string,
336  input: boolean,
337  stderr = false,
338  extra = false,
339): NativeStdio {
340  const spec: NativeStdio = {
341    __proto__: null,
342    mode: fallback,
343    file: null,
344    fileMode: null,
345    handle: null,
346    direction: null,
347  };
348  if (typeof value === "string" || value === undefined) {
349    if (value !== undefined) spec.mode = value;
350  } else {
351    const options = dictionary(value);
352    keys(options, ["file", "mode", "handle", "pipe", "direction"]);
353    const filename = options.file,
354      mode = options.mode,
355      handle = options.handle,
356      pipe = options.pipe,
357      direction = options.direction;
358    if (extra) {
359      spec.direction = (pipe === undefined ? direction : pipe) as string;
360      input = spec.direction === "parent-to-child";
361    }
362    if (filename !== undefined) {
363      if (handle !== undefined || pipe !== undefined) {
364        throw new TypeError("Conflicting stdio options");
365      }
366      spec.mode = "file";
367      spec.file = path(filename);
368      if (input) {
369        if (mode !== undefined) {
370          throw new TypeError("Input redirection has no file mode");
371        }
372      } else spec.fileMode = mode === undefined ? "truncate" : usv(mode);
373    } else if (handle !== undefined) {
374      if (pipe !== undefined || mode !== undefined) {
375        throw new TypeError("Conflicting stdio handle options");
376      }
377      spec.mode = "handle";
378      spec.handle = processHandle(handle);
379    } else if (extra && pipe !== undefined) {
380      if (mode !== undefined || direction !== undefined) {
381        throw new TypeError("Conflicting pipe options");
382      }
383      spec.mode = "piped";
384    } else throw new TypeError("Expected a stdio mode, file, or handle");
385  }
386  if (
387    spec.mode !== "piped" && spec.mode !== "inherit" && spec.mode !== "null" &&
388    spec.mode !== "file" && spec.mode !== "handle" &&
389    !(stderr && spec.mode === "stdout")
390  ) throw new TypeError("Invalid stdio mode");
391  if (extra && spec.mode === "inherit") {
392    throw new TypeError("Extra stdio cannot inherit internal descriptors");
393  }
394  if (
395    extra && spec.direction !== "parent-to-child" &&
396    spec.direction !== "child-to-parent"
397  ) throw new TypeError("Extra stdio needs a pipe direction");
398  if (
399    spec.fileMode !== null && spec.fileMode !== "append" &&
400    spec.fileMode !== "truncate" && spec.fileMode !== "create-new"
401  ) throw new TypeError("Invalid stdio file mode");
402  return spec;
403}
404/** Splits `(args?, options?)`: an iterable second argument is the argument list. */
405function invocation(
406  second: unknown,
407  third: unknown,
408): { __proto__?: null; args: unknown; options: unknown } {
409  if (isObject(second) && typeof second[SymbolIterator] === "function") {
410    return { __proto__: null, args: second, options: third };
411  }
412  if (
413    typeof second === "string" || (second !== undefined && third !== undefined)
414  ) {
415    throw new TypeError("Arguments must be an iterable of strings");
416  }
417  return { __proto__: null, args: undefined, options: second ?? third };
418}
419function normalize(
420  command: unknown,
421  positional: unknown,
422  raw: unknown,
423  capture: boolean,
424  sync: boolean,
425): Prepared {
426  command = path(command);
427  const options = dictionary(raw);
428  keys(options, [
429    "args",
430    "cwd",
431    "searchPath",
432    "clearEnv",
433    "env",
434    "stdin",
435    "stdout",
436    "stderr",
437    "input",
438    "encoding",
439    "check",
440    "signal",
441    "timeoutMs",
442    "maxCollectedBytes",
443    "maxBufferedBytes",
444    "termination",
445    "drainTimeoutMs",
446    "extraStdio",
447    "terminal",
448  ]);
449  if (positional !== undefined && options.args !== undefined) {
450    throw new TypeError("Arguments were given both positionally and as args");
451  }
452  const argv = sequence(positional ?? options.args), cwdValue = options.cwd;
453  const cwd = cwdValue === undefined ? null : path(cwdValue);
454  const searchPath = boolean(options.searchPath, true),
455    clearEnv = boolean(options.clearEnv, false);
456  const environment = dictionary(options.env),
457    env = array<[string, string]>(),
458    names = ReflectOwnKeys(environment);
459  for (let i = 0; i < names.length; i++) {
460    const key = names[i];
461    if (typeof key !== "string") {
462      throw new TypeError("Environment names must be strings");
463    }
464    if (ObjectGetOwnPropertyDescriptor(environment, key)!.enumerable) {
465      push(env, args(key, usv(environment[key])));
466    }
467  }
468  let input: unknown = options.input, stream = false;
469  if (input !== undefined) {
470    if (typeof input !== "string" && tag(input) !== "Uint8Array") {
471      try {
472        if (streamLocked(input)) throw new TypeError("Input stream is locked");
473        stream = true;
474      } catch {
475        throw new TypeError(
476          "Input requires string, bytes, or an unlocked ReadableStream",
477        );
478      }
479      if (sync) throw new TypeError("runSync does not accept streaming input");
480    } else input = bytes(input, true);
481  }
482  const terminalValue = options.terminal;
483  let terminal: NativeOptions["terminal"] = null;
484  if (terminalValue !== undefined) {
485    if (capture || sync) throw new TypeError("Terminals require spawn");
486    const size = dictionary(terminalValue);
487    keys(size, ["columns", "rows"]);
488    terminal = {
489      __proto__: null,
490      columns: integer(size.columns, 80, 1, 65535),
491      rows: integer(size.rows, 24, 1, 65535),
492    };
493  }
494  const stdinValue = options.stdin,
495    stdoutValue = options.stdout,
496    stderrValue = options.stderr;
497  if (
498    input !== undefined && stdinValue !== undefined && stdinValue !== "piped"
499  ) throw new TypeError("input requires piped stdin");
500  const stdin = stdio(
501      stdinValue,
502      input !== undefined || terminal ? "piped" : "null",
503      true,
504    ),
505    stdout = stdio(stdoutValue, "piped", false),
506    stderr = stdio(stderrValue, terminal ? "stdout" : "piped", false, true);
507  if (capture && stdin.mode === "piped" && input === undefined) {
508    input = new Uint8Array(0);
509  }
510  const extra = dictionary(options.extraStdio),
511    extraStdio = array<[number, NativeStdio]>(),
512    descriptors = ReflectOwnKeys(extra);
513  for (let i = 0; i < descriptors.length; i++) {
514    const key = descriptors[i];
515    if (typeof key !== "string") {
516      throw new TypeError("Extra stdio descriptor must be an integer");
517    }
518    const fd = integer(Number(key), undefined, 3, 2147483615)!;
519    if (String(fd) !== key) {
520      throw new TypeError("Noncanonical stdio descriptor");
521    }
522    const value = dictionary(extra[key]);
523    push(extraStdio, args(fd, stdio(value, "piped", false, false, true)));
524  }
525  if ((capture || sync) && extraStdio.length) {
526    throw new TypeError("Additional streams require spawn");
527  }
528  const termination = dictionary(options.termination);
529  keys(termination, ["scope", "signal", "graceMs"]);
530  const scopeValue = termination.scope, signalValue = termination.signal;
531  const scope = scopeValue === undefined ? "group" : usv(scopeValue),
532    signalName = signalValue === undefined ? "SIGTERM" : usv(signalValue),
533    graceMs = integer(termination.graceMs, 1000);
534  if (scope !== "group" && scope !== "child") {
535    throw new TypeError("Invalid termination scope");
536  }
537  const timeoutMs = integer(options.timeoutMs, null),
538    maxCollectedBytes = integer(
539      options.maxCollectedBytes,
540      128 * 1024 * 1024,
541      1,
542    ),
543    maxBufferedBytes = integer(options.maxBufferedBytes, 1024 * 1024, 1),
544    drainTimeoutMs = integer(options.drainTimeoutMs, 1000);
545  const check = boolean(options.check, true), encodingValue = options.encoding;
546  const encoding = encodingValue === undefined ? null : usv(encodingValue);
547  const decoder = encoding === null ? null : new TextDecoder(encoding);
548  const signal = options.signal;
549  if (signal !== undefined) {
550    if (sync) throw new TypeError("runSync does not accept AbortSignal");
551    const aborted = state(signals, signal);
552    if (aborted.aborted) throw new ProcessAbortError(null, aborted.reason);
553  }
554  if (stream && streamLocked(input)) {
555    throw new TypeError("Input stream is locked");
556  }
557  // run() always has a reader draining its output, collecting it or
558  // streaming lines, so the child waits for that reader rather than racing
559  // the unread limit.
560  const backpressure = array<number>();
561  if (capture && !sync) {
562    if (stdout.mode === "piped") push(backpressure, 1);
563    if (stderr.mode === "piped") push(backpressure, 2);
564  }
565  return {
566    __proto__: null,
567    native: {
568      __proto__: null,
569      command: command as string,
570      args: argv,
571      cwd,
572      searchPath,
573      clearEnv,
574      env,
575      stdin,
576      stdout,
577      stderr,
578      extraStdio,
579      terminal,
580      timeoutMs,
581      maxCollectedBytes: sync ? maxCollectedBytes : null,
582      maxBufferedBytes,
583      termination: { __proto__: null, scope, signal: signalName, graceMs },
584      drainTimeoutMs,
585      backpressure,
586    },
587    input: input as Prepared["input"],
588    stream,
589    check,
590    decoder,
591    decode: (bytes) => decoder === null ? bytes : decode(decoder, bytes),
592    signal,
593    maxCollectedBytes: maxCollectedBytes as number,
594  };
595}
596function keepAlive(record: ProcessRecord, epoch: number): void {
597  const pending = then(op_process_keepalive(args(record.id, epoch)), result);
598  markPromiseHandled(pending);
599}
600function readable(record: ProcessRecord, fd: number): ReadSource {
601  const source: ReadSource = {
602    __proto__: null,
603    owner: null,
604    used: false,
605    stream: null!,
606  };
607  source.stream = new ReadableStream<ProcessBytes>({
608    __proto__: null,
609    type: "bytes",
610    async pull(controller) {
611      source.used = true;
612      if (record.disposed) {
613        source.owner = null;
614        finish(controller);
615        return;
616      }
617      const chunk = result(
618        await safePromise(op_process_read(args(record.id, fd))),
619      );
620      if (chunk === null) {
621        source.owner = null;
622        finish(controller);
623        return;
624      }
625      enqueue(controller, chunk);
626    },
627    cancel() {
628      source.used = true;
629      source.owner = null;
630      if (!record.disposed) result(op_process_cancel_read(args(record.id, fd)));
631    },
632  }, { __proto__: null, highWaterMark: 0 });
633  return source;
634}
635function writable(record: ProcessRecord, fd: number): WriteSource {
636  const source: WriteSource = { __proto__: null, owner: null, stream: null! };
637  source.stream = new WritableStream<ProcessBytes>({
638    __proto__: null,
639    write(value) {
640      if (record.disposed) {
641        throw failure("BadResource", "Subprocess is disposed");
642      }
643      return then(op_process_write(args(record.id, fd, bytes(value))), result);
644    },
645    close() {
646      source.owner = null;
647      if (!record.disposed) {
648        return then(op_process_end(args(record.id, fd)), result);
649      }
650    },
651    abort() {
652      source.owner = null;
653      if (!record.disposed) {
654        return then(op_process_end(args(record.id, fd)), result);
655      }
656    },
657  }, {
658    __proto__: null,
659    highWaterMark: 65536,
660    size(value) {
661      return tag(value) === "Uint8Array" ? byteLength(value) : 1;
662    },
663  });
664  return source;
665}
666async function feed(
667  record: ProcessRecord,
668  value: ProcessBytes | ReadableStream<ProcessBytes>,
669  streaming: boolean,
670): Promise<void> {
671  const writer = writerOf(record.writes[0]!.stream);
672  let reader: ReadableStreamDefaultReader<ProcessBytes> | null = null,
673    writing = false;
674  const cancellation = then(record.completed, () => {
675    if (reader) markPromiseHandled(readerCancel(reader));
676  });
677  markPromiseHandled(cancellation);
678  try {
679    if (streaming) reader = streamReader(value as ReadableStream<ProcessBytes>);
680    if (reader) {
681      while (true) {
682        writing = false;
683        const step = await safePromise(read(reader));
684        if (step.done) break;
685        writing = true;
686        await safePromise(write(writer, bytes(step.value)));
687      }
688    } else if (byteLength(value)) {
689      writing = true;
690      await safePromise(write(writer, value));
691    }
692    writing = true;
693    await safePromise(closeWriter(writer));
694  } catch (error) {
695    if (!writing || errorCode(error) !== "BrokenPipe") {
696      record.inputFailed = true;
697      record.inputError = error;
698      if (!record.finished && !record.disposed) {
699        result(op_process_stop(args(record.id, "io-error")));
700      }
701    }
702  } finally {
703    if (reader) {
704      markPromiseHandled(readerCancel(reader));
705      releaseReader(reader);
706    }
707    releaseWriter(writer);
708  }
709}
710async function collect(
711  source: ReadSource | undefined,
712  budget: CollectionBudget,
713  record: ProcessRecord,
714): Promise<ProcessBytes | null> {
715  if (!source) return null;
716  const reader = streamReader(source.stream), chunks = array<ProcessBytes>();
717  let size = 0;
718  try {
719    while (true) {
720      const step = await safePromise(read(reader));
721      if (step.done) break;
722      const count = byteLength(step.value as ProcessBytes),
723        allowed = min(count, budget.limit - budget.used);
724      if (allowed) {
725        push(
726          chunks,
727          allowed === count ? step.value as ProcessBytes : new Uint8Array(
728            buffer(step.value as ProcessBytes),
729            byteOffset(step.value as ProcessBytes),
730            allowed,
731          ),
732        );
733        size += allowed;
734        budget.used += allowed;
735      }
736      if (allowed < count && !budget.exceeded) {
737        budget.exceeded = true;
738        if (!record.disposed) {
739          result(op_process_stop(args(record.id, "output-limit")));
740        }
741      }
742    }
743  } finally {
744    releaseReader(reader);
745  }
746  const bytes = new Uint8Array(size);
747  let offset = 0;
748  for (let i = 0; i < chunks.length; i++) {
749    setBytes(bytes, chunks[i]!, offset);
750    offset += byteLength(chunks[i]!);
751  }
752  return bytes;
753}
754function output<T>(
755  status: ProcessStatusValue,
756  stdout: ProcessBytes | null,
757  stderr: ProcessBytes | null,
758  decodeOutput: (bytes: ProcessBytes) => T,
759): ProcessOutputValue<T> {
760  return freeze({
761    __proto__: null,
762    code: status.code,
763    signal: status.signal,
764    success: status.success,
765    reason: status.reason,
766    error: status.error,
767    stdout: stdout === null ? null : decodeOutput(stdout),
768    stderr: stderr === null ? null : decodeOutput(stderr),
769  });
770}
771function claim(record: ProcessRecord): CollectionBudget {
772  if (record.collected) {
773    throw new TypeError("Subprocess output can only be collected once");
774  }
775  for (let fd = 1; fd <= 2; fd++) {
776    const source = record.reads[fd];
777    if (source && (source.used || streamLocked(source.stream))) {
778      throw new TypeError("Subprocess output was already consumed or locked");
779    }
780  }
781  record.collected = true;
782  return {
783    __proto__: null,
784    limit: record.collectLimit,
785    used: 0,
786    exceeded: false,
787  };
788}
789/** Settles a collection: the status reflects output and input failures. */
790function settle(
791  record: ProcessRecord,
792  budget: CollectionBudget,
793  values: [ProcessBytes | null, ProcessBytes | null, ProcessStatusValue, void],
794): Collection {
795  const status: ProcessStatusValue & { __proto__?: null } = budget.exceeded
796    ? { __proto__: null, ...values[2], success: false, reason: "output-limit" }
797    : record.inputFailed && values[2].reason === null
798    ? {
799      __proto__: null,
800      ...values[2],
801      success: false,
802      reason: "io-error",
803      error: "Subprocess input failed",
804    }
805    : values[2];
806  return {
807    __proto__: null,
808    status,
809    stdout: values[0],
810    stderr: values[1],
811    cause: status.reason === "abort" ? record.abortReason : record.inputError,
812  };
813}
814/** Collects raw output and status, unchecked. */
815function gather(
816  child: ChildProcess,
817  record: ProcessRecord,
818): Promise<Collection> {
819  const budget = claim(record);
820  retain(collecting, child);
821  const pending = then(
822    promiseAll(
823      [
824        collect(record.reads[1], budget, record),
825        collect(record.reads[2], budget, record),
826        record.completed,
827        record.inputTask,
828      ] as const,
829    ),
830    (values) => settle(record, budget, values),
831  );
832  return then(pending, (value) => {
833    release(collecting, child);
834    return value;
835  }, (error) => {
836    release(collecting, child);
837    throw error;
838  });
839}
840/**
841 * Collects like gather, but leaves stdout to lines(), which reads it as the
842 * child writes. Settles independently of iteration.
843 */
844function stream(child: ChildProcess, record: ProcessRecord): LaunchedStream {
845  const budget = claim(record);
846  retain(collecting, child);
847  const reader = streamReader(record.reads[1]!.stream);
848  const pending = then(
849    promiseAll(
850      [
851        null,
852        collect(record.reads[2], budget, record),
853        record.completed,
854        record.inputTask,
855      ] as const,
856    ),
857    (values) => settle(record, budget, values),
858  );
859  const settled = then(pending, (value) => {
860    release(collecting, child);
861    return value;
862  }, (error) => {
863    release(collecting, child);
864    throw error;
865  });
866  return { __proto__: null, record, reader, settled };
867}
868async function* streamLines(launched: LaunchedStream): AsyncGenerator<string> {
869  const record = launched.record, reader = launched.reader;
870  const decoder = new TextDecoder(record.encoding),
871    options = { __proto__: null, stream: true };
872  let pending = "", finished = false, released = false, halted = false;
873  // A stopped run (timeout, abort, termination) ends the iteration without
874  // the lines still queued behind the reader.
875  const halt = () => {
876    halted = true;
877    if (!released) markPromiseHandled(readerCancel(reader));
878  };
879  markPromiseHandled(then(launched.settled, (raw) => {
880    if (raw.status.reason !== null) halt();
881  }, halt));
882  const line = (end: number) => {
883    const text = slice(pending, 0, end);
884    pending = slice(pending, end + 1);
885    return unterminated(text);
886  };
887  try {
888    lines: while (true) {
889      const step = await safePromise(read(reader));
890      if (halted) break;
891      if (step.done) {
892        pending += decode(decoder);
893        if (pending !== "") yield line(pending.length);
894        break;
895      }
896      pending += decode(decoder, step.value as ProcessBytes, options);
897      for (let end; (end = indexOf(pending, "\n")) !== -1;) {
898        if (halted) break lines;
899        yield line(end);
900      }
901    }
902    finished = true;
903  } finally {
904    if (!halted) markPromiseHandled(readerCancel(reader));
905    if (!finished && !record.finished && !record.disposed) {
906      result(op_process_stop(args(record.id, "terminated")));
907    }
908    released = true;
909    releaseReader(reader);
910  }
911  const raw = await safePromise(launched.settled);
912  checked(
913    output(raw.status, null, raw.stderr, record.decode),
914    record.check,
915    raw.cause,
916  );
917}
918function method<
919  T extends object,
920  K extends PropertyKey,
921  V extends () => unknown,
922>(
923  object: T,
924  name: K,
925  value: V,
926): asserts object is T & Record<K, V> {
927  define(object, name, {
928    __proto__: null,
929    value,
930    writable: true,
931    configurable: true,
932  } as PropertyDescriptor);
933}
934function split(text: string): string[] {
935  const lines = array<string>();
936  let start = 0;
937  for (let end; (end = indexOf(text, "\n", start)) !== -1; start = end + 1) {
938    push(lines, unterminated(slice(text, start, end)));
939  }
940  if (start < text.length) push(lines, unterminated(slice(text, start)));
941  return lines;
942}
943function unterminated(line: string): string {
944  return line[line.length - 1] === "\r" ? slice(line, 0, -1) : line;
945}
946/**
947 * The run() promise, with methods reading stdout from the same result.
948 * Collection starts on the next microtask unless lines() claims stdout
949 * first, in which case it streams.
950 */
951function invoke<T extends ProcessBytes | string>(
952  child: ChildProcess<T> | null,
953  failed: boolean,
954  failure: unknown,
955): ProcessPromise<T> {
956  const record = failed ? null : childRecord(child!);
957  let streamed: LaunchedStream | null = null;
958  const raw = then(promiseResolve(undefined), () => {
959    if (failed) throw failure;
960    return streamed ? streamed.settled : gather(child!, record!);
961  });
962  const settled = then(raw, (raw) => ({
963    __proto__: null,
964    output: checked(
965      output(raw.status, raw.stdout, raw.stderr, record!.decode),
966      record!.check,
967      raw.cause,
968    ),
969    stdout: raw.stdout,
970  }));
971  markPromiseHandled(settled);
972  const promise = then(settled, (value) => value.output);
973  const stdout = () => {
974    markPromiseHandled(promise);
975    return then(settled, (value) => {
976      if (value.stdout === null) {
977        throw new TypeError(
978          streamed
979            ? "Subprocess stdout was streamed by lines()"
980            : "Subprocess stdout is not piped",
981        );
982      }
983      return value.stdout;
984    });
985  };
986  const decodeText = (bytes: ProcessBytes) =>
987    decode(record!.decoder ?? utf8, bytes);
988  method(promise, "bytes", function bytes() {
989    return stdout();
990  });
991  method(promise, "text", function text() {
992    return then(stdout(), decodeText);
993  });
994  method(promise, "json", function json() {
995    return then(stdout(), (bytes) => parseJSON(decodeText(bytes)));
996  });
997  method(promise, "lines", function lines() {
998    if (failed) {
999      markPromiseHandled(promise);
1000      return (async function* () {
1001        throw failure;
1002      })();
1003    }
1004    if (streamed) throw new TypeError("Subprocess lines can only be read once");
1005    if (record!.collected) {
1006      // Collection already began: split the buffered result instead.
1007      const pending = stdout();
1008      markPromiseHandled(pending);
1009      return (async function* () {
1010        const lines = split(decodeText(await safePromise(pending)));
1011        for (let i = 0; i < lines.length; i++) yield lines[i];
1012      })();
1013    }
1014    if (!record!.reads[1]) {
1015      throw new TypeError("Subprocess stdout is not piped");
1016    }
1017    streamed = stream(child!, record!);
1018    markPromiseHandled(promise);
1019    return streamLines(streamed);
1020  });
1021  return promise;
1022}
1023declare const secretBrand: unique symbol;
1024type Secret = { readonly [secretBrand]: true };
1025// Keep the output type tied to its private record across the shared resource table.
1026let childRecord: <T extends ProcessBytes | string>(
1027  child: ChildProcess<T>,
1028) => ProcessRecord<T>;
1029class ChildProcess<T extends ProcessBytes | string = ProcessBytes | string> {
1030  declare static readonly [Symbol.hasInstance]: (value: unknown) => value is ChildProcess;
1031  #record: ProcessRecord<T>;
1032
1033  static {
1034    childRecord = (child) => {
1035      state(children, child);
1036      return child.#record;
1037    };
1038  }
1039
1040  constructor(key: Secret, record: ProcessRecord<T>) {
1041    if (key !== secret) throw new TypeError("Illegal constructor");
1042    this.#record = record;
1043    set(children, this, record);
1044    register(finalizer, this, record.id, this);
1045  }
1046  get pid() {
1047    return state(children, this).pid;
1048  }
1049  get stdin() {
1050    const record = state(children, this);
1051    return record.automaticInput ? null : record.writes[0]?.stream ?? null;
1052  }
1053  get stdout() {
1054    return state(children, this).reads[1]?.stream ?? null;
1055  }
1056  get stderr() {
1057    return state(children, this).reads[2]?.stream ?? null;
1058  }
1059  get stdio() {
1060    return state(children, this).stdio;
1061  }
1062  get terminal() {
1063    return state(children, this).terminal;
1064  }
1065  get status() {
1066    return state(children, this).status;
1067  }
1068  get completed() {
1069    return state(children, this).completed;
1070  }
1071  output(): Promise<ProcessOutputValue<T>> {
1072    const record = childRecord(this);
1073    return then(
1074      gather(this, record),
1075      (raw) =>
1076        checked(
1077          output(raw.status, raw.stdout, raw.stderr, record.decode),
1078          record.check,
1079          raw.cause,
1080        ),
1081    );
1082  }
1083  kill(signal: string = "SIGTERM"): boolean {
1084    const record = state(children, this);
1085    return record.disposed
1086      ? false
1087      : result(op_process_kill(args(record.id, usv(signal))));
1088  }
1089  terminate(): Promise<ProcessStatusValue> {
1090    const record = state(children, this);
1091    if (!record.disposed) {
1092      result(op_process_stop(args(record.id, "terminated")));
1093    }
1094    return record.completed;
1095  }
1096  ref(): void {
1097    const record = state(children, this);
1098    if (!record.disposed && !record.referenced) {
1099      record.referenced = true;
1100      keepAlive(record, result(op_process_ref(args(record.id, true))));
1101    }
1102  }
1103  unref(): void {
1104    const record = state(children, this);
1105    if (!record.disposed && record.referenced) {
1106      record.referenced = false;
1107      result(op_process_ref(args(record.id, false)));
1108    }
1109  }
1110  async [asyncDispose](): Promise<void> {
1111    const record = state(children, this);
1112    if (!record.disposed) {
1113      result(op_process_stop(args(record.id, "dispose")));
1114      await safePromise(record.completed);
1115      record.disposed = true;
1116      unregister(finalizer, this);
1117      op_process_release(record.id);
1118    }
1119  }
1120}
1121define(ChildProcess.prototype, SymbolToStringTag, {
1122  value: "ChildProcess",
1123  configurable: true,
1124});
1125function start<T extends ProcessBytes | string>(
1126  options: Prepared<T>,
1127): ChildProcess<T> {
1128  const info = result(op_process_spawn(options.native));
1129  const record: ProcessRecord<T> = {
1130    __proto__: null,
1131    id: info.id,
1132    pid: info.pid,
1133    reads: ObjectCreate(null),
1134    writes: ObjectCreate(null),
1135    stdio: ObjectCreate(null),
1136    terminal: null,
1137    collectLimit: options.maxCollectedBytes,
1138    decoder: options.decoder,
1139    decode: options.decode,
1140    encoding: options.decoder === null ? "utf-8" : encodingOf(options.decoder),
1141    check: options.check,
1142    automaticInput: options.input !== undefined,
1143    collected: false,
1144    disposed: false,
1145    finished: false,
1146    referenced: true,
1147    inputTask: undefined,
1148    inputFailed: false,
1149    inputError: undefined,
1150    abortReason: undefined,
1151    status: undefined!,
1152    completed: undefined!,
1153  };
1154  const child = new ChildProcess(secret, record);
1155  for (let i = 0; i < info.readable.length; i++) {
1156    const fd = info.readable[i];
1157    const source = readable(record, fd);
1158    source.owner = child;
1159    record.reads[fd] = source;
1160    if (fd >= 3) own(record.stdio, fd, source.stream);
1161  }
1162  for (let i = 0; i < info.writable.length; i++) {
1163    const fd = info.writable[i];
1164    const source = writable(record, fd);
1165    source.owner = child;
1166    record.writes[fd] = source;
1167    if (fd >= 3) own(record.stdio, fd, source.stream);
1168  }
1169  freeze(record.stdio);
1170  record.status = then(
1171    op_process_wait(args(info.id, false)),
1172    (reply) => freeze(result(reply)),
1173  );
1174  record.completed = then(
1175    op_process_wait(args(info.id, true)),
1176    (reply) => freeze(result(reply)),
1177  );
1178  markPromiseHandled(record.status);
1179  markPromiseHandled(record.completed);
1180  keepAlive(record, 0);
1181  retain(live, child);
1182  let removeAbort: (() => void) | undefined;
1183  if (options.signal !== undefined) {
1184    removeAbort = addAbort(options.signal as AbortSignal, () => {
1185      record.abortReason = state(signals, options.signal).reason;
1186      if (!record.finished && !record.disposed) {
1187        result(op_process_stop(args(record.id, "abort")));
1188      }
1189    });
1190  }
1191  const cleanup = then(record.completed, () => {
1192    record.finished = true;
1193    removeAbort?.();
1194    release(live, child);
1195  });
1196  markPromiseHandled(cleanup);
1197  if (info.terminal) {
1198    record.terminal = freeze({
1199      readable: record.reads[1]!.stream,
1200      writable: record.writes[0]!.stream,
1201      resize(columns: number, rows: number) {
1202        required(arguments.length, 2);
1203        if (record.disposed) {
1204          throw failure("BadResource", "Subprocess is disposed");
1205        }
1206        result(
1207          op_process_resize(
1208            args(
1209              record.id,
1210              integer(columns, undefined, 1, 65535)!,
1211              integer(rows, undefined, 1, 65535)!,
1212            ),
1213          ),
1214        );
1215      },
1216    });
1217  }
1218  if (options.input !== undefined) {
1219    record.inputTask = feed(record, options.input!, options.stream);
1220    markPromiseHandled(record.inputTask);
1221  }
1222  return child;
1223}
1224// Static methods retain the non-constructible function values while allowing overloads.
1225class Launch {
1226  static spawn(
1227    command: string | URL,
1228    options?: SpawnOptions & { encoding?: undefined },
1229  ): ChildProcess<ProcessBytes>;
1230  static spawn(
1231    command: string | URL,
1232    options: SpawnOptions & { encoding: string },
1233  ): ChildProcess<string>;
1234  static spawn(
1235    command: string | URL,
1236    args: Iterable<string> | undefined,
1237    options?: SpawnOptions & { args?: undefined; encoding?: undefined },
1238  ): ChildProcess<ProcessBytes>;
1239  static spawn(
1240    command: string | URL,
1241    args: Iterable<string> | undefined,
1242    options: SpawnOptions & { args?: undefined; encoding: string },
1243  ): ChildProcess<string>;
1244  static spawn(
1245    command: string | URL,
1246    second: unknown = undefined,
1247    third: unknown = undefined,
1248  ): ChildProcess {
1249    required(arguments.length, 1);
1250    const call = invocation(second, third);
1251    return start(normalize(command, call.args, call.options, false, false));
1252  }
1253  static run(
1254    command: string | URL,
1255    options?: ProcessOptions & { encoding?: undefined },
1256  ): ProcessPromise<ProcessBytes>;
1257  static run(
1258    command: string | URL,
1259    options: ProcessOptions & { encoding: string },
1260  ): ProcessPromise<string>;
1261  static run(
1262    command: string | URL,
1263    args: Iterable<string> | undefined,
1264    options?: ProcessOptions & { args?: undefined; encoding?: undefined },
1265  ): ProcessPromise<ProcessBytes>;
1266  static run(
1267    command: string | URL,
1268    args: Iterable<string> | undefined,
1269    options: ProcessOptions & { args?: undefined; encoding: string },
1270  ): ProcessPromise<string>;
1271  static run(
1272    command: string | URL,
1273    second: unknown = undefined,
1274    third: unknown = undefined,
1275  ): ProcessPromise {
1276    let child: ChildProcess | null = null, failed = false, failure: unknown;
1277    try {
1278      required(arguments.length, 1);
1279      const call = invocation(second, third);
1280      child = start(normalize(command, call.args, call.options, true, false));
1281    } catch (error) {
1282      failed = true;
1283      failure = error;
1284    }
1285    return invoke(child, failed, failure);
1286  }
1287  static runSync(
1288    command: string | URL,
1289    options?: RunSyncOptions & { encoding?: undefined },
1290  ): ProcessOutputValue<ProcessBytes>;
1291  static runSync(
1292    command: string | URL,
1293    options: RunSyncOptions & { encoding: string },
1294  ): ProcessOutputValue<string>;
1295  static runSync(
1296    command: string | URL,
1297    args: Iterable<string> | undefined,
1298    options?: RunSyncOptions & { args?: undefined; encoding?: undefined },
1299  ): ProcessOutputValue<ProcessBytes>;
1300  static runSync(
1301    command: string | URL,
1302    args: Iterable<string> | undefined,
1303    options: RunSyncOptions & { args?: undefined; encoding: string },
1304  ): ProcessOutputValue<string>;
1305  static runSync(
1306    command: string | URL,
1307    second: unknown = undefined,
1308    third: unknown = undefined,
1309  ): ProcessOutputValue {
1310    required(arguments.length, 1);
1311    const call = invocation(second, third);
1312    const prepared = normalize(command, call.args, call.options, true, true);
1313    const reply = result(
1314      op_process_run_sync(
1315        args(
1316          prepared.native,
1317          prepared.input === undefined ? null : prepared.input as ProcessBytes,
1318        ),
1319      ),
1320    );
1321    return checked(
1322      output(reply.status, reply.stdout, reply.stderr, prepared.decode),
1323      prepared.check,
1324    );
1325  }
1326}
1327const { spawn, run, runSync } = Launch;
1328export {
1329  addSignalListener,
1330  ChildProcess,
1331  OutputLimitError,
1332  ProcessAbortError,
1333  ProcessError,
1334  ProcessTimeoutError,
1335  removeSignalListener,
1336  run,
1337  runSync,
1338  spawn,
1339};