1/**
2 * Shared Node.js implementation of the child process spawner service.
3 *
4 * This module adapts `node:child_process.spawn` to the Effect
5 * `ChildProcessSpawner` service. Provide {@link layer} to run `ChildProcess`
6 * commands in Node-compatible runtimes: commands get scoped process handles
7 * with stdin sinks, stdout and stderr streams, exit-code waiting,
8 * interruption-time cleanup, process killing, and custom file-descriptor pipes.
9 *
10 * The implementation sits below the command-building API. It validates and
11 * resolves `cwd` through the Effect `FileSystem` and `Path` services,
12 * translates Node errno failures to `PlatformError`, and uses scopes to
13 * terminate referenced children when the owning effect is interrupted or
14 * finalized. Pipelines are flattened by {@link flattenCommand} and spawned one
15 * process at a time, wiring the selected source stream (`stdout`, `stderr`,
16 * `all`, or `fdN`) to the destination `stdin` or `fdN`.
17 *
18 * Scoped release and `kill` wait for the signalled process group. Without
19 * `forceKillAfter`, the wait is limited to one second and never escalates.
20 * With it, cleanup may take the configured duration plus a final one-second
21 * wait after `SIGKILL`. Zombie descendants can consume either full bound. On
22 * Windows, `taskkill` terminates the tree and only the leader's exit is awaited.
23 *
24 * POSIX cleanup targets a numeric process-group ID. If the group disappears
25 * and its ID is reused before cleanup, an unrelated group may be signalled.
26 *
27 * @since 4.0.0
28 */
29import type * as Arr from "effect/Array"
30import * as Deferred from "effect/Deferred"
31import * as Duration from "effect/Duration"
32import * as Effect from "effect/Effect"
33import * as Exit from "effect/Exit"
34import * as FileSystem from "effect/FileSystem"
35import * as Layer from "effect/Layer"
36import * as Path from "effect/Path"
37import type * as PlatformError from "effect/PlatformError"
38import * as Predicate from "effect/Predicate"
39import * as ChildProcess from "effect/process/ChildProcess"
40import type { ChildProcessHandle } from "effect/process/ChildProcessSpawner"
41import {
42 ChildProcessSpawner,
43 ExitCode,
44 make as makeSpawner,
45 makeHandle,
46 ProcessId
47} from "effect/process/ChildProcessSpawner"
48import type * as Scope from "effect/Scope"
49import * as Sink from "effect/Sink"
50import * as Stream from "effect/Stream"
51import * as NodeChildProcess from "node:child_process"
52import { PassThrough } from "node:stream"
53import { buildSpawnOptions } from "./internal/nodeChildProcessSpawner.ts"
54import { handleErrnoException } from "./internal/utils.ts"
55import * as NodeSink from "./NodeSink.ts"
56import * as NodeStream from "./NodeStream.ts"
57
58const toError = (error: unknown): Error =>
59 error instanceof globalThis.Error
60 ? error
61 : new globalThis.Error(String(error))
62
63const toPlatformError = (
64 method: string,
65 error: NodeJS.ErrnoException,
66 command: ChildProcess.Command
67): PlatformError.PlatformError => {
68 const { commands } = flattenCommand(command)
69 const commandStr = commands.reduce((acc, curr) => {
70 const cmd = `${curr.command} ${curr.args.join(" ")}`
71 return acc.length === 0 ? cmd : `${acc} | ${cmd}`
72 }, "")
73 return handleErrnoException("ChildProcess", method)(error, [commandStr])
74}
75
76type ExitCodeWithSignal = readonly [code: number | null, signal: NodeJS.Signals | null]
77type ExitSignal = Deferred.Deferred<ExitCodeWithSignal>
78
79const processGroupGraceMillis = 1_000
80const processGroupPollIntervalMillis = 10
81
82/** The leader has not exited, or a member of its process group still exists. */
83const isProcessAlive = (childProcess: NodeChildProcess.ChildProcess, exitSignal: ExitSignal): boolean => {
84 if (!Deferred.isDoneUnsafe(exitSignal)) {
85 return true
86 }
87 if (globalThis.process.platform === "win32") {
88 return false
89 }
90 try {
91 globalThis.process.kill(-childProcess.pid!, 0)
92 return true
93 } catch {
94 return false
95 }
96}
97
98const taskkill = (
99 childProcess: NodeChildProcess.ChildProcess,
100 onExit: (error: NodeChildProcess.ExecException | null) => void = () => {}
101) =>
102 NodeChildProcess.execFile(
103 "taskkill",
104 ["/pid", String(childProcess.pid!), "/T", "/F"],
105 { windowsHide: true },
106 onExit
107 )
108
109const make = Effect.gen(function*() {
110 const fs = yield* FileSystem.FileSystem
111 const path = yield* Path.Path
112
113 const resolveWorkingDirectory = Effect.fnUntraced(
114 function*(options: ChildProcess.CommandOptions) {
115 if (Predicate.isUndefined(options.cwd)) return undefined
116 // Validate that the specified directory is accessible
117 yield* fs.access(options.cwd)
118 return path.resolve(options.cwd)
119 }
120 )
121
122 const resolveEnvironment = (options: ChildProcess.CommandOptions) => {
123 return options.extendEnv
124 ? { ...globalThis.process.env, ...options.env }
125 : options.env
126 }
127
128 const inputToStdioOption = (input: ChildProcess.CommandInput | undefined): NodeChildProcess.IOType | undefined =>
129 Stream.isStream(input) ? "pipe" : input
130
131 const outputToStdioOption = (input: ChildProcess.CommandOutput | undefined): NodeChildProcess.IOType | undefined =>
132 Sink.isSink(input) ? "pipe" : input
133
134 const resolveStdinOption = (options: ChildProcess.CommandOptions): ChildProcess.StdinConfig => {
135 const defaultConfig: ChildProcess.StdinConfig = { stream: "pipe", encoding: "utf-8", endOnDone: true }
136 if (Predicate.isUndefined(options.stdin)) {
137 return defaultConfig
138 }
139 if (typeof options.stdin === "string") {
140 return { ...defaultConfig, stream: options.stdin }
141 }
142 if (Stream.isStream(options.stdin)) {
143 return { ...defaultConfig, stream: options.stdin }
144 }
145 return {
146 stream: options.stdin.stream,
147 encoding: options.stdin.encoding ?? defaultConfig.encoding,
148 endOnDone: options.stdin.endOnDone ?? defaultConfig.endOnDone
149 }
150 }
151
152 const resolveOutputOption = (
153 options: ChildProcess.CommandOptions,
154 streamName: "stdout" | "stderr"
155 ): ChildProcess.StdoutConfig => {
156 const option = options[streamName]
157 if (Predicate.isUndefined(option)) {
158 return { stream: "pipe" }
159 }
160 if (typeof option === "string") {
161 return { stream: option }
162 }
163 if (Sink.isSink(option)) {
164 return { stream: option }
165 }
166 return { stream: option.stream }
167 }
168
169 interface ResolvedAdditionalFd {
170 readonly fd: number
171 readonly config: ChildProcess.AdditionalFdConfig
172 }
173
174 const resolveAdditionalFds = (
175 options: ChildProcess.CommandOptions
176 ): ReadonlyArray<ResolvedAdditionalFd> => {
177 if (Predicate.isUndefined(options.additionalFds)) {
178 return []
179 }
180 const result: Array<ResolvedAdditionalFd> = []
181 for (const [name, config] of Object.entries(options.additionalFds)) {
182 const fd = ChildProcess.parseFdName(name)
183 if (Predicate.isNotUndefined(fd)) {
184 result.push({ fd, config })
185 }
186 }
187 // Sort by fd number to ensure correct ordering
188 return result.sort((a, b) => a.fd - b.fd)
189 }
190
191 const buildStdioArray = (
192 stdinConfig: ChildProcess.StdinConfig,
193 stdoutConfig: ChildProcess.StdoutConfig,
194 stderrConfig: ChildProcess.StderrConfig,
195 additionalFds: ReadonlyArray<ResolvedAdditionalFd>
196 ): NodeChildProcess.StdioOptions => {
197 const stdio: Array<NodeChildProcess.IOType | undefined> = [
198 inputToStdioOption(stdinConfig.stream),
199 outputToStdioOption(stdoutConfig.stream),
200 outputToStdioOption(stderrConfig.stream)
201 ]
202
203 if (additionalFds.length === 0) {
204 return stdio as NodeChildProcess.StdioOptions
205 }
206
207 // Find the maximum fd number to size the array correctly
208 const maxFd = additionalFds.reduce((max, { fd }) => Math.max(max, fd), 2)
209
210 // Fill gaps with "ignore"
211 for (let i = 3; i <= maxFd; i++) {
212 stdio[i] = "ignore"
213 }
214
215 // Set up additional fds as "pipe"
216 for (const { fd } of additionalFds) {
217 stdio[fd] = "pipe"
218 }
219
220 return stdio as NodeChildProcess.StdioOptions
221 }
222
223 const setupAdditionalFds = Effect.fnUntraced(function*(
224 command: ChildProcess.StandardCommand,
225 childProcess: NodeChildProcess.ChildProcess,
226 additionalFds: ReadonlyArray<ResolvedAdditionalFd>
227 ) {
228 if (additionalFds.length === 0) {
229 return {
230 getInputFd: () => Sink.drain,
231 getOutputFd: () => Stream.empty
232 }
233 }
234
235 const inputSinks = new Map<number, Sink.Sink<void, Uint8Array, never, PlatformError.PlatformError>>()
236 const outputStreams = new Map<number, Stream.Stream<Uint8Array, PlatformError.PlatformError>>()
237
238 for (const { config, fd } of additionalFds) {
239 const nodeStream = childProcess.stdio[fd]
240
241 switch (config.type) {
242 case "input": {
243 // Create a sink to write to for input file descriptors
244 let sink: Sink.Sink<void, Uint8Array, never, PlatformError.PlatformError> = Sink.drain
245 if (nodeStream && "write" in nodeStream) {
246 sink = NodeSink.fromWritable({
247 evaluate: () => nodeStream,
248 onError: (error) => toPlatformError(`fromWritable(fd${fd})`, toError(error), command)
249 })
250 }
251
252 // If user provided a stream, pipe it into the sink
253 if (config.stream) {
254 yield* Effect.forkScoped(Stream.run(config.stream, sink))
255 }
256
257 inputSinks.set(fd, sink)
258
259 break
260 }
261 case "output": {
262 // Create a stream to read from for output file descriptors
263 let stream: Stream.Stream<Uint8Array, PlatformError.PlatformError> = Stream.empty
264 if (nodeStream && "read" in nodeStream) {
265 const passThrough = new PassThrough()
266 nodeStream.on("error", (error) => passThrough.destroy(error))
267 nodeStream.pipe(passThrough)
268 stream = NodeStream.fromReadable({
269 evaluate: () => passThrough,
270 onError: (error) => toPlatformError(`fromReadable(fd${fd})`, toError(error), command)
271 })
272 }
273
274 // If user provided a sink, transduce the stream through it
275 if (config.sink) {
276 stream = Stream.transduce(stream, config.sink)
277 }
278
279 outputStreams.set(fd, stream)
280
281 break
282 }
283 }
284 }
285
286 return {
287 getInputFd: (fd: number) => inputSinks.get(fd) ?? Sink.drain,
288 getOutputFd: (fd: number) => outputStreams.get(fd) ?? Stream.empty
289 }
290 })
291
292 const setupChildStdin = (
293 command: ChildProcess.StandardCommand,
294 childProcess: NodeChildProcess.ChildProcess,
295 config: ChildProcess.StdinConfig
296 ) =>
297 Effect.suspend(() => {
298 // If the child process has a standard input stream, connect it to the
299 // sink that will attached to the process handle
300 let sink: Sink.Sink<void, unknown, never, PlatformError.PlatformError> = Sink.drain
301 if (Predicate.isNotNull(childProcess.stdin)) {
302 sink = NodeSink.fromWritable({
303 evaluate: () => childProcess.stdin!,
304 onError: (error) => toPlatformError("fromWritable(stdin)", toError(error), command),
305 endOnDone: config.endOnDone,
306 encoding: config.encoding
307 })
308 }
309
310 // If the user provided a `Stream`, run it into the stdin sink
311 if (Stream.isStream(config.stream)) {
312 return Effect.as(Effect.forkScoped(Stream.run(config.stream, sink)), sink)
313 }
314
315 return Effect.succeed(sink)
316 })
317
318 const setupChildOutputStreams = (
319 command: ChildProcess.StandardCommand,
320 childProcess: NodeChildProcess.ChildProcess,
321 stdoutConfig: ChildProcess.StdoutConfig,
322 stderrConfig: ChildProcess.StderrConfig
323 ): {
324 stdout: Stream.Stream<Uint8Array, PlatformError.PlatformError>
325 stderr: Stream.Stream<Uint8Array, PlatformError.PlatformError>
326 all: Stream.Stream<Uint8Array, PlatformError.PlatformError>
327 } => {
328 let stdout = childProcess.stdout ?
329 (() => {
330 const passThrough = new PassThrough()
331 childProcess.stdout!.on("error", (error) => passThrough.destroy(error))
332 childProcess.stdout!.pipe(passThrough)
333 return NodeStream.fromReadable({
334 evaluate: () => passThrough,
335 onError: (error) => toPlatformError("fromReadable(stdout)", toError(error), command)
336 })
337 })() :
338 Stream.empty
339 let stderr = childProcess.stderr ?
340 (() => {
341 const passThrough = new PassThrough()
342 childProcess.stderr!.on("error", (error) => passThrough.destroy(error))
343 childProcess.stderr!.pipe(passThrough)
344 return NodeStream.fromReadable({
345 evaluate: () => passThrough,
346 onError: (error) => toPlatformError("fromReadable(stderr)", toError(error), command)
347 })
348 })() :
349 Stream.empty
350
351 if (Sink.isSink(stdoutConfig.stream)) {
352 stdout = Stream.transduce(stdout, stdoutConfig.stream)
353 }
354 if (Sink.isSink(stderrConfig.stream)) {
355 stderr = Stream.transduce(stderr, stderrConfig.stream)
356 }
357
358 const all = Stream.merge(stdout, stderr)
359
360 return { stdout, stderr, all }
361 }
362
363 const spawn = (
364 command: ChildProcess.StandardCommand,
365 spawnOptions: NodeChildProcess.SpawnOptions
366 ) =>
367 Effect.callback<
368 readonly [NodeChildProcess.ChildProcess, ExitSignal],
369 PlatformError.PlatformError
370 >((resume) => {
371 const deferred = Deferred.makeUnsafe<ExitCodeWithSignal>()
372 const handle = NodeChildProcess.spawn(
373 command.command,
374 command.args,
375 spawnOptions
376 )
377 handle.on("error", (error) => {
378 resume(Effect.fail(toPlatformError("spawn", error, command)))
379 })
380 handle.on("exit", (...args) => {
381 Deferred.doneUnsafe(deferred, Exit.succeed(args))
382 })
383 handle.on("spawn", () => {
384 resume(Effect.succeed([handle, deferred]))
385 })
386 return Effect.sync(() => {
387 handle.kill("SIGTERM")
388 })
389 })
390
391 const killProcessGroup = (
392 command: ChildProcess.StandardCommand,
393 childProcess: NodeChildProcess.ChildProcess,
394 signal: NodeJS.Signals
395 ) => {
396 if (globalThis.process.platform === "win32") {
397 return Effect.callback<void, PlatformError.PlatformError>((resume) => {
398 taskkill(childProcess, (error) => {
399 if (error) {
400 resume(Effect.fail(toPlatformError("kill", toError(error), command)))
401 } else {
402 resume(Effect.void)
403 }
404 })
405 })
406 }
407 return Effect.try({
408 try: () => {
409 globalThis.process.kill(-childProcess.pid!, signal)
410 },
411 catch: (error) => toPlatformError("kill", toError(error), command)
412 })
413 }
414
415 const killProcessGroupOnExit = (
416 childProcess: NodeChildProcess.ChildProcess,
417 signal: NodeJS.Signals
418 ): void => {
419 if (globalThis.process.platform === "win32") {
420 // ignore errors during best-effort cleanup
421 taskkill(childProcess)
422 return
423 }
424 try {
425 globalThis.process.kill(-childProcess.pid!, signal)
426 } catch {
427 // ignore errors during best-effort cleanup
428 }
429 }
430
431 const killProcess = (
432 command: ChildProcess.StandardCommand,
433 childProcess: NodeChildProcess.ChildProcess,
434 signal: NodeJS.Signals
435 ) =>
436 Effect.suspend(() => {
437 const killed = childProcess.kill(signal)
438 if (!killed) {
439 const error = new globalThis.Error("Failed to kill child process")
440 return Effect.fail(toPlatformError("kill", error, command))
441 }
442 return Effect.void
443 })
444
445 /** Waits for the leader and its process group, bounded by native time. */
446 const awaitProcessExit = (
447 childProcess: NodeChildProcess.ChildProcess,
448 exitSignal: ExitSignal,
449 timeoutMillis: number
450 ): Effect.Effect<void> =>
451 Effect.callback<void>((resume) => {
452 const deadline = Date.now() + timeoutMillis
453 let timer: NodeJS.Timeout | undefined
454 const stop = () => {
455 clearTimeout(timer)
456 childProcess.removeListener("exit", poll)
457 }
458 const poll = () => {
459 clearTimeout(timer)
460 if (Date.now() >= deadline || !isProcessAlive(childProcess, exitSignal)) {
461 stop()
462 resume(Effect.void)
463 return
464 }
465 timer = setTimeout(poll, processGroupPollIntervalMillis)
466 }
467 // spawn's exit listener completes exitSignal before this listener runs
468 childProcess.on("exit", poll)
469 poll()
470 return Effect.sync(stop)
471 })
472
473 const terminateProcessGroup = Effect.fnUntraced(function*(
474 command: ChildProcess.StandardCommand,
475 childProcess: NodeChildProcess.ChildProcess,
476 exitSignal: ExitSignal,
477 options: ChildProcess.KillOptions | undefined
478 ) {
479 const signalGroup = (signal: NodeJS.Signals) =>
480 killProcessGroup(command, childProcess, signal).pipe(
481 Effect.catch(() => killProcess(command, childProcess, signal))
482 )
483 yield* signalGroup(options?.killSignal ?? "SIGTERM")
484 if (Predicate.isUndefined(options?.forceKillAfter)) {
485 yield* awaitProcessExit(childProcess, exitSignal, processGroupGraceMillis)
486 } else {
487 yield* awaitProcessExit(childProcess, exitSignal, Duration.toMillis(options.forceKillAfter))
488 if (isProcessAlive(childProcess, exitSignal)) {
489 yield* signalGroup("SIGKILL")
490 yield* awaitProcessExit(childProcess, exitSignal, processGroupGraceMillis)
491 }
492 }
493 yield* Deferred.await(exitSignal)
494 })
495
496 /**
497 * Get the appropriate source stream from a process handle based on the
498 * `from` pipe option.
499 */
500 const getSourceStream = (
501 handle: ChildProcessHandle,
502 from: ChildProcess.PipeFromOption | undefined
503 ): Stream.Stream<Uint8Array, PlatformError.PlatformError> => {
504 const fromOption = from ?? "stdout"
505 switch (fromOption) {
506 case "stdout":
507 return handle.stdout
508 case "stderr":
509 return handle.stderr
510 case "all":
511 return handle.all
512 default: {
513 // Handle fd3, fd4, etc.
514 const fd = ChildProcess.parseFdName(fromOption)
515 if (Predicate.isNotUndefined(fd)) {
516 return handle.getOutputFd(fd)
517 }
518 // Fallback to stdout for invalid fd names
519 return handle.stdout
520 }
521 }
522 }
523
524 const spawnCommand: (
525 command: ChildProcess.Command
526 ) => Effect.Effect<
527 ChildProcessHandle,
528 PlatformError.PlatformError,
529 Scope.Scope
530 > = Effect.fnUntraced(function*(cmd) {
531 switch (cmd._tag) {
532 case "StandardCommand": {
533 const stdinConfig = resolveStdinOption(cmd.options)
534 const stdoutConfig = resolveOutputOption(cmd.options, "stdout")
535 const stderrConfig = resolveOutputOption(cmd.options, "stderr")
536 const resolvedAdditionalFds = resolveAdditionalFds(cmd.options)
537 let isReferenced = true
538
539 const cwd = yield* resolveWorkingDirectory(cmd.options)
540 const env = resolveEnvironment(cmd.options)
541 const stdio = buildStdioArray(stdinConfig, stdoutConfig, stderrConfig, resolvedAdditionalFds)
542
543 const [childProcess, exitSignal] = yield* Effect.acquireRelease(
544 spawn(cmd, buildSpawnOptions(cmd.options, { cwd, env, stdio }, process.platform)),
545 Effect.fnUntraced(function*([childProcess, exitSignal]) {
546 const exited = yield* Deferred.isDone(exitSignal)
547 if (exited) {
548 const [code] = yield* Deferred.await(exitSignal)
549 if (code !== 0 && Predicate.isNotNull(code)) {
550 yield* Effect.ignore(terminateProcessGroup(cmd, childProcess, exitSignal, cmd.options))
551 } else if (isReferenced && process.platform !== "win32" && cmd.options.detached !== false) {
552 yield* Effect.ignore(terminateProcessGroup(cmd, childProcess, exitSignal, cmd.options))
553 }
554 return
555 }
556 if (!isReferenced) {
557 return
558 }
559 yield* Effect.ignore(terminateProcessGroup(cmd, childProcess, exitSignal, cmd.options))
560 })
561 )
562
563 const pid = ProcessId(childProcess.pid!)
564 childProcess.on("exit", (code) => {
565 if (code !== 0 && Predicate.isNotNull(code)) {
566 killProcessGroupOnExit(childProcess, cmd.options.killSignal ?? "SIGTERM")
567 }
568 })
569 const reref = Effect.sync(() => {
570 if (!isReferenced) {
571 childProcess.ref()
572 isReferenced = true
573 }
574 })
575 const unref = Effect.sync(() => {
576 if (isReferenced) {
577 childProcess.unref()
578 isReferenced = false
579 }
580 return reref
581 })
582 const stdin = yield* setupChildStdin(cmd, childProcess, stdinConfig)
583 const { all, stderr, stdout } = setupChildOutputStreams(cmd, childProcess, stdoutConfig, stderrConfig)
584 const { getInputFd, getOutputFd } = yield* setupAdditionalFds(cmd, childProcess, resolvedAdditionalFds)
585 const isRunning = Effect.map(Deferred.isDone(exitSignal), (done) => !done)
586 const exitCode = Effect.flatMap(Deferred.await(exitSignal), ([code, signal]) => {
587 if (Predicate.isNotNull(code)) {
588 return Effect.succeed(ExitCode(code))
589 }
590 // If code is `null`, then `signal` must be defined. See the NodeJS
591 // documentation for the `"exit"` event on a `child_process`.
592 // https://nodejs.org/api/child_process.html#child_process_event_exit
593 const error = new globalThis.Error(`Process interrupted due to receipt of signal: '${signal}'`)
594 return Effect.fail(toPlatformError("exitCode", error, cmd))
595 })
596 const kill = (options?: ChildProcess.KillOptions | undefined) =>
597 terminateProcessGroup(cmd, childProcess, exitSignal, options)
598
599 return makeHandle({
600 pid,
601 exitCode,
602 isRunning,
603 kill,
604 stdin,
605 stdout,
606 stderr,
607 all,
608 getInputFd,
609 getOutputFd,
610 unref
611 })
612 }
613 case "PipedCommand": {
614 const { commands, pipeOptions } = flattenCommand(cmd)
615 const [root, ...pipeline] = commands
616
617 const handles = [yield* spawnCommand(root)]
618
619 for (let i = 0; i < pipeline.length; i++) {
620 const command = pipeline[i]
621 const options = pipeOptions[i] ?? {}
622 const stdinConfig = resolveStdinOption(command.options)
623
624 // Get the appropriate stream from the source based on `from` option
625 const sourceStream = Stream.unwrap(
626 Effect.succeed(getSourceStream(handles[handles.length - 1], options.from))
627 )
628
629 // Determine where to pipe: stdin or custom fd
630 const toOption = options.to ?? "stdin"
631
632 if (toOption === "stdin") {
633 // Pipe to stdin (default behavior)
634 handles.push(
635 yield* spawnCommand(ChildProcess.make(command.command, command.args, {
636 ...command.options,
637 stdin: { ...stdinConfig, stream: sourceStream }
638 }))
639 )
640 } else {
641 // Pipe to custom fd (fd3, fd4, etc.)
642 const fd = ChildProcess.parseFdName(toOption)
643 if (Predicate.isNotUndefined(fd)) {
644 const fdName = ChildProcess.fdName(fd) as `fd${number}`
645 const existingFds = command.options.additionalFds ?? {}
646 handles.push(
647 yield* spawnCommand(ChildProcess.make(command.command, command.args, {
648 ...command.options,
649 additionalFds: {
650 ...existingFds,
651 [fdName]: { type: "input" as const, stream: sourceStream }
652 }
653 }))
654 )
655 } else {
656 // Invalid fd name, fall back to stdin
657 handles.push(
658 yield* spawnCommand(ChildProcess.make(command.command, command.args, {
659 ...command.options,
660 stdin: { ...stdinConfig, stream: sourceStream }
661 }))
662 )
663 }
664 }
665 }
666
667 const handle = handles[handles.length - 1]
668 const kill = (options?: ChildProcess.KillOptions | undefined) =>
669 Effect.forEach([...handles].reverse(), (handle) => Effect.ignore(handle.kill(options)), { discard: true })
670 const unref = Effect.gen(function*() {
671 const rerefs: Array<Effect.Effect<void, PlatformError.PlatformError>> = []
672 for (const handle of handles) {
673 rerefs.push(yield* handle.unref)
674 }
675 return Effect.forEach([...rerefs].reverse(), (reref) => reref, { discard: true })
676 })
677
678 return makeHandle({
679 pid: handle.pid,
680 exitCode: handle.exitCode,
681 isRunning: handle.isRunning,
682 kill,
683 stdin: handles[0].stdin,
684 stdout: handle.stdout,
685 stderr: handle.stderr,
686 all: handle.all,
687 getInputFd: handle.getInputFd,
688 getOutputFd: handle.getOutputFd,
689 unref
690 })
691 }
692 }
693 })
694
695 return makeSpawner(spawnCommand)
696})
697
698/**
699 * Layer that provides the `NodeChildProcessSpawner` implementation.
700 *
701 * @category layers
702 * @since 4.0.0
703 */
704export const layer: Layer.Layer<
705 ChildProcessSpawner,
706 never,
707 FileSystem.FileSystem | Path.Path
708> = Layer.effect(ChildProcessSpawner, make)
709
710// =============================================================================
711// Internal Helpers
712// =============================================================================
713
714/**
715 * Result of flattening a pipeline of commands.
716 *
717 * @category models
718 * @since 4.0.0
719 */
720export interface FlattenedPipeline {
721 readonly commands: Arr.NonEmptyReadonlyArray<ChildProcess.StandardCommand>
722 readonly pipeOptions: ReadonlyArray<ChildProcess.PipeOptions>
723}
724
725/**
726 * Flattens a `Command` into an array of `StandardCommand`s along with pipe
727 * options for each connection.
728 *
729 * @category transforming
730 * @since 4.0.0
731 */
732export const flattenCommand = (
733 command: ChildProcess.Command
734): FlattenedPipeline => {
735 const commands: Array<ChildProcess.StandardCommand> = []
736 const pipeOptions: Array<ChildProcess.PipeOptions> = []
737
738 const flatten = (cmd: ChildProcess.Command): void => {
739 switch (cmd._tag) {
740 case "StandardCommand": {
741 commands.push(cmd)
742 break
743 }
744 case "PipedCommand": {
745 // Recursively flatten left side first
746 flatten(cmd.left)
747 // Store the pipe options for this connection
748 pipeOptions.push(cmd.options)
749 // Then flatten right side
750 flatten(cmd.right)
751 break
752 }
753 }
754 }
755
756 flatten(command)
757
758 // The commands array is guaranteed to be non-empty since we always have at
759 // least one command in the input. We validate this at runtime and return a
760 // properly typed tuple.
761 if (commands.length === 0) {
762 // This should never happen given a valid Command input
763 throw new Error("flattenCommand produced empty commands array")
764 }
765
766 const [first, ...rest] = commands
767 const nonEmptyCommands: Arr.NonEmptyReadonlyArray<ChildProcess.StandardCommand> = [first, ...rest]
768
769 return { commands: nonEmptyCommands, pipeOptions }
770}