diff --git a/packages/core/src/cross-spawn-spawner.ts b/packages/core/src/cross-spawn-spawner.ts index ad8d4126d45..ca3c4d9d19e 100644 --- a/packages/core/src/cross-spawn-spawner.ts +++ b/packages/core/src/cross-spawn-spawner.ts @@ -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((resume) => { const signal = Deferred.makeUnsafe() 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) diff --git a/packages/core/src/process-output.ts b/packages/core/src/process-output.ts new file mode 100644 index 00000000000..cdfff7175df --- /dev/null +++ b/packages/core/src/process-output.ts @@ -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() + +export const register = (handle: ChildProcessHandle, process: ChildProcess) => { + processes.set(handle, process) +} + +export const drain = ( + handle: ChildProcessHandle, + drains: ReadonlyArray>, + 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>() + 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" diff --git a/packages/core/src/process.ts b/packages/core/src/process.ts index 4555b28017a..a6b83219f5d 100644 --- a/packages/core/src/process.ts +++ b/packages/core/src/process.ts @@ -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", { command: Schema.String, @@ -125,6 +126,19 @@ export const collectStream = (stream: Stream.Stream, }, ).pipe(Effect.map((x) => ({ buffer: Buffer.concat(x.chunks), truncated: x.truncated }))) +const collector = (stream: Stream.Stream, 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 }), ) diff --git a/packages/core/test/effect/cross-spawn-spawner.test.ts b/packages/core/test/effect/cross-spawn-spawner.test.ts index 8a2fab49306..e9d88365c5a 100644 --- a/packages/core/test/effect/cross-spawn-spawner.test.ts +++ b/packages/core/test/effect/cross-spawn-spawner.test.ts @@ -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", () => { diff --git a/packages/core/test/process/process.test.ts b/packages/core/test/process/process.test.ts index f8377f718aa..ffaa9e57296 100644 --- a/packages/core/test/process/process.test.ts +++ b/packages/core/test/process/process.test.ts @@ -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((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( diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index d036c6542d3..09fc66699e9 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -23,6 +23,7 @@ import { LSP } from "@/lsp/lsp" import { ulid } from "ulid" 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 * as Stream from "effect/Stream" import { Command } from "../command" import { pathToFileURL, fileURLToPath } from "url" @@ -563,38 +564,48 @@ export const layer = Layer.effect( const args = Shell.args(sh, input.command, cwd) let output = "" let aborted = false + let acceptingOutput = true - const finish = Effect.uninterruptible( - Effect.gen(function* () { - if (aborted) { - output += "\n\n" + ["", "User aborted the command", ""].join("\n") - } - const completed = Date.now() - if (flags.experimentalEventSystem) { - yield* events.publish(SessionEvent.Shell.Ended, { - sessionID: input.sessionID, - timestamp: DateTime.makeUnsafe(completed), - callID: part.callID, - output, - }) - } - if (!msg.time.completed) { - msg.time.completed = completed - yield* sessions.updateMessage(msg) - } - if (part.state.status === "running") { - part.state = { - status: "completed", - time: { ...part.state.time, end: completed }, - input: part.state.input, - title: "", - metadata: { output, description: "" }, - output, + const finish = (error?: string) => + Effect.uninterruptible( + Effect.gen(function* () { + if (aborted) { + output += "\n\n" + ["", "User aborted the command", ""].join("\n") } - yield* sessions.updatePart(part) - } - }), - ) + const completed = Date.now() + if (flags.experimentalEventSystem) { + yield* events.publish(SessionEvent.Shell.Ended, { + sessionID: input.sessionID, + timestamp: DateTime.makeUnsafe(completed), + callID: part.callID, + output, + }) + } + if (!msg.time.completed) { + msg.time.completed = completed + yield* sessions.updateMessage(msg) + } + if (part.state.status === "running") { + part.state = error + ? { + status: "error", + error, + time: { ...part.state.time, end: completed }, + input: part.state.input, + metadata: { output, description: "" }, + } + : { + status: "completed", + time: { ...part.state.time, end: completed }, + input: part.state.input, + title: "", + metadata: { output, description: "" }, + output, + } + yield* sessions.updatePart(part) + } + }), + ) const exit = yield* restore( Effect.gen(function* () { @@ -611,27 +622,39 @@ export const layer = Layer.effect( forceKillAfter: "3 seconds", }) const handle = yield* spawner.spawn(cmd) - yield* Stream.runForEach(Stream.decodeText(handle.all), (chunk) => - Effect.gen(function* () { - output += chunk - if (part.state.status === "running") { - part.state.metadata = { output, description: "" } - yield* sessions.updatePart(part) - } - }), - ) - yield* handle.exitCode + yield* ProcessOutput.drain( + handle, + [ + Stream.runForEach(Stream.decodeText(handle.all), (chunk) => + Effect.gen(function* () { + if (!acceptingOutput) return + output += chunk + if (part.state.status === "running") { + part.state.metadata = { output, description: "" } + yield* sessions.updatePart(part) + } + }), + ), + ], + { + grace: "250 millis", + onClose: () => { + acceptingOutput = false + }, + }, + ).pipe(Effect.asVoid) }).pipe(Effect.scoped, Effect.orDie), ).pipe(Effect.exit) if (Exit.isFailure(exit) && Cause.hasInterrupts(exit.cause) && !Cause.hasDies(exit.cause)) { aborted = true } - yield* finish + const failure = + Exit.isFailure(exit) && !aborted && !Cause.hasInterruptsOnly(exit.cause) ? exit.cause : undefined + const error = failure && Cause.squash(failure) + yield* finish(error instanceof Error ? error.message : error ? String(error) : undefined) - if (Exit.isFailure(exit) && !aborted && !Cause.hasInterruptsOnly(exit.cause)) { - return yield* Effect.failCause(exit.cause) - } + if (failure) return yield* Effect.failCause(failure) return { info: msg, parts: [part] } }), diff --git a/packages/opencode/src/tool/shell.ts b/packages/opencode/src/tool/shell.ts index 7726c1bd496..440073bb6cf 100644 --- a/packages/opencode/src/tool/shell.ts +++ b/packages/opencode/src/tool/shell.ts @@ -1,4 +1,4 @@ -import { Effect, Stream } from "effect" +import { Effect, Fiber, Stream } from "effect" import os from "os" import { createWriteStream } from "node:fs" import * as Tool from "./tool" @@ -20,6 +20,7 @@ import * as Truncate from "./truncate" import { Plugin } from "@/plugin" import { ChildProcess } from "effect/unstable/process" import { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner" +import { ProcessOutput } from "@opencode-ai/core/process-output" import { ShellPrompt, type Parameters } from "./shell/prompt" import { BashArity } from "@/permission/arity" @@ -456,6 +457,7 @@ export const ShellTool = Tool.define( let cut = false let expired = false let aborted = false + let acceptingOutput = true const closeSink = Effect.fnUntraced(function* () { const stream = sink @@ -494,54 +496,59 @@ export const ShellTool = Tool.define( yield* Effect.addFinalizer(closeSink) const handle = yield* spawner.spawn(cmd(input.shell, input.command, input.cwd, input.env)) - yield* Effect.forkScoped( - Stream.runForEach(Stream.decodeText(handle.all), (chunk) => { - const size = Buffer.byteLength(chunk, "utf-8") - list.push({ text: chunk, size }) - used += size - while (used > keep && list.length > 1) { - const item = list.shift() - if (!item) break - used -= item.size - cut = true + const output = Stream.runForEach(Stream.decodeText(handle.all), (chunk) => { + if (!acceptingOutput) return Effect.void + const size = Buffer.byteLength(chunk, "utf-8") + list.push({ text: chunk, size }) + used += size + while (used > keep && list.length > 1) { + const item = list.shift() + if (!item) break + used -= item.size + cut = true + } + + last = preview(last + chunk) + + if (file) { + sink?.write(chunk) + } else { + full += chunk + if (Buffer.byteLength(full, "utf-8") > limits.maxBytes) { + return trunc.write(full).pipe( + Effect.andThen((next) => + Effect.sync(() => { + file = next + cut = true + sink = createWriteStream(next, { flags: "a" }) + full = "" + }), + ), + Effect.andThen( + ctx.metadata({ + metadata: { + output: last, + description: input.description, + }, + }), + ), + ) } + } - last = preview(last + chunk) - - if (file) { - sink?.write(chunk) - } else { - full += chunk - if (Buffer.byteLength(full, "utf-8") > limits.maxBytes) { - return trunc.write(full).pipe( - Effect.andThen((next) => - Effect.sync(() => { - file = next - cut = true - sink = createWriteStream(next, { flags: "a" }) - full = "" - }), - ), - Effect.andThen( - ctx.metadata({ - metadata: { - output: last, - description: input.description, - }, - }), - ), - ) - } - } - - return ctx.metadata({ - metadata: { - output: last, - description: input.description, - }, - }) - }), - ) + return ctx.metadata({ + metadata: { + output: last, + description: input.description, + }, + }) + }) + const drained = yield* ProcessOutput.drain(handle, [output], { + grace: "1500 millis", + onClose: () => { + acceptingOutput = false + }, + }).pipe(Effect.forkScoped) const abort = Effect.callback((resume) => { if (ctx.abort.aborted) return resume(Effect.void) @@ -552,8 +559,8 @@ export const ShellTool = Tool.define( const timeout = Effect.sleep(`${input.timeout + 100} millis`) - const exit = yield* Effect.raceAll([ - handle.exitCode.pipe(Effect.map((code) => ({ kind: "exit" as const, code }))), + const exit = yield* Effect.raceAllFirst([ + Fiber.join(drained).pipe(Effect.map((code) => ({ kind: "exit" as const, code }))), abort.pipe(Effect.map(() => ({ kind: "abort" as const, code: null }))), timeout.pipe(Effect.map(() => ({ kind: "timeout" as const, code: null }))), ]) @@ -566,6 +573,7 @@ export const ShellTool = Tool.define( expired = true yield* handle.kill({ forceKillAfter: "3 seconds" }).pipe(Effect.orDie) } + if (exit.kind !== "exit") yield* Fiber.await(drained) return exit.kind === "exit" ? exit.code : null }), diff --git a/packages/opencode/test/session/prompt.test.ts b/packages/opencode/test/session/prompt.test.ts index cb9694b56db..4a95316527d 100644 --- a/packages/opencode/test/session/prompt.test.ts +++ b/packages/opencode/test/session/prompt.test.ts @@ -6,7 +6,15 @@ import { eq } from "drizzle-orm" import { EventV2Bridge } from "@/event-v2-bridge" import { FetchHttpClient } from "effect/unstable/http" import { expect } from "bun:test" -import { Cause, Deferred, Duration, Effect, Exit, Fiber, Layer } from "effect" +import { Cause, Deferred, Duration, Effect, Exit, Fiber, Layer, Sink, Stream } from "effect" +import * as PlatformError from "effect/PlatformError" +import { + ChildProcessSpawner, + ExitCode, + make as makeSpawner, + makeHandle, + ProcessId, +} from "effect/unstable/process/ChildProcessSpawner" import path from "path" import { fileURLToPath, pathToFileURL } from "url" import { NamedError } from "@opencode-ai/core/util/error" @@ -157,8 +165,6 @@ const lsp = Layer.succeed( const status = SessionStatus.layer.pipe(Layer.provideMerge(EventV2Bridge.defaultLayer)) const run = SessionRunState.layer.pipe(Layer.provide(status)) -const infra = Layer.mergeAll(NodeFileSystem.layer, CrossSpawnSpawner.defaultLayer) - const processorCreateStarted: Array<() => void> = [] const blockingProcessor = Layer.succeed( SessionProcessor.Service, @@ -167,7 +173,8 @@ const blockingProcessor = Layer.succeed( }), ) -function makePrompt(input?: { processor?: "blocking" }) { +function makePrompt(input?: { processor?: "blocking"; spawner?: Layer.Layer }) { + const infra = Layer.mergeAll(NodeFileSystem.layer, input?.spawner ?? CrossSpawnSpawner.defaultLayer) const deps = Layer.mergeAll( Session.defaultLayer, Snapshot.defaultLayer, @@ -236,17 +243,39 @@ function makePrompt(input?: { processor?: "blocking" }) { ) } -function makeHttp(input?: { processor?: "blocking" }) { +function makeHttp(input?: { processor?: "blocking"; spawner?: Layer.Layer }) { return Layer.mergeAll(TestLLMServer.layer, makePrompt(input)) } -function makeHttpNoLLMServer(input?: { processor?: "blocking" }) { +function makeHttpNoLLMServer(input?: { processor?: "blocking"; spawner?: Layer.Layer }) { return makePrompt(input) } const it = testEffect(makeHttp()) const noLLMServer = testEffect(makeHttpNoLLMServer()) const raceNoLLMServer = testEffect(makeHttpNoLLMServer({ processor: "blocking" })) +const streamError = PlatformError.systemError({ _tag: "Unknown", module: "ChildProcess", method: "stdout" }) +const failingHandle = makeHandle({ + pid: ProcessId(1), + stdin: Sink.drain, + stdout: Stream.fail(streamError), + stderr: Stream.empty, + all: Stream.fail(streamError), + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + isRunning: Effect.succeed(false), + exitCode: Effect.succeed(ExitCode(0)), + kill: () => Effect.void, + unref: Effect.succeed(Effect.void), +}) +const failingShell = testEffect( + makeHttpNoLLMServer({ + spawner: Layer.succeed( + ChildProcessSpawner, + makeSpawner(() => Effect.succeed(failingHandle)), + ), + }), +) const unix = process.platform !== "win32" ? it.instance : it.instance.skip const unixNoLLMServer = process.platform !== "win32" ? noLLMServer.instance : noLLMServer.instance.skip @@ -1643,6 +1672,21 @@ unixNoLLMServer( 30_000, ) +failingShell.instance( + "legacy shell persists error when output draining fails", + () => + Effect.gen(function* () { + const { prompt, sessions, chat } = yield* boot() + const exit = yield* prompt.shell({ sessionID: chat.id, agent: "build", command: "ignored" }).pipe(Effect.exit) + const messages = yield* sessions.messages({ sessionID: chat.id }) + const tool = messages.flatMap((message) => message.parts).find((part) => part.type === "tool") + + expect(Exit.isFailure(exit)).toBe(true) + expect(tool?.state.status).toBe("error") + }), + { config: cfg }, +) + unixNoLLMServer( "cancel persists aborted shell result when shell ignores TERM", () => @@ -1744,6 +1788,35 @@ unix( 30_000, ) +unix( + "bash tool settles when a descendant inherits stdout", + () => + Effect.gen(function* () { + const { dir, llm } = yield* useServerConfig(providerCfg) + const prompt = yield* SessionPrompt.Service + const sessions = yield* Session.Service + const chat = yield* sessions.create({ permission: [{ permission: "*", pattern: "*", action: "allow" }] }) + yield* prompt.prompt({ + sessionID: chat.id, + agent: "build", + noReply: true, + parts: [{ type: "text", text: "run" }], + }) + yield* llm.tool("bash", { + command: "sleep 30 & printf done", + description: "background", + timeout: 30_000, + workdir: dir, + }) + yield* llm.text("continued") + + const result = yield* prompt.loop({ sessionID: chat.id }).pipe(Effect.timeout("3 seconds")) + expect(result.parts.some((part) => part.type === "text" && part.text === "continued")).toBe(true) + }), + { git: true }, + 8_000, +) + unixNoLLMServer( "cancel interrupts loop queued behind shell", () =>