diff --git a/packages/opencode/src/tool/shell-metadata-throttle.ts b/packages/opencode/src/tool/shell-metadata-throttle.ts new file mode 100644 index 000000000..23781b6e2 --- /dev/null +++ b/packages/opencode/src/tool/shell-metadata-throttle.ts @@ -0,0 +1,76 @@ +import { Duration, Effect, Schedule, Semaphore, type Scope } from "effect" + +export interface MetadataThrottleOptions { + intervalMillis: number + byteThreshold: number + snapshot: () => string + emit: (output: string) => Effect.Effect +} + +export interface MetadataThrottle { + onChunk: (size: number) => Effect.Effect + flush: (reason: "spill" | "final") => Effect.Effect +} + +// Coalesces the shell tool's per-chunk metadata pushes. Without this, every +// decoded chunk fires a full part update + message.part.updated event, which is +// the dominant per-call cost for chatty commands. Emissions run through a single +// serialized channel so the timer fiber and the stream consumer never interleave +// writes — each emit ships the full preview, so an out-of-order write could let a +// stale preview overwrite a newer one downstream. +export const makeMetadataThrottle = ( + options: MetadataThrottleOptions, +): Effect.Effect => + Effect.gen(function* () { + const lock = yield* Semaphore.make(1) + let dirty = false + let bytesSinceFlush = 0 + let firstFlushed = false + + const emit = (force: boolean) => + lock.withPermits(1)( + Effect.suspend(() => { + if (!dirty && !force) return Effect.void + const output = options.snapshot() + // Clear before awaiting emit so chunks arriving mid-emit stay marked + // dirty for the next flush instead of being silently dropped. + dirty = false + bytesSinceFlush = 0 + // Best-effort: a metadata push is a UI/notification side effect. Swallow + // a defect so it cannot fail the stream consumer (which would stop + // reading stdout and could hang the process) or orDie the final flush. + // Interrupts and typed errors still propagate. + return options.emit(output).pipe(Effect.catchDefect(() => Effect.void)) + }), + ) + + const onChunk = (size: number) => + Effect.suspend(() => { + // A leading empty decode chunk (TextDecoder can emit "" on a partial + // multibyte boundary) carries no new preview; skipping it avoids burning + // the first-flush slot before the first real output arrives. + if (size <= 0) return Effect.void + dirty = true + bytesSinceFlush += size + // First non-empty chunk is visible immediately: downstream consumers + // (e.g. the abort-on-output test path) rely on seeing it synchronously. + if (!firstFlushed) { + firstFlushed = true + return emit(true) + } + if (bytesSinceFlush >= options.byteThreshold) return emit(true) + return Effect.void + }) + + // "spill" forces an emit even under the byte threshold (output just crossed + // to a tempfile); "final" is dirty-gated so it only pushes a pending tail. + const flush = (reason: "spill" | "final") => emit(reason === "spill") + + yield* emit(false).pipe( + Effect.repeat(Schedule.spaced(Duration.millis(options.intervalMillis))), + Effect.delay(Duration.millis(options.intervalMillis)), + Effect.forkScoped, + ) + + return { onChunk, flush } + }) diff --git a/packages/opencode/src/tool/shell.ts b/packages/opencode/src/tool/shell.ts index 2a63f805f..dcd4497f1 100644 --- a/packages/opencode/src/tool/shell.ts +++ b/packages/opencode/src/tool/shell.ts @@ -17,7 +17,7 @@ import { Process } from "@/util/process" import { BashArity } from "@/permission/arity" import * as Truncate from "./truncate" import { Plugin } from "@/plugin" -import { Effect, Stream } from "effect" +import { Duration, Effect, Fiber, Stream } from "effect" import { ChildProcess } from "effect/unstable/process" import { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner" import { envValueCaseInsensitive, prependBundledTools, stripPathKeys, withoutInternalServerAuthEnv } from "@/util/env" @@ -31,8 +31,20 @@ import { discoverOfficeOutputs, readTrackedState } from "./shell-output-capture" import { Parameters, render as renderDescription, type Limits } from "./shell/prompt" import { ToolID as ShellToolID } from "./shell/id" import { orchestrateArtifacts, type ArtifactDeps } from "./shell-artifact-orchestrator" +import { makeMetadataThrottle } from "./shell-metadata-throttle" const MAX_METADATA_LENGTH = 30_000 +// Coalesce streaming metadata pushes: emit the first chunk immediately, then at +// most once per interval or once accumulated input crosses the byte threshold. +const METADATA_FLUSH_INTERVAL_MS = 150 +const METADATA_FLUSH_BYTES = 4 * 1024 +// Cap how long we wait for the consumer to drain buffered output after the +// process exits/aborts/times out. On timeout we fall through to scope cleanup, +// which interrupts the consumer; the final tool-result metadata still carries +// the tail via completeToolCall, so this only bounds a pathological slow/never +// closing stream — it never blocks the normal path where the stream is already +// drained by the time the exit race resolves. +const CONSUMER_DRAIN_TIMEOUT = Duration.seconds(1) const DEFAULT_TIMEOUT = Flag.OPENCODE_EXPERIMENTAL_BASH_DEFAULT_TIMEOUT_MS || 2 * 60 * 1000 const PS = new Set(["powershell", "pwsh"]) const CWD = new Set(["cd", "push-location", "set-location"]) @@ -481,7 +493,14 @@ export const ShellTool = Tool.define( Effect.gen(function* () { const handle = yield* spawner.spawn(cmd(input.shell, input.name, input.command, input.cwd, input.env)) - yield* Effect.forkScoped( + const throttle = yield* makeMetadataThrottle({ + intervalMillis: METADATA_FLUSH_INTERVAL_MS, + byteThreshold: METADATA_FLUSH_BYTES, + snapshot: () => last, + emit: (output) => ctx.metadata({ metadata: { output, description: input.description } }), + }) + + const consumer = yield* Effect.forkScoped( Stream.runForEach(Stream.decodeText(handle.all), (chunk) => { const size = Buffer.byteLength(chunk, "utf-8") list.push({ text: chunk, size }) @@ -497,36 +516,25 @@ export const ShellTool = Tool.define( 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 throttle.onChunk(size) } - return ctx.metadata({ - metadata: { - output: last, - description: input.description, - }, - }) + 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(throttle.flush("spill")), + ) + } + + return throttle.onChunk(size) }), ) @@ -558,6 +566,15 @@ export const ShellTool = Tool.define( ).pipe(Effect.orDie) } + // Drain any chunks the consumer hasn't processed yet, then push the + // final preview. Both happen inside the process scope so the throttle's + // timer fiber is still alive and is interrupted on scope exit. The + // timeout guards against a stream that never closes; ignore mirrors the + // pre-existing fork-without-join behavior where consumer errors were + // unobserved. + yield* Fiber.join(consumer).pipe(Effect.timeout(CONSUMER_DRAIN_TIMEOUT), Effect.ignore) + yield* throttle.flush("final") + return exit.kind === "exit" ? exit.code : null }), ).pipe(Effect.orDie) diff --git a/packages/opencode/test/tool/shell-metadata-throttle.test.ts b/packages/opencode/test/tool/shell-metadata-throttle.test.ts new file mode 100644 index 000000000..7587b9ce2 --- /dev/null +++ b/packages/opencode/test/tool/shell-metadata-throttle.test.ts @@ -0,0 +1,152 @@ +import { describe, expect } from "bun:test" +import { Duration, Effect } from "effect" +import * as TestClock from "effect/testing/TestClock" +import { it } from "../lib/effect" +import { makeMetadataThrottle } from "../../src/tool/shell-metadata-throttle" + +function setup(overrides?: { intervalMillis?: number; byteThreshold?: number }) { + const emits: string[] = [] + const state = { last: "" } + const make = makeMetadataThrottle({ + intervalMillis: overrides?.intervalMillis ?? 150, + byteThreshold: overrides?.byteThreshold ?? 4 * 1024, + snapshot: () => state.last, + emit: (output) => Effect.sync(() => emits.push(output)), + }) + return { emits, state, make } +} + +describe("tool.shell metadata throttle", () => { + it.effect("emits the first chunk synchronously without advancing the clock", () => + Effect.gen(function* () { + const { emits, state, make } = setup() + const throttle = yield* make + state.last = "hello" + yield* throttle.onChunk(5) + expect(emits).toEqual(["hello"]) + }), + ) + + it.effect("emits immediately when accumulated bytes cross the threshold", () => + Effect.gen(function* () { + const { emits, state, make } = setup({ byteThreshold: 100 }) + const throttle = yield* make + state.last = "a" + yield* throttle.onChunk(1) // first chunk flushes immediately + state.last = "ab" + yield* throttle.onChunk(50) // 50 < 100, coalesced + expect(emits).toEqual(["a"]) + state.last = "abc" + yield* throttle.onChunk(60) // 50 + 60 >= 100, flushes + expect(emits).toEqual(["a", "abc"]) + }), + ) + + it.effect("flushes coalesced chunks on the interval timer (progressive updates)", () => + Effect.gen(function* () { + const { emits, state, make } = setup() + const throttle = yield* make + state.last = "1" + yield* throttle.onChunk(1) // first flush + state.last = "12" + yield* throttle.onChunk(1) // coalesced under threshold + yield* TestClock.adjust(Duration.millis(150)) // timer flush + state.last = "123" + yield* throttle.onChunk(1) + yield* TestClock.adjust(Duration.millis(150)) // timer flush + state.last = "1234" + yield* throttle.onChunk(1) + yield* TestClock.adjust(Duration.millis(150)) // timer flush + yield* TestClock.adjust(Duration.millis(150)) // nothing dirty, no flush + expect(emits).toEqual(["1", "12", "123", "1234"]) + expect(emits.length).toBeGreaterThanOrEqual(3) + }), + ) + + it.effect("final flush pushes a pending tail chunk", () => + Effect.gen(function* () { + const { emits, state, make } = setup() + const throttle = yield* make + state.last = "head" + yield* throttle.onChunk(1) // first flush + state.last = "head+tail" + yield* throttle.onChunk(1) // coalesced, not yet emitted + yield* throttle.flush("final") + expect(emits).toEqual(["head", "head+tail"]) + }), + ) + + it.effect("spill flush emits immediately even under the byte threshold", () => + Effect.gen(function* () { + const { emits, state, make } = setup() + const throttle = yield* make + state.last = "x" + yield* throttle.onChunk(1) // first flush + state.last = "x-spilled" + yield* throttle.flush("spill") + expect(emits).toEqual(["x", "x-spilled"]) + }), + ) + + it.effect("timer does not emit when no new output arrived", () => + Effect.gen(function* () { + const { emits, state, make } = setup() + const throttle = yield* make + state.last = "a" + yield* throttle.onChunk(1) // first flush clears dirty + yield* TestClock.adjust(Duration.millis(150)) + yield* TestClock.adjust(Duration.millis(150)) + expect(emits).toEqual(["a"]) + }), + ) + + it.effect("final flush is a no-op when the latest output was already emitted", () => + Effect.gen(function* () { + const { emits, state, make } = setup() + const throttle = yield* make + state.last = "done" + yield* throttle.onChunk(1) // first flush clears dirty + yield* throttle.flush("final") + expect(emits).toEqual(["done"]) + }), + ) + + it.effect("ignores empty chunks and keeps the first-flush slot for the first real output", () => + Effect.gen(function* () { + const { emits, state, make } = setup() + const throttle = yield* make + state.last = "" // empty decode chunk: nothing new to preview + yield* throttle.onChunk(0) + expect(emits).toEqual([]) // no emit, and first-flush slot not consumed + state.last = "real" + yield* throttle.onChunk(4) // first real output still flushes synchronously + expect(emits).toEqual(["real"]) + }), + ) + + it.effect("swallows a defect from emit across first-chunk, timer, and final flush", () => + Effect.gen(function* () { + let calls = 0 + const state = { last: "" } + const throttle = yield* makeMetadataThrottle({ + intervalMillis: 150, + byteThreshold: 4 * 1024, + snapshot: () => state.last, + emit: () => { + calls += 1 + return Effect.die(new Error("metadata channel boom")) + }, + }) + state.last = "first" + yield* throttle.onChunk(1) // first-chunk synchronous flush + state.last = "first+timer" + yield* throttle.onChunk(1) + yield* TestClock.adjust(Duration.millis(150)) // timer flush + state.last = "first+timer+final" + yield* throttle.onChunk(1) + yield* throttle.flush("final") // final flush + // Reaching here proves none of the three flush paths propagated the defect. + expect(calls).toBeGreaterThanOrEqual(3) + }), + ) +})