perf(opencode): throttle shell tool metadata streaming

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<void>
+}
+
+export interface MetadataThrottle {
+  onChunk: (size: number) => Effect.Effect<void>
+  flush: (reason: "spill" | "final") => Effect.Effect<void>
+}
+
+// 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<MetadataThrottle, never, Scope.Scope> =>
+  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 }
+  })

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)

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)
+    }),
+  )
+})