mirror of
https://github.com/anomalyco/opencode.git
synced 2026-07-24 12:15:51 -04:00
refactor(test/cli): migrate serve/acp builders to AppProcess.spawn
Slice 2 of the CLI harness Effect migration. Drops the last raw Bun.spawn call sites in withCliFixture. - `serve` and `acp` both move from `Effect.acquireRelease(Bun.spawn(...))` to `appProc.spawn(ChildProcess.make(...))`. The spawner's built-in acquireRelease finalizer handles SIGTERM on scope close — no manual wiring needed. - `handle.stdout` / `handle.stderr` are already Effect Streams, so the `fromBunStream` helper is gone (Stream.fromReadableStream + the per-pipe error-tag boilerplate it wrapped). - acp's stdin moves from imperative `proc.stdin.write` + `proc.stdin.end` to a Queue<Uint8Array> fed into the spawner's stdin Sink via Stream.fromQueue. `send` is `Queue.offer`, `close` is `Queue.shutdown` — shutdown propagates as stdin EOF, which is ACP's graceful-exit signal. - ServeHandle/AcpHandle public shape: `kill`/`close` become Effect<void> and `exited` becomes Effect<number> (was () => void and Promise<number>). The platform error that cross-spawn-spawner raises on signal-kill is collapsed to exit code -1 so `exited` stays a clean Effect<number> — matches the test contract (just needs proof of exit). Two consuming tests updated to yield the Effect instead of awaiting the Promise.
This commit is contained in:
@@ -51,19 +51,21 @@ describe("opencode acp (subprocess)", () => {
|
||||
"exits cleanly when stdin is closed (scope close)",
|
||||
({ opencode }) =>
|
||||
Effect.gen(function* () {
|
||||
const exitedPromise = yield* Effect.scoped(
|
||||
const exited = yield* Effect.scoped(
|
||||
Effect.gen(function* () {
|
||||
const acp = yield* opencode.acp()
|
||||
// Capture the Promise — scope-close fires the finalizer which
|
||||
// ends stdin, and ACP should exit gracefully.
|
||||
// Capture the Effect — scope-close shuts down stdinQueue, which
|
||||
// propagates as stdin EOF; ACP exits gracefully. The exitCode
|
||||
// Effect itself has no Scope requirement so yielding it after
|
||||
// scope close is safe.
|
||||
return acp.exited
|
||||
}),
|
||||
)
|
||||
|
||||
const code = yield* Effect.promise(() => exitedPromise)
|
||||
// Bun returns a number for normal exit. Anything goes for SIGTERM,
|
||||
// but we still require resolution within the test timeout.
|
||||
expect(typeof code === "number" || code === null).toBe(true)
|
||||
const code = yield* exited
|
||||
// Signal-killed processes surface as -1; clean EOF gives 0. Either
|
||||
// way we just need a number — proves the process exited.
|
||||
expect(typeof code).toBe("number")
|
||||
}),
|
||||
60_000,
|
||||
)
|
||||
|
||||
@@ -41,20 +41,20 @@ describe("opencode serve (subprocess)", () => {
|
||||
({ opencode }) =>
|
||||
Effect.gen(function* () {
|
||||
// Inner scope so we can observe `.exited` resolving after it closes.
|
||||
const exitedPromise = yield* Effect.scoped(
|
||||
const exited = yield* Effect.scoped(
|
||||
Effect.gen(function* () {
|
||||
const server = yield* opencode.serve()
|
||||
// Capture the Promise, not the resolved value — scope closes after
|
||||
// this gen returns, at which point the finalizer kills the child.
|
||||
// Capture the Effect, not its result — scope closes after this
|
||||
// gen returns, at which point the finalizer kills the child.
|
||||
// handle.exitCode itself has no Scope requirement, so yielding
|
||||
// it after scope close is fine.
|
||||
return server.exited
|
||||
}),
|
||||
)
|
||||
// After scope close: finalizer fired, process must have exited.
|
||||
const code = yield* Effect.promise(() => exitedPromise)
|
||||
// Bun reports the exit code; SIGTERM-killed processes return non-null
|
||||
// (typically 143 on POSIX). We just require resolution within a sane
|
||||
// window — anything else means the kill didn't take.
|
||||
expect(typeof code === "number" || code === null).toBe(true)
|
||||
// Signal-killed processes surface as -1 (see ServeHandle.exited).
|
||||
const code = yield* exited
|
||||
expect(typeof code).toBe("number")
|
||||
}),
|
||||
60_000,
|
||||
)
|
||||
|
||||
@@ -33,23 +33,13 @@ const cliEntry = path.join(opencodeRoot, "src/index.ts")
|
||||
|
||||
export const testModelID = "test/test-model"
|
||||
|
||||
// Wrap a Bun subprocess pipe (or any ReadableStream<Uint8Array>) as a Stream.
|
||||
// Centralizes the `evaluate` + `onError` boilerplate and tags errors with the
|
||||
// stream name so a stderr/stdout failure is greppable in logs.
|
||||
function fromBunStream(name: string, get: () => ReadableStream<Uint8Array>) {
|
||||
return Stream.fromReadableStream({
|
||||
evaluate: get,
|
||||
onError: (cause) => new Error(`${name} stream error: ${String(cause)}`),
|
||||
})
|
||||
}
|
||||
|
||||
// Long-lived processes (serve, acp) all want the same stderr drain: read every
|
||||
// chunk, push to a tail buffer, swallow stream errors (the child closing the
|
||||
// pipe is normal). `log: true` surfaces a real protocol error to logs so a
|
||||
// regression doesn't silently disappear.
|
||||
function forkStderrDrain(stream: ReadableStream<Uint8Array>, into: string[]) {
|
||||
function forkStderrDrain(stream: Stream.Stream<Uint8Array, unknown>, into: string[]) {
|
||||
return Effect.forkScoped(
|
||||
fromBunStream("stderr", () => stream).pipe(
|
||||
stream.pipe(
|
||||
Stream.decodeText(),
|
||||
Stream.runForEach((chunk) => Effect.sync(() => into.push(chunk))),
|
||||
Effect.ignore({ log: true }),
|
||||
@@ -117,9 +107,10 @@ export type ServeHandle = {
|
||||
readonly port: number
|
||||
// Sends SIGTERM. The scope finalizer also calls this, so tests rarely need
|
||||
// to invoke it directly — useful for tests that assert exit behavior.
|
||||
readonly kill: () => void
|
||||
// Resolves with the exit code once the process exits. Bun returns a number.
|
||||
readonly exited: Promise<number>
|
||||
readonly kill: Effect.Effect<void>
|
||||
// Resolves with the exit code once the process exits. Signal-killed
|
||||
// processes surface as -1 (vs cross-spawn-spawner raising a PlatformError).
|
||||
readonly exited: Effect.Effect<number>
|
||||
}
|
||||
|
||||
// `opencode acp` speaks newline-delimited JSON-RPC over stdin/stdout. It is
|
||||
@@ -140,8 +131,10 @@ export type AcpHandle = {
|
||||
readonly receive: Effect.Effect<unknown>
|
||||
// Closes stdin. ACP exits cleanly on stdin EOF; the scope finalizer also
|
||||
// calls this, so tests only need it when asserting exit behavior.
|
||||
readonly close: () => void
|
||||
readonly exited: Promise<number>
|
||||
readonly close: Effect.Effect<void>
|
||||
// Resolves with the exit code once the process exits. Signal-killed
|
||||
// processes surface as -1 (see ServeHandle.exited for the same convention).
|
||||
readonly exited: Effect.Effect<number>
|
||||
}
|
||||
|
||||
export type OpencodeCli = {
|
||||
@@ -256,29 +249,22 @@ export function withCliFixture<A, E>(
|
||||
if (opts?.hostname) argv.push("--hostname", opts.hostname)
|
||||
if (opts?.extraArgs) argv.push(...opts.extraArgs)
|
||||
|
||||
// Acquire the subprocess; release sends SIGTERM and awaits exit on
|
||||
// scope close. Wrapped in Effect.ignore so a flaky kill doesn't surface
|
||||
// as a finalizer error during test teardown.
|
||||
const proc = yield* Effect.acquireRelease(
|
||||
Effect.sync(() =>
|
||||
Bun.spawn(["bun", "run", "--conditions=browser", cliEntry, ...argv], {
|
||||
cwd: home,
|
||||
env: { ...process.env, ...env, ...opts?.env },
|
||||
stdout: "pipe",
|
||||
stderr: "pipe",
|
||||
}),
|
||||
),
|
||||
(p) =>
|
||||
Effect.promise(() => {
|
||||
p.kill()
|
||||
return p.exited
|
||||
}).pipe(Effect.ignore),
|
||||
// ChildProcessSpawner.spawn returns a scoped handle whose acquireRelease
|
||||
// finalizer sends SIGTERM and awaits exit on scope close — same lifecycle
|
||||
// the old Bun.spawn + manual acquireRelease wrapper gave us, no plumbing.
|
||||
const handle = yield* appProc.spawn(
|
||||
ChildProcess.make("bun", ["run", "--conditions=browser", cliEntry, ...argv], {
|
||||
cwd: home,
|
||||
env: { ...env, ...opts?.env },
|
||||
extendEnv: true,
|
||||
stdin: "ignore",
|
||||
}),
|
||||
)
|
||||
|
||||
// Tail buffer so timeout failures can include stderr context. The fork
|
||||
// also keeps the OS pipe buffer from filling and wedging the child.
|
||||
const stderrChunks: string[] = []
|
||||
yield* forkStderrDrain(proc.stderr, stderrChunks)
|
||||
yield* forkStderrDrain(handle.stderr, stderrChunks)
|
||||
|
||||
// Watch stdout line-by-line for the listening sentinel. Format
|
||||
// (see src/cli/cmd/serve.ts):
|
||||
@@ -286,7 +272,7 @@ export function withCliFixture<A, E>(
|
||||
const readyRe = /listening on (http:\/\/([^\s:]+):(\d+))/
|
||||
const readyDeferred = yield* Deferred.make<{ url: string; hostname: string; port: number }>()
|
||||
yield* Effect.forkScoped(
|
||||
fromBunStream("stdout", () => proc.stdout).pipe(
|
||||
handle.stdout.pipe(
|
||||
Stream.decodeText(),
|
||||
Stream.splitLines,
|
||||
Stream.runForEach((line) => {
|
||||
@@ -315,10 +301,14 @@ export function withCliFixture<A, E>(
|
||||
url: match.url,
|
||||
hostname: match.hostname,
|
||||
port: match.port,
|
||||
kill: () => {
|
||||
proc.kill()
|
||||
},
|
||||
exited: proc.exited as Promise<number>,
|
||||
kill: handle.kill().pipe(Effect.ignore),
|
||||
// handle.exitCode fails with PlatformError if the process was killed
|
||||
// by signal (the normal scope-close path). Swallow → -1 so the test
|
||||
// can still distinguish exited-vs-not without crashing.
|
||||
exited: handle.exitCode.pipe(
|
||||
Effect.orElseSucceed(() => -1),
|
||||
Effect.map((c) => Number(c)),
|
||||
),
|
||||
} satisfies ServeHandle
|
||||
})
|
||||
|
||||
@@ -327,47 +317,30 @@ export function withCliFixture<A, E>(
|
||||
if (opts?.cwd) argv.push("--cwd", opts.cwd)
|
||||
if (opts?.extraArgs) argv.push(...opts.extraArgs)
|
||||
|
||||
// Acquire the subprocess. Release ends stdin (clean shutdown — ACP exits
|
||||
// on stdin EOF) and falls back to SIGTERM if it doesn't exit promptly.
|
||||
// Either way we await proc.exited so the test scope doesn't leak.
|
||||
const proc = yield* Effect.acquireRelease(
|
||||
Effect.sync(() =>
|
||||
Bun.spawn(["bun", "run", "--conditions=browser", cliEntry, ...argv], {
|
||||
cwd: opts?.cwd ?? home,
|
||||
env: { ...process.env, ...env, ...opts?.env },
|
||||
stdin: "pipe",
|
||||
stdout: "pipe",
|
||||
stderr: "pipe",
|
||||
}),
|
||||
),
|
||||
(p) =>
|
||||
// Graceful shutdown: close stdin (ACP exits on EOF), give it a
|
||||
// window to exit, then SIGTERM. The Effect.timeoutOrElse expresses
|
||||
// exactly that race without raw setTimeout or Promise.race.
|
||||
Effect.gen(function* () {
|
||||
yield* Effect.sync(() => p.stdin.end())
|
||||
yield* Effect.promise(() => p.exited).pipe(
|
||||
Effect.timeoutOrElse({
|
||||
duration: Duration.seconds(2),
|
||||
orElse: () =>
|
||||
Effect.sync(() => {
|
||||
p.kill()
|
||||
}),
|
||||
}),
|
||||
)
|
||||
yield* Effect.promise(() => p.exited)
|
||||
}).pipe(Effect.ignore),
|
||||
// stdin is fed by a Queue<Uint8Array>: send() offers bytes, close()
|
||||
// shuts the queue down. The spawner drains the Queue-backed Stream into
|
||||
// the child's stdin Sink (endOnDone: true by default), so a queue
|
||||
// shutdown propagates as stdin EOF → ACP exits gracefully. Scope-close
|
||||
// is the backstop via the spawner's kill finalizer.
|
||||
const stdinQueue = yield* Queue.unbounded<Uint8Array>()
|
||||
const handle = yield* appProc.spawn(
|
||||
ChildProcess.make("bun", ["run", "--conditions=browser", cliEntry, ...argv], {
|
||||
cwd: opts?.cwd ?? home,
|
||||
env: { ...env, ...opts?.env },
|
||||
extendEnv: true,
|
||||
stdin: Stream.fromQueue(stdinQueue),
|
||||
}),
|
||||
)
|
||||
|
||||
const stderrChunks: string[] = []
|
||||
yield* forkStderrDrain(proc.stderr, stderrChunks)
|
||||
yield* forkStderrDrain(handle.stderr, stderrChunks)
|
||||
|
||||
// Each ndjson line becomes one queue entry. JSON.parse failures are
|
||||
// surfaced as the raw string so a malformed protocol message doesn't
|
||||
// silently wedge the test in `receive`.
|
||||
const responses = yield* Queue.unbounded<unknown>()
|
||||
yield* Effect.forkScoped(
|
||||
fromBunStream("stdout", () => proc.stdout).pipe(
|
||||
handle.stdout.pipe(
|
||||
Stream.decodeText(),
|
||||
Stream.splitLines,
|
||||
Stream.runForEach((line) => {
|
||||
@@ -384,20 +357,20 @@ export function withCliFixture<A, E>(
|
||||
),
|
||||
)
|
||||
|
||||
const encoder = new TextEncoder()
|
||||
return {
|
||||
// `proc.stdin.write` returns `number | Promise<number>`. The promise
|
||||
// form is the backpressure signal — if we don't await it, rapid
|
||||
// successive sends can interleave under pipe-buffer-full conditions
|
||||
// and corrupt the ndjson framing.
|
||||
send: (msg: object) =>
|
||||
Effect.promise(async () => {
|
||||
const ret = proc.stdin.write(JSON.stringify(msg) + "\n")
|
||||
if (typeof ret !== "number") await ret
|
||||
}),
|
||||
Queue.offer(stdinQueue, encoder.encode(JSON.stringify(msg) + "\n")).pipe(Effect.asVoid),
|
||||
receive: Queue.take(responses),
|
||||
// proc.stdin.end() is idempotent in Bun; no try/catch needed.
|
||||
close: () => proc.stdin.end(),
|
||||
exited: proc.exited as Promise<number>,
|
||||
// Queue shutdown → Stream.fromQueue completes → spawner ends stdin.
|
||||
// Idempotent: shutting down an already-shut-down queue is a no-op.
|
||||
close: Queue.shutdown(stdinQueue).pipe(Effect.asVoid),
|
||||
// handle.exitCode fails with PlatformError on signal-kill; collapse
|
||||
// to -1 to match the ServeHandle.exited convention.
|
||||
exited: handle.exitCode.pipe(
|
||||
Effect.orElseSucceed(() => -1),
|
||||
Effect.map((c) => Number(c)),
|
||||
),
|
||||
} satisfies AcpHandle
|
||||
})
|
||||
|
||||
|
||||
Reference in New Issue
Block a user