ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(core): settle owned process output

Browse source
This commit is contained in:
Kit Langton 2026-06-05 21:09:02 -04:00
commit 8ccc575747
8 changed files with 366 additions and 123 deletions
Download patch file Download diff file Expand all files Collapse all files

30
packages/core/src/cross-spawn-spawner.ts
View file

@ -24,6 +24,7 @@ import {
import * as NodeChildProcess from "node:child_process"
import { PassThrough } from "node:stream"
import launch from "cross-spawn"
import { ProcessOutput } from "./process-output"
const toError = (err: unknown): Error => (err instanceof globalThis.Error ? err : new globalThis.Error(String(err)))
@ -266,18 +267,11 @@ export const make = Effect.gen(function* () {
Effect.callback<readonly [NodeChildProcess.ChildProcess, ExitSignal], PlatformError.PlatformError>((resume) => {
const signal = Deferred.makeUnsafe<readonly [code: number | null, signal: NodeJS.Signals | null]>()
const proc = launch(command.command, command.args, opts)
let end = false
let exit: readonly [code: number | null, signal: NodeJS.Signals | null] | undefined
proc.on("error", (err) => {
resume(Effect.fail(toPlatformError("spawn", err, command)))
})
proc.on("exit", (...args) => {
exit = args
})
proc.on("close", (...args) => {
if (end) return
end = true
Deferred.doneUnsafe(signal, Exit.succeed(exit ?? args))
Deferred.doneUnsafe(signal, Exit.succeed(args))
})
proc.on("spawn", () => {
resume(Effect.succeed([proc, signal]))
@ -340,6 +334,16 @@ export const make = Effect.gen(function* () {
})
}
const groupAlive = (proc: NodeChildProcess.ChildProcess) => {
if (process.platform === "win32") return false
try {
process.kill(-proc.pid!, 0)
return true
} catch {
return false
}
}
const source = (handle: ChildProcessHandle, from: ChildProcess.PipeFromOption | undefined) => {
const opt = from ?? "stdout"
switch (opt) {
@ -381,9 +385,11 @@ export const make = Effect.gen(function* () {
const done = yield* Deferred.isDone(signal)
const kill = timeout(proc, command, command.options)
if (done) {
const [code] = yield* Deferred.await(signal)
if (process.platform === "win32") return yield* Effect.void
if (code !== 0 && Predicate.isNotNull(code)) return yield* Effect.ignore(kill(killGroup))
if (!groupAlive(proc)) return yield* Effect.void
yield* Effect.ignore(killGroup(command, proc, command.options.killSignal ?? "SIGTERM"))
yield* Effect.sleep("100 millis")
if (groupAlive(proc)) yield* Effect.ignore(killGroup(command, proc, "SIGKILL"))
return yield* Effect.void
}
const send = (s: NodeJS.Signals) =>
@ -403,7 +409,7 @@ export const make = Effect.gen(function* () {
const fd = yield* setupFds(command, proc, extra)
const out = setupOutput(command, proc, sout, serr)
let ref = true
return makeHandle({
const handle = makeHandle({
pid: ProcessId(proc.pid!),
stdin: yield* setupStdin(command, proc, sin),
stdout: out.stdout,
@ -446,6 +452,8 @@ export const make = Effect.gen(function* () {
})
}),
})
ProcessOutput.register(handle, proc)
return handle
}
case "PipedCommand": {
const flat = flatten(command)

68
packages/core/src/process-output.ts Normal file
View file

@ -0,0 +1,68 @@
import { Cause, Deferred, Duration, Effect, Exit, Fiber } from "effect"
import type { ChildProcess } from "node:child_process"
import type { ChildProcessHandle } from "effect/unstable/process/ChildProcessSpawner"
const processes = new WeakMap<ChildProcessHandle, ChildProcess>()
export const register = (handle: ChildProcessHandle, process: ChildProcess) => {
processes.set(handle, process)
}
export const drain = <E, R>(
handle: ChildProcessHandle,
drains: ReadonlyArray<Effect.Effect<unknown, E, R>>,
options?: { readonly grace?: Duration.Input; readonly onClose?: () => void },
) =>
Effect.uninterruptibleMask((restore) =>
Effect.gen(function* () {
const fibers = yield* Effect.forEach(drains, (drain) => Effect.forkDetach(drain))
let closed = false
const close = Effect.sync(() => {
if (closed) return
closed = true
options?.onClose?.()
const process = processes.get(handle)
process?.stdout?.destroy()
process?.stderr?.destroy()
for (const stream of process?.stdio.slice(3) ?? []) {
if (stream && "readable" in stream && stream.readable) stream.destroy()
}
for (const fiber of fibers) fiber.interruptUnsafe()
})
const failed = yield* Deferred.make<Cause.Cause<E>>()
const observers = fibers.map((fiber) =>
fiber.addObserver((exit) => {
if (Exit.isFailure(exit)) Deferred.doneUnsafe(failed, Effect.succeed(exit.cause))
}),
)
const run = Effect.gen(function* () {
const outcome = yield* Effect.raceAllFirst([
handle.exitCode.pipe(
Effect.exit,
Effect.map((exit) => ({ type: "exit" as const, exit })),
),
Deferred.await(failed).pipe(Effect.map((cause) => ({ type: "failure" as const, cause }))),
])
for (const remove of observers) remove()
if (outcome.type === "failure") {
yield* close
yield* handle.kill({ forceKillAfter: "1 second" }).pipe(Effect.ignore)
return yield* Effect.failCause(outcome.cause)
}
const exits = yield* Effect.forEach(fibers, (fiber) => Fiber.await(fiber), { concurrency: "unbounded" }).pipe(
Effect.timeoutOrElse({ duration: options?.grace ?? "1 second", orElse: () => Effect.succeed(undefined) }),
)
if (exits) {
const failure = exits.find((exit) => Exit.isFailure(exit))
if (failure) return yield* Effect.failCause(failure.cause)
} else {
yield* close
}
return Exit.isFailure(outcome.exit) ? yield* Effect.failCause(outcome.exit.cause) : outcome.exit.value
})
return yield* restore(run).pipe(Effect.onInterrupt(() => close))
}),
)
export * as ProcessOutput from "./process-output"

35
packages/core/src/process.ts
View file

@ -3,6 +3,7 @@ import type { PlatformError } from "effect/PlatformError"
import { ChildProcess } from "effect/unstable/process"
import { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner"
import { CrossSpawnSpawner } from "./cross-spawn-spawner"
import { ProcessOutput } from "./process-output"
export class AppProcessError extends Schema.TaggedErrorClass<AppProcessError>()("AppProcessError", {
command: Schema.String,
@ -125,6 +126,19 @@ export const collectStream = (stream: Stream.Stream<Uint8Array, PlatformError>,
},
).pipe(Effect.map((x) => ({ buffer: Buffer.concat(x.chunks), truncated: x.truncated })))
const collector = (stream: Stream.Stream<Uint8Array, PlatformError>, maxBytes: number | undefined) => {
const state = { chunks: [] as Uint8Array[], bytes: 0, truncated: false }
const drain = Stream.runForEach(stream, (chunk) =>
Effect.sync(() => {
const remaining = maxBytes === undefined ? chunk.length : maxBytes - state.bytes
if (remaining > 0) state.chunks.push(remaining >= chunk.length ? chunk : chunk.slice(0, remaining))
state.bytes += chunk.length
state.truncated = maxBytes !== undefined && state.bytes > maxBytes
}),
)
return { drain, result: () => ({ buffer: Buffer.concat(state.chunks), truncated: state.truncated }) }
}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
@ -135,21 +149,18 @@ export const layer = Layer.effect(
const collect = Effect.scoped(
Effect.gen(function* () {
const handle = yield* spawner.spawn(command)
const [stdout, stderr, exitCode] = yield* Effect.all(
[
collectStream(handle.stdout, options?.maxOutputBytes),
collectStream(handle.stderr, options?.maxErrorBytes),
handle.exitCode,
],
{ concurrency: "unbounded" },
)
const stdout = collector(handle.stdout, options?.maxOutputBytes)
const stderr = collector(handle.stderr, options?.maxErrorBytes)
const exitCode = yield* ProcessOutput.drain(handle, [stdout.drain, stderr.drain])
const out = stdout.result()
const err = stderr.result()
return {
command: description,
exitCode,
stdout: stdout.buffer,
stderr: stderr.buffer,
stdoutTruncated: stdout.truncated,
stderrTruncated: stderr.truncated,
stdout: out.buffer,
stderr: err.buffer,
stdoutTruncated: out.truncated,
stderrTruncated: err.truncated,
} satisfies RunResult
}),
)

23
packages/core/test/effect/cross-spawn-spawner.test.ts
View file

@ -6,6 +6,7 @@ import { Effect, Exit, Stream } from "effect"
import type * as PlatformError from "effect/PlatformError"
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
import { ProcessOutput } from "@opencode-ai/core/process-output"
import { testEffect } from "../lib/effect"
const live = CrossSpawnSpawner.defaultLayer
@ -96,6 +97,28 @@ describe("cross-spawn spawner", () => {
expect(code).toBe(ChildProcessSpawner.ExitCode(42))
}),
)
fx.live(
"reports direct exit while a descendant holds stdout open",
Effect.gen(function* () {
if (process.platform === "win32") return
const handle = yield* ChildProcess.make("sh", ["-c", "sleep 30 &"])
expect(yield* handle.exitCode.pipe(Effect.timeout("1 second"))).toBe(ChildProcessSpawner.ExitCode(0))
}),
)
fx.live(
"terminates the process when an owned drain fails before exit",
Effect.gen(function* () {
const handle = yield* js("setInterval(() => {}, 10_000)")
const pid = Number(handle.pid)
const exit = yield* ProcessOutput.drain(handle, [Effect.fail("drain failed")]).pipe(Effect.exit)
expect(Exit.isFailure(exit)).toBe(true)
expect(yield* Effect.promise(() => gone(pid))).toBe(true)
}),
5_000,
)
})
describe("cwd option", () => {

29
packages/core/test/process/process.test.ts
View file

@ -25,6 +25,19 @@ const waitForFile = (file: string) =>
}
})
const gone = (pid: number) =>
Effect.promise(async () => {
for (let i = 0; i < 200; i++) {
try {
process.kill(pid, 0)
} catch {
return true
}
await new Promise<void>((resolve) => setTimeout(resolve, 25))
}
return false
})
describe("AppProcess", () => {
describe("run", () => {
it.effect(
@ -134,6 +147,22 @@ describe("AppProcess", () => {
)
if (process.platform !== "win32") {
it.live(
"captures large output and cleans a descendant holding stdout open",
Effect.gen(function* () {
const svc = yield* AppProcess.Service
const size = 8 * 1024 * 1024
const command = `sleep 30 & child=$!; printf "%s\\n" "$child"; dd if=/dev/zero bs=${size} count=1 2>/dev/null | tr '\\0' x`
const result = yield* svc.run(ChildProcess.make("sh", ["-c", command]))
const newline = result.stdout.indexOf(10)
const pid = Number(result.stdout.subarray(0, newline).toString("utf8"))
expect(result.stdout.length).toBe(newline + 1 + size)
expect(yield* gone(pid)).toBe(true)
}),
8_000,
)
it.live(
"timeout cleans up the scoped child process",
Effect.acquireUseRelease(