From c31464da702b15977f5bef0023c2b1a7db7a680b Mon Sep 17 00:00:00 2001 From: flowluap Date: Sun, 19 Jul 2026 18:37:58 +0200 Subject: [PATCH] fix(opencode): batch direct shell output updates --- packages/opencode/src/session/prompt.ts | 23 +++- packages/opencode/test/session/prompt.test.ts | 127 ++++++++++++++++++ 2 files changed, 144 insertions(+), 6 deletions(-) diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index eb116f6b960f..259c3f057bc1 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -42,7 +42,7 @@ import { Truncate } from "@/tool/truncate" import { Image } from "@/image/image" import { decodeDataUrl } from "@/util/data-url" import { Process } from "@/util/process" -import { Cause, Effect, Exit, Latch, Layer, Option, Scope, Context, Schema, Types } from "effect" +import { Cause, Effect, Exit, Fiber, Latch, Layer, Option, Scope, Context, Schema, Types } from "effect" import { InstanceState } from "@/effect/instance-state" import { TaskTool, type TaskPromptOps } from "@/tool/task" import { SessionRunState } from "./run-state" @@ -523,6 +523,8 @@ const layer = Layer.effect( const sh = Shell.preferred(cfg.shell) const args = Shell.args(sh, input.command, cwd) let output = "" + let outputVersion = 0 + let publishedVersion = 0 let aborted = false const finish = Effect.uninterruptible( @@ -564,16 +566,25 @@ const layer = Layer.effect( forceKillAfter: "3 seconds", }) const handle = yield* spawner.spawn(cmd) + const updates = yield* Effect.gen(function* () { + while (true) { + yield* Effect.sleep("100 millis") + const version = outputVersion + if (version === publishedVersion || part.state.status !== "running") continue + part.state.metadata = { output } + yield* sessions.updatePart(part) + // Output received during the durable update keeps a newer version dirty. + publishedVersion = version + } + }).pipe(Effect.forkScoped) yield* Stream.runForEach(Stream.decodeText(handle.all), (chunk) => - Effect.gen(function* () { + Effect.sync(() => { output += chunk - if (part.state.status === "running") { - part.state.metadata = { output } - yield* sessions.updatePart(part) - } + outputVersion++ }), ) yield* handle.exitCode + yield* Fiber.interrupt(updates) }).pipe(Effect.scoped, Effect.orDie), ).pipe(Effect.exit) diff --git a/packages/opencode/test/session/prompt.test.ts b/packages/opencode/test/session/prompt.test.ts index 491ad06aaf47..574b31866145 100644 --- a/packages/opencode/test/session/prompt.test.ts +++ b/packages/opencode/test/session/prompt.test.ts @@ -1667,6 +1667,133 @@ unixNoLLMServer( 30_000, ) +unixNoLLMServer( + "shell batches frequent running output updates while preserving cumulative output", + () => + withSh(() => + Effect.gen(function* () { + const events = yield* EventV2Bridge.Service + const { prompt, chat } = yield* boot() + const command = "i=0; while [ $i -lt 20 ]; do printf '%02d\\n' $i; i=$((i + 1)); sleep 0.02; done" + const updates: Array<{ part: SessionV1.ToolPart; bytes: number }> = [] + const off = yield* events.listen((event) => { + if (event.type !== MessageV2.Event.PartUpdated.type) return Effect.void + const data = event.data as typeof MessageV2.Event.PartUpdated.data.Type + const part = data.part as SessionV1.Part + if (part.type === "tool" && part.state.input.command === command) { + updates.push({ part: structuredClone(part), bytes: Buffer.byteLength(JSON.stringify(event.data)) }) + } + return Effect.void + }) + + const result = yield* prompt.shell({ sessionID: chat.id, agent: "build", command }) + yield* off + const running = updates.flatMap((update) => + update.part.state.status === "running" && update.part.state.metadata?.output + ? [{ output: update.part.state.metadata.output }] + : [], + ) + const tool = completedTool(result.parts) + if (!tool) return + + expect(running.length).toBeGreaterThanOrEqual(2) + expect(running.length).toBeLessThan(10) + expect(running[0]?.output).toStartWith("00\n") + expect(tool.state.output).toBe( + Array.from({ length: 20 }, (_, index) => `${index.toString().padStart(2, "0")}\n`).join(""), + ) + expect(tool.state.metadata.output).toBe(tool.state.output) + + if (Bun.env.OPENCODE_PROMPT_SHELL_BENCH === "1") { + console.log( + `PROMPT_SHELL_BENCH ${JSON.stringify({ + writes: 20, + legacyRunningUpdateUpperBound: 20, + runningUpdates: running.length, + totalUpdates: updates.length, + durablePayloadBytes: updates.reduce((sum, update) => sum + update.bytes, 0), + outputBytes: Buffer.byteLength(tool.state.output), + outputCorrect: tool.state.output.endsWith("19\n"), + })}`, + ) + } + }), + ), + { config: cfg }, + 30_000, +) + +unixNoLLMServer( + "shell completion contains the final byte and is never followed by a running update", + () => + withSh(() => + Effect.gen(function* () { + const events = yield* EventV2Bridge.Service + const { prompt, chat } = yield* boot() + const command = "printf first; sleep 0.15; printf Z" + const states: Array<{ status: SessionV1.ToolPart["state"]["status"]; output?: string }> = [] + const off = yield* events.listen((event) => { + if (event.type !== MessageV2.Event.PartUpdated.type) return Effect.void + const part = (event.data as typeof MessageV2.Event.PartUpdated.data.Type).part as SessionV1.Part + if (part.type === "tool" && part.state.input.command === command) { + states.push({ + status: part.state.status, + output: + part.state.status === "running" + ? part.state.metadata?.output + : part.state.status === "completed" + ? part.state.output + : undefined, + }) + } + return Effect.void + }) + + const result = yield* prompt.shell({ sessionID: chat.id, agent: "build", command }) + const countAtCompletion = states.length + yield* Effect.sleep("200 millis") + yield* off + const tool = completedTool(result.parts) + if (!tool) return + + expect(tool.state.output).toBe("firstZ") + expect(states.length).toBe(countAtCompletion) + expect(states.at(-1)).toEqual({ status: "completed", output: "firstZ" }) + const completed = states.findIndex((state) => state.status === "completed") + expect(states.slice(completed + 1).some((state) => state.status === "running")).toBe(false) + }), + ), + { config: cfg }, + 30_000, +) + +unixNoLLMServer( + "shell does not publish idle running updates for a quiet command", + () => + withSh(() => + Effect.gen(function* () { + const events = yield* EventV2Bridge.Service + const { prompt, chat } = yield* boot() + const command = "sleep 0.25" + const states: SessionV1.ToolPart["state"][] = [] + const off = yield* events.listen((event) => { + if (event.type !== MessageV2.Event.PartUpdated.type) return Effect.void + const part = (event.data as typeof MessageV2.Event.PartUpdated.data.Type).part as SessionV1.Part + if (part.type === "tool" && part.state.input.command === command) states.push(structuredClone(part.state)) + return Effect.void + }) + + yield* prompt.shell({ sessionID: chat.id, agent: "build", command }) + yield* off + + expect(states.map((state) => state.status)).toEqual(["running", "completed"]) + expect(states.some((state) => state.status === "running" && state.metadata !== undefined)).toBe(false) + }), + ), + { config: cfg }, + 30_000, +) + it.instance( "loop waits while shell runs and starts after shell exits", () =>