openAgent

1import { Context, Duration, Effect, Fiber, Layer, Schema, Stream } from "effect"
2import type { PlatformError } from "effect/PlatformError"
3import { ChildProcess } from "effect/unstable/process"
4import { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner"
5import { CrossSpawnSpawner } from "./cross-spawn-spawner"
6import { makeGlobalNode } from "./effect/app-node"
7 
8export class AppProcessError extends Schema.TaggedErrorClass<AppProcessError>()("AppProcessError", {
9  command: Schema.String,
10  exitCode: Schema.optional(Schema.Number),
11  stderr: Schema.optional(Schema.String),
12  cause: Schema.optional(Schema.Defect()),
13}) {
14  override get message() {
15    const detail =
16      this.stderr?.trim() || (this.cause instanceof Error ? this.cause.message : this.cause && String(this.cause))
17    const status = this.exitCode === undefined ? "" : ` (exit ${this.exitCode})`
18    return `Command failed${status}: ${this.command}${detail ? `: ${detail}` : ""}`
19  }
20}
21 
22export interface RunOptions {
23  readonly combineOutput?: boolean
24  readonly maxOutputBytes?: number
25  readonly maxErrorBytes?: number
26  readonly signal?: AbortSignal
27  readonly timeout?: Duration.Input
28  readonly stdin?: string | Uint8Array | Stream.Stream<Uint8Array, PlatformError>
29}
30 
31export interface RunStreamOptions {
32  readonly signal?: AbortSignal
33  readonly includeStderr?: boolean
34  readonly okExitCodes?: ReadonlyArray<number>
35  readonly maxErrorBytes?: number
36}
37 
38export interface RunResult {
39  readonly command: string
40  readonly exitCode: number
41  readonly output?: Buffer
42  readonly stdout: Buffer
43  readonly stderr: Buffer
44  readonly outputTruncated?: boolean
45  readonly stdoutTruncated: boolean
46  readonly stderrTruncated: boolean
47}
48 
49export type Interface = ChildProcessSpawner["Service"] & {
50  readonly run: (command: ChildProcess.Command, options?: RunOptions) => Effect.Effect<RunResult, AppProcessError>
51  readonly runStream: (
52    command: ChildProcess.Command,
53    options?: RunStreamOptions,
54  ) => Stream.Stream<string, AppProcessError>
55}
56 
57export class Service extends Context.Service<Service, Interface>()("@opencode/AppProcess") {}
58 
59export const requireSuccess = (result: RunResult): Effect.Effect<RunResult, AppProcessError> =>
60  result.exitCode === 0
61    ? Effect.succeed(result)
62    : Effect.fail(
63        new AppProcessError({
64          command: result.command,
65          exitCode: result.exitCode,
66          stderr: result.stderr.toString("utf8"),
67        }),
68      )
69 
70export const requireExitIn =
71  (codes: ReadonlyArray<number>) =>
72  (result: RunResult): Effect.Effect<RunResult, AppProcessError> =>
73    codes.includes(result.exitCode)
74      ? Effect.succeed(result)
75      : Effect.fail(
76          new AppProcessError({
77            command: result.command,
78            exitCode: result.exitCode,
79            stderr: result.stderr.toString("utf8"),
80          }),
81        )
82 
83const describeCommand = (command: ChildProcess.Command): string => {
84  if (command._tag === "StandardCommand") {
85    return command.args.length ? `${command.command} ${command.args.join(" ")}` : command.command
86  }
87  return `${describeCommand(command.left)} | ${describeCommand(command.right)}`
88}
89 
90const wrapError = (description: string, cause: unknown): AppProcessError =>
91  cause instanceof AppProcessError ? cause : new AppProcessError({ command: description, cause })
92 
93export const abortError = (signal: AbortSignal): Error => {
94  const reason = signal.reason
95  if (reason instanceof Error) return reason
96  const err = new Error("Aborted")
97  err.name = "AbortError"
98  return err
99}
100 
101export const waitForAbort = (signal: AbortSignal) =>
102  Effect.callback<never, Error>((resume) => {
103    if (signal.aborted) {
104      resume(Effect.fail(abortError(signal)))
105      return
106    }
107    const onabort = () => resume(Effect.fail(abortError(signal)))
108    signal.addEventListener("abort", onabort, { once: true })
109    return Effect.sync(() => signal.removeEventListener("abort", onabort))
110  })
111 
112const normalizeStdin = (
113  input: string | Uint8Array | Stream.Stream<Uint8Array, PlatformError>,
114): Stream.Stream<Uint8Array, PlatformError> =>
115  typeof input === "string"
116    ? Stream.make(new TextEncoder().encode(input))
117    : input instanceof Uint8Array
118      ? Stream.make(input)
119      : input
120 
121export const collectStream = (stream: Stream.Stream<Uint8Array, PlatformError>, maxOutputBytes: number | undefined) =>
122  Stream.runFold(
123    stream,
124    () => ({ chunks: [] as Uint8Array[], bytes: 0, truncated: false }),
125    (acc, chunk) => {
126      if (maxOutputBytes === undefined) {
127        acc.chunks.push(chunk)
128        acc.bytes += chunk.length
129        return acc
130      }
131      const remaining = maxOutputBytes - acc.bytes
132      if (remaining > 0) acc.chunks.push(remaining >= chunk.length ? chunk : chunk.slice(0, remaining))
133      acc.bytes += chunk.length
134      acc.truncated = acc.truncated || acc.bytes > maxOutputBytes
135      return acc
136    },
137  ).pipe(Effect.map((x) => ({ buffer: Buffer.concat(x.chunks), truncated: x.truncated })))
138 
139const layer = Layer.effect(
140  Service,
141  Effect.gen(function* () {
142    const spawner = yield* ChildProcessSpawner
143 
144    const runCommand = (command: ChildProcess.Command, options?: RunOptions) => {
145      const description = describeCommand(command)
146      const collect = Effect.scoped(
147        Effect.gen(function* () {
148          const handle = yield* spawner.spawn(command)
149          if (options?.combineOutput) {
150            const [output, exitCode] = yield* Effect.all(
151              [collectStream(handle.all, options.maxOutputBytes), handle.exitCode],
152              { concurrency: "unbounded" },
153            )
154            return {
155              command: description,
156              exitCode,
157              output: output.buffer,
158              stdout: Buffer.alloc(0),
159              stderr: Buffer.alloc(0),
160              outputTruncated: output.truncated,
161              stdoutTruncated: false,
162              stderrTruncated: false,
163            } satisfies RunResult
164          }
165          const [stdout, stderr, exitCode] = yield* Effect.all(
166            [
167              collectStream(handle.stdout, options?.maxOutputBytes),
168              collectStream(handle.stderr, options?.maxErrorBytes),
169              handle.exitCode,
170            ],
171            { concurrency: "unbounded" },
172          )
173          return {
174            command: description,
175            exitCode,
176            stdout: stdout.buffer,
177            stderr: stderr.buffer,
178            stdoutTruncated: stdout.truncated,
179            stderrTruncated: stderr.truncated,
180          } satisfies RunResult
181        }),
182      )
183      const timed = options?.timeout
184        ? Effect.timeoutOrElse(collect, {
185            duration: options.timeout,
186            orElse: () => Effect.fail(new AppProcessError({ command: description, cause: new Error("Timed out") })),
187          })
188        : collect
189      const aborted = options?.signal
190        ? timed.pipe(
191            Effect.raceFirst(
192              waitForAbort(options.signal).pipe(Effect.mapError((cause) => wrapError(description, cause))),
193            ),
194          )
195        : timed
196      return aborted.pipe(Effect.catch((cause) => Effect.fail(wrapError(description, cause))))
197    }
198 
199    const run = Effect.fn("AppProcess.run")(function* (command: ChildProcess.Command, options?: RunOptions) {
200      if (options?.stdin === undefined) return yield* runCommand(command, options)
201      if (command._tag !== "StandardCommand") {
202        return yield* new AppProcessError({
203          command: describeCommand(command),
204          cause: new Error("stdin option only supports StandardCommand; received PipedCommand"),
205        })
206      }
207      const next = ChildProcess.make(command.command, command.args, {
208        ...command.options,
209        stdin: normalizeStdin(options.stdin),
210      })
211      return yield* runCommand(next, options)
212    })
213 
214    const runStream = (
215      command: ChildProcess.Command,
216      options?: RunStreamOptions,
217    ): Stream.Stream<string, AppProcessError> => {
218      const description = describeCommand(command)
219      const okExitCodes = options?.okExitCodes
220      const built: Stream.Stream<string, AppProcessError | PlatformError> = Stream.unwrap(
221        Effect.gen(function* () {
222          const handle = yield* spawner.spawn(command)
223          const stderrFiber = yield* Effect.forkScoped(
224            collectStream(handle.stderr, options?.maxErrorBytes).pipe(Effect.map((x) => x.buffer.toString("utf8"))),
225          )
226          const source = options?.includeStderr === true ? handle.all : handle.stdout
227          const lines = source.pipe(
228            Stream.decodeText,
229            Stream.splitLines,
230            Stream.filter((line) => line.length > 0),
231          )
232          const tail = Stream.unwrap(
233            Effect.gen(function* () {
234              const code = yield* handle.exitCode
235              if (okExitCodes && okExitCodes.length > 0 && !okExitCodes.includes(code)) {
236                const stderr = yield* Fiber.join(stderrFiber)
237                return Stream.fail(new AppProcessError({ command: description, exitCode: code, stderr }))
238              }
239              return Stream.empty
240            }),
241          )
242          return Stream.concat(lines, tail) as Stream.Stream<string, AppProcessError | PlatformError>
243        }),
244      )
245      const mapped = built.pipe(
246        Stream.catch((cause): Stream.Stream<string, AppProcessError> => Stream.fail(wrapError(description, cause))),
247      )
248      if (!options?.signal) return mapped
249      const signal = options.signal
250      return mapped.pipe(
251        Stream.interruptWhen(waitForAbort(signal).pipe(Effect.mapError((cause) => wrapError(description, cause)))),
252      )
253    }
254 
255    return Service.of({ ...spawner, run, runStream })
256  }),
257)
258 
259export const node = makeGlobalNode({ service: Service, layer: layer, deps: [CrossSpawnSpawner.node] })
260 
261export * as AppProcess from "./process"
262