1import * as NodeUtil from "node:util";
2
3import * as Cause from "effect/Cause";
4import * as Fiber from "effect/Fiber";
5import * as Effect from "effect/Effect";
6import * as Queue from "effect/Queue";
7import * as Ref from "effect/Ref";
8import * as Sink from "effect/Sink";
9import * as Stdio from "effect/Stdio";
10import * as Stream from "effect/Stream";
11import { ChildProcessSpawner } from "effect/process";
12
13import * as CodexError from "../errors.ts";
14
15const encoder = new TextEncoder();
16
17export const makeChildStdio = (
18 handle: Pick<ChildProcessSpawner.ChildProcessHandle, "stdin" | "stdout">,
19 stderrDrained: Effect.Effect<void>,
20) =>
21 Stdio.make({
22 args: Effect.succeed([]),
23 // Finish stderr separately from the protocol's exit-status deadline.
24 stdin: handle.stdout.pipe(Stream.onEnd(stderrDrained)),
25 stdout: () =>
26 Sink.mapInput(handle.stdin, (chunk: string | Uint8Array) =>
27 typeof chunk === "string" ? encoder.encode(chunk) : chunk,
28 ),
29 stderr: () => Sink.drain,
30 });
31
32export const makeInMemoryStdio = Effect.fn("makeInMemoryStdio")(function* () {
33 const input = yield* Queue.unbounded<Uint8Array, Cause.Done<void>>();
34 const output = yield* Queue.unbounded<string>();
35 const decoder = new TextDecoder();
36
37 return {
38 stdio: Stdio.make({
39 args: Effect.succeed([]),
40 stdin: Stream.fromQueue(input),
41 stdout: () =>
42 Sink.forEach((chunk: string | Uint8Array) =>
43 Queue.offer(
44 output,
45 typeof chunk === "string" ? chunk : decoder.decode(chunk, { stream: true }),
46 ),
47 ),
48 stderr: () => Sink.drain,
49 }),
50 input,
51 output,
52 };
53});
54
55const STDERR_TAIL_MAX_CHARS = 8_192;
56const STDERR_EXCERPT_MAX_CHARS = 512;
57// `Error: …`, `TypeError: …`, `Error [ERR_MODULE_NOT_FOUND]: …`, `error: …`, `ERROR: …`, `fatal: …`
58const ERROR_LINE_PATTERN = /^(?:[a-z]*error|fatal)(?: ?\[[^\]]+\])?:\s*\S/i;
59const EXIT_SUMMARY_PATTERN = /\b(?:exit status|exit code|exited with code)\s+\d+\.?$/i;
60
61export const appendStderrTail = (current: string, chunk: string): string => {
62 const next = `${current}${chunk}`;
63 return next.length <= STDERR_TAIL_MAX_CHARS ? next : next.slice(-STDERR_TAIL_MAX_CHARS);
64};
65
66// Node launchers print the actionable error between a source excerpt and stack frames.
67export const stderrExcerpt = (tail: string): string | undefined => {
68 const lines = NodeUtil.stripVTControlCharacters(tail)
69 .split(/\r?\n/)
70 .map((line) => line.trim().replace(/\s+/g, " "))
71 .filter((line) => line.length > 0);
72 const errorIndex = lines.findLastIndex((candidate) => ERROR_LINE_PATTERN.test(candidate));
73 const lineIndex =
74 errorIndex === -1 || EXIT_SUMMARY_PATTERN.test(lines.at(-1) ?? "")
75 ? lines.length - 1
76 : errorIndex;
77 const line = lines[lineIndex];
78 if (line === undefined) {
79 return undefined;
80 }
81 // Wrappers can print the actual failure before their generic exit-status summary.
82 const excerpt = EXIT_SUMMARY_PATTERN.test(line)
83 ? lines.slice(Math.max(0, lineIndex - 2), lineIndex + 1).join(" | ")
84 : line;
85 if (excerpt.length <= STDERR_EXCERPT_MAX_CHARS) {
86 return excerpt;
87 }
88 return excerpt !== line
89 ? `…${excerpt.slice(-(STDERR_EXCERPT_MAX_CHARS - 1))}`
90 : `${excerpt.slice(0, STDERR_EXCERPT_MAX_CHARS - 1)}…`;
91};
92
93export const makeStderrTail = Effect.fn("makeStderrTail")(function* (
94 stderr: ChildProcessSpawner.ChildProcessHandle["stderr"],
95) {
96 const tail = yield* Ref.make("");
97 const drain = yield* stderr.pipe(
98 Stream.decodeText(),
99 Stream.runForEach((chunk) => Ref.update(tail, (current) => appendStderrTail(current, chunk))),
100 Effect.ignore,
101 Effect.forkScoped,
102 );
103 return {
104 read: Ref.get(tail),
105 awaitDrained: Fiber.join(drain).pipe(Effect.timeout("250 millis"), Effect.ignore),
106 };
107});
108
109type ChildProcessTerminationHandle = Pick<
110 ChildProcessSpawner.ChildProcessHandle,
111 "exitCode" | "pid"
112>;
113
114export const makeTerminationError = (
115 handle: ChildProcessTerminationHandle,
116 stderrTail: Effect.Effect<string>,
117): Effect.Effect<CodexError.CodexAppServerError> =>
118 Effect.matchEffect(handle.exitCode, {
119 onFailure: (cause) =>
120 Effect.succeed(
121 new CodexError.CodexAppServerTransportError({
122 operation: "read-process-exit-status",
123 pid: handle.pid,
124 cause,
125 }),
126 ),
127 onSuccess: (code) =>
128 Effect.map(stderrTail, (tail) => {
129 const stderr = stderrExcerpt(tail);
130 return new CodexError.CodexAppServerProcessExitedError({
131 code,
132 pid: handle.pid,
133 ...(stderr ? { stderr } : {}),
134 });
135 }),
136 });