1import * as Context from "effect/Context";
2import * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Option from "effect/Option";
6import * as PlatformError from "effect/PlatformError";
7import * as Schema from "effect/Schema";
8import * as Scope from "effect/Scope";
9import * as Stream from "effect/Stream";
10import * as ChildProcess from "effect/process/ChildProcess";
11import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
12import { resolveSpawnCommand } from "@t3tools/shared/shell";
13import {
14 collectUint8StreamText,
15 decodeUtf8,
16 type CollectedUint8StreamText,
17} from "@t3tools/provider-core/server/collectStreamText";
18
19export interface ProcessRunInput {
20 readonly command: string;
21 readonly args: ReadonlyArray<string>;
22 readonly cwd?: string | undefined;
23 readonly spawnCwd?: string | undefined;
24 readonly timeout?: Duration.Input | undefined;
25 readonly env?: NodeJS.ProcessEnv | undefined;
26 readonly stdin?: string | undefined;
27 /** Receives every stdout chunk, including bytes beyond the buffered output limit. */
28 readonly onStdoutChunk?: ((chunk: Uint8Array) => void) | undefined;
29 readonly maxOutputBytes?: number | undefined;
30 readonly outputMode?: "error" | "truncate" | undefined;
31 readonly truncatedMarker?: string | undefined;
32 /**
33 * On timeout, return a synthetic timedOut result.
34 * Partial stdout/stderr are not preserved.
35 */
36 readonly timeoutBehavior?: "error" | "timedOutResult" | undefined;
37}
38
39export interface ProcessRunOutput {
40 readonly stdout: string;
41 readonly stderr: string;
42 readonly code: ChildProcessSpawner.ExitCode | null;
43 readonly timedOut: boolean;
44 readonly stdoutTruncated: boolean;
45 readonly stderrTruncated: boolean;
46 readonly stdoutInvalidUtf8: boolean;
47 readonly stderrInvalidUtf8: boolean;
48}
49
50const ProcessInvocationFields = {
51 command: Schema.String,
52 argumentCount: Schema.Number,
53 cwd: Schema.optional(Schema.String),
54 spawnCwd: Schema.optional(Schema.String),
55};
56
57const formatProcessInvocation = (input: {
58 readonly command: string;
59 readonly cwd?: string | undefined;
60 readonly spawnCwd?: string | undefined;
61}): string => {
62 const executionCwd = input.spawnCwd ?? input.cwd;
63 return executionCwd === undefined
64 ? `'${input.command}'`
65 : `'${input.command}' in '${executionCwd}'`;
66};
67
68export class ProcessSpawnError extends Schema.TaggedError<ProcessSpawnError>()(
69 "ProcessSpawnError",
70 {
71 ...ProcessInvocationFields,
72 resolvedCommand: Schema.optional(Schema.String),
73 resolvedArgumentCount: Schema.optional(Schema.Number),
74 shell: Schema.optional(Schema.Boolean),
75 cause: Schema.Defect(),
76 },
77) {
78 override get message(): string {
79 return `Failed to spawn process ${formatProcessInvocation(this)}`;
80 }
81}
82
83export class ProcessStdinError extends Schema.TaggedError<ProcessStdinError>()(
84 "ProcessStdinError",
85 {
86 ...ProcessInvocationFields,
87 stdinBytes: Schema.Number,
88 cause: Schema.Defect(),
89 },
90) {
91 override get message(): string {
92 return `Failed to write stdin for process ${formatProcessInvocation(this)}`;
93 }
94}
95
96export class ProcessOutputLimitError extends Schema.TaggedError<ProcessOutputLimitError>()(
97 "ProcessOutputLimitError",
98 {
99 ...ProcessInvocationFields,
100 stream: Schema.Literals(["stdout", "stderr"]),
101 maxBytes: Schema.Number,
102 observedBytes: Schema.Number,
103 },
104) {
105 override get message(): string {
106 return `Process ${formatProcessInvocation(this)} ${this.stream} produced ${this.observedBytes} bytes, exceeding the ${this.maxBytes} byte limit`;
107 }
108}
109
110export class ProcessReadError extends Schema.TaggedError<ProcessReadError>()("ProcessReadError", {
111 ...ProcessInvocationFields,
112 stream: Schema.Literals(["stdout", "stderr", "exitCode"]),
113 cause: Schema.Defect(),
114}) {
115 override get message(): string {
116 return `Failed to read ${this.stream} for process ${formatProcessInvocation(this)}`;
117 }
118}
119
120export class ProcessTimeoutError extends Schema.TaggedError<ProcessTimeoutError>()(
121 "ProcessTimeoutError",
122 {
123 ...ProcessInvocationFields,
124 timeoutMs: Schema.Number,
125 },
126) {
127 override get message(): string {
128 return `Process ${formatProcessInvocation(this)} timed out after ${this.timeoutMs}ms`;
129 }
130}
131
132export const ProcessRunError = Schema.Union([
133 ProcessSpawnError,
134 ProcessStdinError,
135 ProcessOutputLimitError,
136 ProcessReadError,
137 ProcessTimeoutError,
138]);
139export type ProcessRunError = typeof ProcessRunError.Type;
140
141export class ProcessRunner extends Context.Service<
142 ProcessRunner,
143 {
144 readonly run: (input: ProcessRunInput) => Effect.Effect<ProcessRunOutput, ProcessRunError>;
145 }
146>()("t3/processRunner") {}
147
148const DEFAULT_TIMEOUT = "60 seconds";
149const DEFAULT_MAX_OUTPUT_BYTES = 8 * 1024 * 1024;
150
151export { isWindowsCommandNotFound } from "@t3tools/provider-core/server/snapshotProbe";
152
153// Untraced: no attributes, and its time is the runProcessCore span. Errors fail that span.
154const collectText = Effect.fnUntraced(function* (input: {
155 readonly command: string;
156 readonly args: ReadonlyArray<string>;
157 readonly cwd?: string | undefined;
158 readonly spawnCwd?: string | undefined;
159 readonly streamName: "stdout" | "stderr";
160 readonly stream: Stream.Stream<Uint8Array, PlatformError.PlatformError>;
161 readonly maxOutputBytes: number;
162 readonly outputMode: "error" | "truncate";
163 readonly truncatedMarker: string;
164}) {
165 const stream = input.stream.pipe(
166 Stream.mapError(
167 (cause) =>
168 new ProcessReadError({
169 command: input.command,
170 argumentCount: input.args.length,
171 cwd: input.cwd,
172 spawnCwd: input.spawnCwd,
173 stream: input.streamName,
174 cause,
175 }),
176 ),
177 );
178
179 if (input.outputMode === "truncate") {
180 return yield* collectUint8StreamText({
181 stream,
182 maxBytes: input.maxOutputBytes,
183 truncatedMarker: input.truncatedMarker,
184 });
185 }
186
187 return yield* stream.pipe(
188 Stream.runFoldEffect<
189 {
190 readonly chunks: Uint8Array<ArrayBufferLike>[];
191 readonly bytes: number;
192 },
193 Uint8Array<ArrayBufferLike>,
194 ProcessOutputLimitError | ProcessReadError,
195 never
196 >(
197 () => ({ chunks: [], bytes: 0 }),
198 (state, chunk) => {
199 const remainingBytes = input.maxOutputBytes - state.bytes;
200 if (chunk.byteLength > remainingBytes) {
201 return Effect.fail(
202 new ProcessOutputLimitError({
203 command: input.command,
204 argumentCount: input.args.length,
205 cwd: input.cwd,
206 spawnCwd: input.spawnCwd,
207 stream: input.streamName,
208 maxBytes: input.maxOutputBytes,
209 observedBytes: state.bytes + chunk.byteLength,
210 }),
211 );
212 }
213
214 state.chunks.push(chunk);
215 return Effect.succeed({
216 chunks: state.chunks,
217 bytes: state.bytes + chunk.byteLength,
218 });
219 },
220 ),
221 Effect.map((state): CollectedUint8StreamText => ({
222 ...decodeUtf8(Buffer.concat(state.chunks, state.bytes)),
223 bytes: state.bytes,
224 truncated: false,
225 })),
226 );
227});
228
229function finalizeRunProcess<R>(
230 effect: Effect.Effect<ProcessRunOutput, ProcessRunError, R | Scope.Scope>,
231 input: ProcessRunInput,
232): Effect.Effect<ProcessRunOutput, ProcessRunError, Exclude<R, Scope.Scope>> {
233 const timeout = Duration.fromInputUnsafe(input.timeout ?? DEFAULT_TIMEOUT);
234 const timeoutBehavior = input.timeoutBehavior ?? "error";
235
236 return effect.pipe(
237 Effect.scoped,
238 Effect.timeoutOption(timeout),
239 Effect.flatMap((result) => {
240 if (Option.isSome(result)) {
241 return Effect.succeed(result.value);
242 }
243 if (timeoutBehavior === "timedOutResult") {
244 return Effect.succeed({
245 stdout: "",
246 stderr: "",
247 code: null,
248 timedOut: true,
249 stdoutTruncated: false,
250 stderrTruncated: false,
251 stdoutInvalidUtf8: false,
252 stderrInvalidUtf8: false,
253 } satisfies ProcessRunOutput);
254 }
255 return Effect.fail(
256 new ProcessTimeoutError({
257 command: input.command,
258 argumentCount: input.args.length,
259 cwd: input.cwd,
260 spawnCwd: input.spawnCwd,
261 timeoutMs: Duration.toMillis(timeout),
262 }),
263 );
264 }),
265 );
266}
267
268/** The executable name without its directory, recorded as `process.command` on process spans. */
269export const commandName = (command: string) => command.replace(/^.*[\\/]/, "");
270
271const runProcessCore = Effect.fn("processRunner.runProcessCore")(function* (
272 spawner: ChildProcessSpawner.ChildProcessSpawner["Service"],
273 input: ProcessRunInput,
274): Effect.fn.Return<ProcessRunOutput, ProcessRunError, Scope.Scope> {
275 yield* Effect.annotateCurrentSpan("process.command", commandName(input.command));
276 const maxOutputBytes = input.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES;
277 const outputMode = input.outputMode ?? "error";
278 const truncatedMarker = input.truncatedMarker ?? "";
279 const extendEnv = input.env !== undefined;
280 const spawnCommand = yield* resolveSpawnCommand(
281 input.command,
282 input.args,
283 input.env === undefined ? {} : { env: input.env, extendEnv },
284 );
285
286 const child = yield* spawner
287 .spawn(
288 ChildProcess.make(spawnCommand.command, spawnCommand.args, {
289 ...((input.spawnCwd ?? input.cwd) ? { cwd: input.spawnCwd ?? input.cwd } : {}),
290 ...(input.env !== undefined
291 ? {
292 env: input.env,
293 extendEnv,
294 }
295 : {}),
296 shell: spawnCommand.shell,
297 }),
298 )
299 .pipe(
300 Effect.mapError(
301 (cause) =>
302 new ProcessSpawnError({
303 command: input.command,
304 argumentCount: input.args.length,
305 cwd: input.cwd,
306 spawnCwd: input.spawnCwd,
307 resolvedCommand: spawnCommand.command,
308 resolvedArgumentCount: spawnCommand.args.length,
309 shell: spawnCommand.shell,
310 cause,
311 }),
312 ),
313 );
314
315 const stdin = input.stdin;
316 const onStdoutChunk = input.onStdoutChunk;
317 const writeStdin =
318 stdin === undefined
319 ? Effect.void
320 : Stream.run(Stream.encodeText(Stream.make(stdin)), child.stdin).pipe(
321 Effect.mapError(
322 (cause) =>
323 new ProcessStdinError({
324 command: input.command,
325 argumentCount: input.args.length,
326 cwd: input.cwd,
327 spawnCwd: input.spawnCwd,
328 stdinBytes: Buffer.byteLength(stdin),
329 cause,
330 }),
331 ),
332 );
333
334 const [stdout, stderr] = yield* Effect.all(
335 [
336 collectText({
337 command: input.command,
338 args: input.args,
339 cwd: input.cwd,
340 spawnCwd: input.spawnCwd,
341 streamName: "stdout",
342 stream: onStdoutChunk
343 ? child.stdout.pipe(Stream.tap((chunk) => Effect.sync(() => onStdoutChunk(chunk))))
344 : child.stdout,
345 maxOutputBytes,
346 outputMode,
347 truncatedMarker,
348 }),
349 collectText({
350 command: input.command,
351 args: input.args,
352 cwd: input.cwd,
353 spawnCwd: input.spawnCwd,
354 streamName: "stderr",
355 stream: child.stderr,
356 maxOutputBytes,
357 outputMode,
358 truncatedMarker,
359 }),
360 writeStdin,
361 ],
362 { concurrency: "unbounded" },
363 );
364
365 const exitCode = yield* child.exitCode.pipe(
366 Effect.mapError(
367 (cause) =>
368 new ProcessReadError({
369 command: input.command,
370 argumentCount: input.args.length,
371 cwd: input.cwd,
372 spawnCwd: input.spawnCwd,
373 stream: "exitCode",
374 cause,
375 }),
376 ),
377 );
378
379 return {
380 stdout: stdout.text,
381 stderr: stderr.text,
382 code: exitCode,
383 timedOut: false,
384 stdoutTruncated: stdout.truncated,
385 stderrTruncated: stderr.truncated,
386 stdoutInvalidUtf8: stdout.invalidUtf8,
387 stderrInvalidUtf8: stderr.invalidUtf8,
388 } satisfies ProcessRunOutput;
389});
390
391/** @public Service construction is part of the canonical Effect module API. */
392export const make = Effect.fn("ProcessRunner.make")(function* () {
393 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
394
395 const run: ProcessRunner["Service"]["run"] = (input) =>
396 finalizeRunProcess(runProcessCore(spawner, input), input);
397
398 return ProcessRunner.of({
399 run,
400 });
401});
402
403export const layer = Layer.effect(ProcessRunner, make());