diff --git a/packages/opencode/specs/effect-migration.md b/packages/opencode/specs/effect-migration.md index 74a4712e2..ba3b11347 100644 --- a/packages/opencode/specs/effect-migration.md +++ b/packages/opencode/specs/effect-migration.md @@ -308,7 +308,9 @@ Current raw fs users that will convert during tool migration: - [ ] `util/flock.ts` — `packages/core/src/util/effect-flock.ts` is the Effect-native implementation; Effect/service callers should use `EffectFlock.Service`, while legacy Promise callers still use the `packages/opencode/src/util/flock.ts` facade - Converted in this slice: `Config.updateGlobal` now uses `EffectFlock.Service.withLock` - Retained Promise boundary: provider models, plugin config patching/meta reads, automation run leases/scheduler ownership, and direct flock compatibility tests -- [ ] `util/process.ts` — child process spawn wrapper → return Effect instead of Promise +- [x] `util/process.ts` — `Process.Service` and Effect-native `run/text/lines/stop/descendants/terminateTree` now own execution and cleanup; the async facade delegates through `runPromise` + - Retained compatibility boundary: `Process.spawn` still returns the Node child facade because CLI pager/auth flows, long-lived LSP launch, Windows cmd script spawning, and stream ownership still depend on that shape + - Converted in this slice: `session/prompt.ts` inline shell expansion, `pty/index.ts` teardown cleanup, and `tool/shell.ts` abort/timeout cleanup use `Process.*Effect` directly - [ ] `util/lazy.ts` — sync-only route factories, shell selection, native module loading, and zod recursion stay on `lazy`; async Effect code should use `Effect.cached` - Converted in this slice: `tool/shell.ts` parser initialization now uses `Effect.cached` inside the tool's Effect definition - Retained async legacy boundary: provider models catalog cache and control-plane built-in adaptor import still expose Promise facades @@ -319,6 +321,8 @@ Every service currently exports async facade functions at the bottom of its name ### Process +Process execution is now effect-native at the utility boundary. Keep the Node child `spawn` facade until long-lived process owners no longer need direct `stdin`/`stdout` streams and `exited` promises, then delete it as a separate compatibility cleanup. + For each service, the migration is roughly: 1. **Find callers.** `grep -n "Namespace\.(methodA|methodB|...)"` across `src/` and `test/`. Skip the service file itself. diff --git a/packages/opencode/src/pty/index.ts b/packages/opencode/src/pty/index.ts index 82c13ca31..102494f37 100644 --- a/packages/opencode/src/pty/index.ts +++ b/packages/opencode/src/pty/index.ts @@ -169,13 +169,11 @@ export namespace Pty { return yield* Effect.promise(() => exited) } - yield* Effect.promise(() => - Process.terminateTree({ - pid: session.process.pid, - signalRoot: (signal) => session.process.kill(signal), - waitForExit: exited, - }), - ) + yield* Process.terminateTreeEffect({ + pid: session.process.pid, + signalRoot: (signal) => session.process.kill(signal), + waitForExit: exited, + }).pipe(Effect.orDie) if (hasExited(session)) return session.exitCode return yield* Effect.promise(() => exited) }) diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index ed1e3ea34..c0f7376b1 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -1495,10 +1495,14 @@ NOTE: At any point in time through this workflow you should feel free to ask the const shellMatches = ConfigMarkdown.shell(template) if (shellMatches.length > 0) { const sh = Shell.preferred() - const results = yield* Effect.promise(() => - Promise.all( - shellMatches.map(async ([, shellCmd]) => (await Process.text([shellCmd], { shell: sh, nothrow: true })).text), + const results = yield* Effect.all( + shellMatches.map(([, shellCmd]) => + Process.textEffect([shellCmd], { shell: sh, nothrow: true }).pipe( + Effect.map((result) => result.text), + Effect.orDie, + ), ), + { concurrency: "unbounded" }, ) let index = 0 template = template.replace(bashRegex, () => results[index++]) diff --git a/packages/opencode/src/tool/shell.ts b/packages/opencode/src/tool/shell.ts index 30be18ee7..a44c273af 100644 --- a/packages/opencode/src/tool/shell.ts +++ b/packages/opencode/src/tool/shell.ts @@ -607,15 +607,15 @@ export const ShellTool = Tool.define( if (exit.kind === "abort") { aborted = true - yield* Effect.promise(() => - Process.terminateTree({ pid: handle.pid, waitForExit: Effect.runPromise(handle.exitCode) }), - ).pipe(Effect.orDie) + yield* Process.terminateTreeEffect({ pid: handle.pid, waitForExit: Effect.runPromise(handle.exitCode) }).pipe( + Effect.orDie, + ) } if (exit.kind === "timeout") { expired = true - yield* Effect.promise(() => - Process.terminateTree({ pid: handle.pid, waitForExit: Effect.runPromise(handle.exitCode) }), - ).pipe(Effect.orDie) + yield* Process.terminateTreeEffect({ pid: handle.pid, waitForExit: Effect.runPromise(handle.exitCode) }).pipe( + Effect.orDie, + ) } // Drain any chunks the consumer hasn't processed yet, then push the diff --git a/packages/opencode/src/util/process.ts b/packages/opencode/src/util/process.ts index 8be1feb0f..11bd1220c 100644 --- a/packages/opencode/src/util/process.ts +++ b/packages/opencode/src/util/process.ts @@ -1,5 +1,6 @@ import { type ChildProcess } from "child_process" import launch from "cross-spawn" +import { Context, Effect, Layer, ManagedRuntime } from "effect" import { buffer } from "node:stream/consumers" import { setTimeout as sleep } from "node:timers/promises" import { errorMessage } from "./error" @@ -62,6 +63,25 @@ export namespace Process { export type Child = ChildProcess & { exited: Promise } + export interface Interface { + readonly run: (cmd: string[], opts?: RunOptions) => Effect.Effect + readonly text: (cmd: string[], opts?: RunOptions) => Effect.Effect + readonly lines: (cmd: string[], opts?: RunOptions) => Effect.Effect + readonly stop: (proc: ChildProcess) => Effect.Effect + readonly descendants: (pid: number) => Effect.Effect + readonly terminateTree: (input: TerminateTreeInput) => Effect.Effect + } + + export class Service extends Context.Service()("@opencode/Process") {} + + export interface TerminateTreeInput { + pid: number + graceMs?: number + signalRoot?: (signal: NodeJS.Signals) => void + waitForExit?: Promise + findDescendants?: (pid: number) => Promise + } + export function spawn(cmd: string[], opts: Options = {}): Child { if (cmd.length === 0) throw new Error("Command is required") opts.abort?.throwIfAborted() @@ -130,8 +150,12 @@ export namespace Process { return child } - export async function run(cmd: string[], opts: RunOptions = {}): Promise { - const proc = spawn(cmd, { + export const spawnEffect = Effect.fn("Process.spawn")(function* (cmd: string[], opts: Options = {}) { + return yield* Effect.sync(() => spawn(cmd, opts)) + }) + + export const runEffect = Effect.fn("Process.run")(function* (cmd: string[], opts: RunOptions = {}) { + const proc = yield* spawnEffect(cmd, { cwd: opts.cwd, env: opts.env, stdin: opts.stdin, @@ -143,29 +167,31 @@ export namespace Process { stderr: "pipe", }) - if (!proc.stdout || !proc.stderr) throw new Error("Process output not available") + if (!proc.stdout || !proc.stderr) return yield* Effect.fail(new Error("Process output not available")) - const out = await Promise.all([proc.exited, buffer(proc.stdout), buffer(proc.stderr)]) - .then(([code, stdout, stderr]) => ({ + const out = yield* Effect.tryPromise(() => + Promise.all([proc.exited, buffer(proc.stdout!), buffer(proc.stderr!)]).then(([code, stdout, stderr]) => ({ code, stdout, stderr, - })) - .catch((err: unknown) => { - if (!opts.nothrow) throw err - return { + })), + ).pipe( + Effect.catch((err: unknown) => { + if (!opts.nothrow) return Effect.fail(err) + return Effect.succeed({ code: 1, stdout: Buffer.alloc(0), stderr: Buffer.from(errorMessage(err)), - } - }) + }) + }), + ) if (out.code === 0 || opts.nothrow) return out - throw new RunFailedError(cmd, out.code, out.stdout, out.stderr) - } + return yield* Effect.fail(new RunFailedError(cmd, out.code, out.stdout, out.stderr)) + }) // The SDK keeps a sync stop variant because it cannot import opencode without // creating a cycle. Keep platform behavior aligned when changing this path. - export async function stop(proc: ChildProcess) { + export const stopEffect = Effect.fn("Process.stop")(function* (proc: ChildProcess) { if (proc.exitCode !== null || proc.signalCode !== null) return if (!proc.pid) { @@ -183,12 +209,12 @@ export namespace Process { proc.once("error", done) }) - await terminateTree({ + yield* terminateTreeEffect({ pid: proc.pid, signalRoot: (signal) => proc.kill(signal), waitForExit, }) - } + }) export function exists(pid: number) { try { @@ -199,18 +225,19 @@ export namespace Process { } } - export async function descendants(pid: number): Promise { + export const descendantsEffect = Effect.fn("Process.descendants")(function* (pid: number) { if (process.platform === "win32") return [] const seen = new Set() const pending = [pid] while (pending.length) { const parent = pending.pop()! - const out = await text(["pgrep", "-P", String(parent)], { nothrow: true }) - .then((result) => result.text) - .catch((error) => { + const out = yield* textEffect(["pgrep", "-P", String(parent)], { nothrow: true }).pipe( + Effect.map((result) => result.text), + Effect.catch((error) => { log.debug("failed to enumerate child processes", { pid: parent, error: errorMessage(error) }) - return "" - }) + return Effect.succeed("") + }), + ) for (const line of out.split(/\s+/)) { const child = Number(line) if (!Number.isInteger(child) || child <= 0 || seen.has(child)) continue @@ -219,7 +246,7 @@ export namespace Process { } } return Array.from(seen) - } + }) function signalPid(pid: number, signal: NodeJS.Signals) { try { @@ -239,25 +266,25 @@ export namespace Process { } } - export async function terminateTree(input: { - pid: number - graceMs?: number - signalRoot?: (signal: NodeJS.Signals) => void - waitForExit?: Promise - findDescendants?: (pid: number) => Promise - }) { + export const terminateTreeEffect = Effect.fn("Process.terminateTree")(function* (input: TerminateTreeInput) { const graceMs = input.graceMs ?? TERMINATION_GRACE_MS if (process.platform === "win32") { - await run(["taskkill", "/pid", String(input.pid), "/f", "/t"], { nothrow: true }) + yield* runEffect(["taskkill", "/pid", String(input.pid), "/f", "/t"], { nothrow: true }) return } // Descendants are a best-effort snapshot for normal child processes. A // daemonized double-fork can intentionally leave this tree before cleanup. - const children = await (input.findDescendants ?? descendants)(input.pid).catch((error) => { - log.debug("failed to enumerate process tree", { pid: input.pid, error: errorMessage(error) }) - return [] - }) + const children = yield* ( + input.findDescendants + ? Effect.tryPromise(() => input.findDescendants!(input.pid)) + : descendantsEffect(input.pid) + ).pipe( + Effect.catch((error) => { + log.debug("failed to enumerate process tree", { pid: input.pid, error: errorMessage(error) }) + return Effect.succeed([]) + }), + ) const signalRoot = (signal: NodeJS.Signals) => { if (input.signalRoot && exists(input.pid)) { try { @@ -277,9 +304,11 @@ export namespace Process { // With waitForExit, worst case is one grace period before SIGKILL and one // bounded wait after SIGKILL so callers can observe the final exit. - const rootExited = await (input.waitForExit - ? Promise.race([input.waitForExit.then(() => true, () => true), sleep(graceMs).then(() => false)]) - : sleep(graceMs).then(() => false)) + const rootExited = yield* Effect.promise(() => + input.waitForExit + ? Promise.race([input.waitForExit.then(() => true, () => true), sleep(graceMs).then(() => false)]) + : sleep(graceMs).then(() => false), + ) if (!exists(input.pid) && children.every((child) => !exists(child))) return if (groupSignaled && !rootExited && exists(input.pid)) { @@ -290,18 +319,58 @@ export namespace Process { } for (const child of children) signalPid(child, "SIGKILL") log.debug("sent process tree kill signals", { pid: input.pid, groupSignaled, descendantCount: children.length }) - if (input.waitForExit) await Promise.race([input.waitForExit.catch(() => undefined), sleep(graceMs)]) - } + if (input.waitForExit) yield* Effect.promise(() => Promise.race([input.waitForExit!.catch(() => undefined), sleep(graceMs)])) + }) - export async function text(cmd: string[], opts: RunOptions = {}): Promise { - const out = await run(cmd, opts) + export const textEffect = Effect.fn("Process.text")(function* (cmd: string[], opts: RunOptions = {}) { + const out = yield* runEffect(cmd, opts) return { ...out, text: out.stdout.toString(), } + }) + + export const linesEffect = Effect.fn("Process.lines")(function* (cmd: string[], opts: RunOptions = {}) { + return (yield* textEffect(cmd, opts)).text.split(/\r?\n/).filter(Boolean) + }) + + export const layer = Layer.succeed( + Service, + Service.of({ + run: runEffect, + text: textEffect, + lines: linesEffect, + stop: stopEffect, + descendants: descendantsEffect, + terminateTree: terminateTreeEffect, + }), + ) + export const defaultLayer = layer + + const runtime = ManagedRuntime.make(defaultLayer) + const runPromise = (fn: (process: Interface) => Effect.Effect) => runtime.runPromise(Service.use(fn)) + + export async function run(cmd: string[], opts: RunOptions = {}): Promise { + return runPromise((process) => process.run(cmd, opts)) + } + + export async function stop(proc: ChildProcess) { + return runPromise((process) => process.stop(proc)) + } + + export async function descendants(pid: number): Promise { + return runPromise((process) => process.descendants(pid)) + } + + export async function terminateTree(input: TerminateTreeInput) { + return runPromise((process) => process.terminateTree(input)) + } + + export async function text(cmd: string[], opts: RunOptions = {}): Promise { + return runPromise((process) => process.text(cmd, opts)) } export async function lines(cmd: string[], opts: RunOptions = {}): Promise { - return (await text(cmd, opts)).text.split(/\r?\n/).filter(Boolean) + return runPromise((process) => process.lines(cmd, opts)) } } diff --git a/packages/opencode/test/util/process.test.ts b/packages/opencode/test/util/process.test.ts index 29d26649f..603978cc6 100644 --- a/packages/opencode/test/util/process.test.ts +++ b/packages/opencode/test/util/process.test.ts @@ -1,4 +1,5 @@ import { describe, expect, test } from "bun:test" +import { Effect } from "effect" import fs from "fs/promises" import path from "path" import { Process } from "../../src/util/process" @@ -18,6 +19,13 @@ async function waitForFile(file: string) { } describe("util.process", () => { + test("captures stdout and stderr through the Effect path", async () => { + const out = await Effect.runPromise(Process.runEffect(node('process.stdout.write("out");process.stderr.write("err")'))) + expect(out.code).toBe(0) + expect(out.stdout.toString()).toBe("out") + expect(out.stderr.toString()).toBe("err") + }) + test("captures stdout and stderr", async () => { const out = await Process.run(node('process.stdout.write("out");process.stderr.write("err")')) expect(out.code).toBe(0)