1import * as Cause from "effect/Cause";
2import * as Effect from "effect/Effect";
3import * as Queue from "effect/Queue";
4import * as Sink from "effect/Sink";
5import * as Stdio from "effect/Stdio";
6import * as Stream from "effect/Stream";
7import { ChildProcessSpawner } from "effect/process";
8
9import * as CodexError from "../errors.ts";
10
11const encoder = new TextEncoder();
12
13export const makeChildStdio = (handle: ChildProcessSpawner.ChildProcessHandle) =>
14 Stdio.make({
15 args: Effect.succeed([]),
16 stdin: handle.stdout,
17 stdout: () =>
18 Sink.mapInput(handle.stdin, (chunk: string | Uint8Array) =>
19 typeof chunk === "string" ? encoder.encode(chunk) : chunk,
20 ),
21 stderr: () => Sink.drain,
22 });
23
24export const makeInMemoryStdio = Effect.fn("makeInMemoryStdio")(function* () {
25 const input = yield* Queue.unbounded<Uint8Array, Cause.Done<void>>();
26 const output = yield* Queue.unbounded<string>();
27 const decoder = new TextDecoder();
28
29 return {
30 stdio: Stdio.make({
31 args: Effect.succeed([]),
32 stdin: Stream.fromQueue(input),
33 stdout: () =>
34 Sink.forEach((chunk: string | Uint8Array) =>
35 Queue.offer(
36 output,
37 typeof chunk === "string" ? chunk : decoder.decode(chunk, { stream: true }),
38 ),
39 ),
40 stderr: () => Sink.drain,
41 }),
42 input,
43 output,
44 };
45});
46
47type ChildProcessTerminationHandle = Pick<
48 ChildProcessSpawner.ChildProcessHandle,
49 "exitCode" | "pid"
50>;
51
52export const makeTerminationError = (
53 handle: ChildProcessTerminationHandle,
54): Effect.Effect<CodexError.CodexAppServerError> =>
55 Effect.match(handle.exitCode, {
56 onFailure: (cause) =>
57 new CodexError.CodexAppServerTransportError({
58 operation: "read-process-exit-status",
59 pid: handle.pid,
60 cause,
61 }),
62 onSuccess: (code) => new CodexError.CodexAppServerProcessExitedError({ code, pid: handle.pid }),
63 });