fix(plugin): scope Promise event iterators

packages/cli/package.json

@@ -41,8 +41,8 @@
     "solid-js": "catalog:",
     "tree-sitter-bash": "0.25.0",
     "tree-sitter-powershell": "0.25.10",
-    "uqr": "0.1.3",
     "web-tree-sitter": "0.25.10",
+    "uqr": "0.1.3",
     "ws": "8.21.0"
   },
   "devDependencies": {

packages/cli/script/build.ts

@@ -102,17 +102,6 @@ for (const item of targets) {
   const name = target.replace(binary, "cli")
   const executablePath = await compileExecutable(item)
   console.log(`building ${name}`)
-  const compileTarget = {
-    autoloadBunfig: false,
-    autoloadDotenv: false,
-    autoloadTsconfig: true,
-    autoloadPackageJson: true,
-    target: target.replace(binary, "bun") as Bun.Build.CompileTarget,
-    ...(executablePath ? { executablePath } : {}),
-    outfile: path.join(outdir, name, "bin", binary),
-    execArgv: [`--user-agent=${binary}/${Script.version}`, "--use-system-ca", "--no-warnings", "--"],
-    windows: {},
-  }
   const result = await Bun.build({
     entrypoints: ["./src/index.ts"],
     tsconfig: "./tsconfig.json",
@@ -122,7 +111,17 @@ for (const item of targets) {
     minify: true,
     sourcemap: "inline",
     splitting: true,
-    compile: compileTarget,
+    compile: {
+      autoloadBunfig: false,
+      autoloadDotenv: false,
+      autoloadTsconfig: true,
+      autoloadPackageJson: true,
+      target: target.replace(binary, "bun") as Bun.Build.CompileTarget,
+      executablePath,
+      outfile: path.join(outdir, name, "bin", binary),
+      execArgv: [`--user-agent=${binary}/${Script.version}`, "--use-system-ca", "--"],
+      windows: {},
+    },
     define: {
       OPENCODE_VERSION: `'${Script.version}'`,
       OPENCODE_CLI_NAME: `'${binary}'`,

packages/core/src/session/runner/publish-llm-event.ts

@@ -150,10 +150,6 @@ export const createLLMEventPublisher = (bus: Pick<Bus.Interface, "publish">, inp
       if (!current) return yield* Effect.die(new Error(`${name} delta before start: ${id}`))
       if (!current.pending) return undefined
       const now = yield* Clock.currentTimeMillis
-      if (!force && current.publishedAt === undefined) {
-        current.publishedAt = now
-        return undefined
-      }
       if (!force && current.publishedAt !== undefined && now - current.publishedAt < deltaBatchInterval)
         return undefined
       yield* delta(id, current.pending, current.ordinal)

packages/core/test/plugin/promise.test.ts

@@ -1,7 +1,8 @@
 import { describe, expect } from "bun:test"
 import { Message, SystemPart } from "@opencode-ai/ai"
-import { DateTime, Effect, Schema } from "effect"
+import { DateTime, Deferred, Effect, Schema, Stream } from "effect"
 import { Agent } from "@opencode-ai/core/agent"
+import { Bus } from "@opencode-ai/core/bus"
 import { Catalog } from "@opencode-ai/core/catalog"
 import { Model } from "@opencode-ai/core/model"
 import { Location } from "@opencode-ai/core/location"
@@ -18,6 +19,7 @@ import { Provider } from "@opencode-ai/core/provider"
 import { Project } from "@opencode-ai/core/project"
 import { AbsolutePath } from "@opencode-ai/core/schema"
 import { define } from "@opencode-ai/plugin/promise/plugin"
+import { Config as ConfigSchema } from "@opencode-ai/schema/config"
 import { Money } from "@opencode-ai/schema/money"
 import type { SessionHooks } from "@opencode-ai/plugin/effect/session"
 import { testEffect } from "../lib/effect"
@@ -437,6 +439,170 @@ describe("fromPromise", () => {
     }),
   )
 
+  it.effect("closes a pending event iterator with the plugin scope", () =>
+    Effect.gen(function* () {
+      const started = yield* Deferred.make<void>()
+      let finalized = 0
+      let iterator: AsyncIterator<unknown> | undefined
+      let pending: Promise<IteratorResult<unknown>> | undefined
+      const host = testHost({
+        event: {
+          subscribe: () =>
+            Stream.fromEffect(Deferred.succeed(started, undefined)).pipe(
+              Stream.flatMap(() => Stream.never),
+              Stream.ensuring(Effect.sync(() => finalized++)),
+            ),
+        },
+      })
+
+      yield* Effect.scoped(
+        PluginPromise.fromPromise(
+          define({
+            id: "promise-event-pending",
+            setup: async (ctx) => {
+              iterator = ctx.event.subscribe()[Symbol.asyncIterator]()
+              pending = iterator.next()
+              await Effect.runPromise(Deferred.await(started))
+            },
+          }),
+        ).effect(host),
+      )
+
+      expect(finalized).toBe(1)
+      if (!pending || !iterator) yield* Effect.die("event iterator was not initialized")
+      expect(yield* Effect.promise(() => pending)).toEqual({ done: true, value: undefined })
+      expect(yield* Effect.promise(() => iterator.next())).toEqual({ done: true, value: undefined })
+    }),
+  )
+
+  it.live("closes event iterators on break, completion, and failure", () =>
+    Effect.gen(function* () {
+      const bus = yield* Bus.Service
+      const plugins = yield* Plugin.Service
+      const closed: string[] = []
+      const broke = yield* Deferred.make<void>()
+      const promisePlugin = define({
+        id: "promise-event-terminal",
+        setup: async (ctx) => {
+          void (async () => {
+            for await (const _event of ctx.event.subscribe()) break
+            await Effect.runPromise(Deferred.succeed(broke, undefined))
+          })()
+        },
+      })
+
+      yield* plugins.activate([{ ...PluginPromise.fromPromise(promisePlugin), version: "1" }])
+      yield* Effect.sleep("10 millis")
+      yield* bus.publish(ConfigSchema.Event.Updated, {})
+      yield* Deferred.await(broke)
+
+      yield* Effect.scoped(
+        PluginPromise.fromPromise(
+          define({
+            id: "promise-event-complete",
+            setup: async (ctx) => {
+              const iterator = ctx.event.subscribe()[Symbol.asyncIterator]()
+              expect(await iterator.next()).toEqual({ done: true, value: undefined })
+              expect(await iterator.next()).toEqual({ done: true, value: undefined })
+            },
+          }),
+        ).effect(
+          testHost({
+            event: {
+              subscribe: () => Stream.empty.pipe(Stream.ensuring(Effect.sync(() => closed.push("complete")))),
+            },
+          }),
+        ),
+      )
+
+      yield* Effect.scoped(
+        PluginPromise.fromPromise(
+          define({
+            id: "promise-event-failure",
+            setup: async (ctx) => {
+              const iterator = ctx.event.subscribe()[Symbol.asyncIterator]()
+              await expect(iterator.next()).rejects.toThrow("event failure")
+              expect(await iterator.next()).toEqual({ done: true, value: undefined })
+            },
+          }),
+        ).effect(
+          testHost({
+            event: {
+              subscribe: () =>
+                Stream.fail(new Error("event failure")).pipe(
+                  Stream.ensuring(Effect.sync(() => closed.push("failure"))),
+                ),
+            },
+          }),
+        ),
+      )
+
+      expect(closed).toEqual(["complete", "failure"])
+    }),
+  )
+
+  it.effect("closes every event iterator when the plugin scope closes", () =>
+    Effect.gen(function* () {
+      const ready = yield* Deferred.make<void>()
+      let started = 0
+      let finalized = 0
+      const host = testHost({
+        event: {
+          subscribe: () =>
+            Stream.fromEffect(
+              Effect.sync(() => ++started).pipe(
+                Effect.tap((count) => (count === 3 ? Deferred.succeed(ready, undefined) : Effect.void)),
+              ),
+            ).pipe(
+              Stream.flatMap(() => Stream.never),
+              Stream.ensuring(Effect.sync(() => finalized++)),
+            ),
+        },
+      })
+
+      yield* Effect.scoped(
+        PluginPromise.fromPromise(
+          define({
+            id: "promise-event-multiple",
+            setup: async (ctx) => {
+              const events = ctx.event.subscribe()
+              void events[Symbol.asyncIterator]().next()
+              void events[Symbol.asyncIterator]().next()
+              void ctx.event.subscribe()[Symbol.asyncIterator]().next()
+              await Effect.runPromise(Deferred.await(ready))
+            },
+          }),
+        ).effect(host),
+      )
+
+      expect(finalized).toBe(3)
+    }),
+  )
+
+  it.effect("closes a Promise event iterator when the plugin is replaced", () =>
+    Effect.gen(function* () {
+      const plugins = yield* Plugin.Service
+      let iterator: AsyncIterator<unknown> | undefined
+      let pending: Promise<IteratorResult<unknown>> | undefined
+      const previous = PluginPromise.fromPromise(
+        define({
+          id: "promise-event-replacement",
+          setup: async (ctx) => {
+            iterator = ctx.event.subscribe()[Symbol.asyncIterator]()
+            pending = iterator.next()
+          },
+        }),
+      )
+
+      yield* plugins.activate([{ ...previous, version: "1" }])
+      yield* plugins.activate([{ id: previous.id, version: "2", effect: () => Effect.void }])
+
+      if (!pending || !iterator) yield* Effect.die("event iterator was not initialized")
+      expect(yield* Effect.promise(() => pending)).toEqual({ done: true, value: undefined })
+      expect(yield* Effect.promise(() => iterator.next())).toEqual({ done: true, value: undefined })
+    }),
+  )
+
   it.effect("constructs plain Promise tool definitions in the host", () =>
     Effect.gen(function* () {
       const plugins = yield* Plugin.Service

packages/core/test/session-runner-tool-events.test.ts

@@ -217,13 +217,16 @@ it.effect("batches text deltas and flushes pending text before the terminal even
       { discard: true },
     )
 
-    expect(published.filter((event) => event.type === "session.text.delta")).toHaveLength(0)
+    expect(published.filter((event) => event.type === "session.text.delta").map((event) => event.data)).toMatchObject([
+      { delta: "one" },
+    ])
     yield* TestClock.adjust("99 millis")
-    expect(published.filter((event) => event.type === "session.text.delta")).toHaveLength(0)
+    expect(published.filter((event) => event.type === "session.text.delta")).toHaveLength(1)
     yield* TestClock.adjust("1 millis")
     yield* publisher.publish(LLMEvent.textDelta({ id: "text", text: " four" }))
     expect(published.filter((event) => event.type === "session.text.delta").map((event) => event.data)).toMatchObject([
-      { delta: "one two three four" },
+      { delta: "one" },
+      { delta: " two three four" },
     ])
 
     yield* publisher.publish(LLMEvent.textDelta({ id: "text", text: " five" }))
@@ -250,7 +253,7 @@ it.effect("batches reasoning deltas and flushes pending reasoning before the ter
 
     expect(
       published.filter((event) => event.type === "session.reasoning.delta").map((event) => event.data),
-    ).toMatchObject([{ delta: "one two three" }])
+    ).toMatchObject([{ delta: "one" }, { delta: " two three" }])
     expect(published.slice(-2).map((event) => event.type)).toEqual([
       "session.reasoning.delta",
       "session.reasoning.ended.1",

packages/core/test/session-runner.test.ts

@@ -767,7 +767,7 @@ const verifyEphemeralDeltas = (kind: FragmentKind) =>
     yield* admit(session, prompt)
     const bus = yield* Bus.Service
     const live = fixture.delta
-      ? yield* bus.subscribe(fixture.delta).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+      ? yield* bus.subscribe(fixture.delta).pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped)
       : undefined
     yield* Effect.yieldNow
     yield* TestLLM.push(fixture.completeEvents)
@@ -785,7 +785,7 @@ const verifyEphemeralDeltas = (kind: FragmentKind) =>
       : []
     if (live) {
       const streamed = Array.from(yield* Fiber.join(live))
-      expect(streamed).toHaveLength(1)
+      expect(streamed).toHaveLength(2)
       expect(
         streamed
           .map((event) => {

packages/plugin/src/promise/adapter.ts

@@ -1,5 +1,5 @@
 import { Tool } from "@opencode-ai/schema/tool"
-import { Effect, Schema, SchemaAST, Scope, Stream } from "effect"
+import { Cause, Effect, Exit, Queue, Schema, SchemaAST, Scope, Stream } from "effect"
 import { HttpApiEndpoint, HttpApiSchema } from "effect/unstable/httpapi"
 import { define } from "../effect/plugin.js"
 import type { Context, Plugin } from "./plugin.js"
@@ -149,13 +149,55 @@ export function fromPromise(plugin: Plugin) {
             reload: () => run(host.command.reload()),
           },
           event: {
-            subscribe: () =>
-              Stream.toAsyncIterable(
-                host.event.subscribe().pipe(
-                  Stream.mapEffect((event) => Schema.encodeUnknownEffect(OpenCodeEvent)(event)),
-                  Stream.map((event) => event as unknown as PromiseEvent),
-                ),
-              ),
+            subscribe: () => {
+              const events = host.event.subscribe().pipe(
+                Stream.mapEffect((event) => Schema.encodeUnknownEffect(OpenCodeEvent)(event)),
+                Stream.map((event) => event as unknown as PromiseEvent),
+              )
+              return {
+                [Symbol.asyncIterator]() {
+                  const child = Scope.forkUnsafe(scope)
+                  const done = { done: true, value: undefined } as const
+                  let terminal = false
+                  let closing: Promise<IteratorResult<PromiseEvent>> | undefined
+                  const queue = Effect.gen(function* () {
+                    const queue = yield* Stream.toQueue(events, { capacity: "unbounded" })
+                    // Finalizers are LIFO: mark terminal before queue shutdown wakes a pending next().
+                    yield* Scope.addFinalizer(
+                      child,
+                      Effect.sync(() => (terminal = true)),
+                    )
+                    return queue
+                  }).pipe(Scope.provide(child), Effect.runPromiseWith(context))
+                  const iterator = {
+                    next: () => {
+                      if (terminal) return closing ?? Promise.resolve(done)
+                      return queue
+                        .then((queue) => Effect.runPromiseWith(context)(Queue.take(queue)))
+                        .then(
+                          (value) => (terminal ? (closing ?? done) : { done: false as const, value }),
+                          async (error) => {
+                            if (terminal) return closing ?? done
+                            await iterator.return()
+                            if (Cause.isDone(error)) return done
+                            throw error
+                          },
+                        )
+                    },
+                    return: () => {
+                      if (closing) return closing
+                      terminal = true
+                      closing = Effect.runPromiseWith(context)(Scope.close(child, Exit.void)).then(
+                        () => done,
+                        () => done,
+                      )
+                      return closing
+                    },
+                  }
+                  return iterator
+                },
+              }
+            },
           },
           integration: {
             list: adaptApiMethod(IntegrationEndpoints["integration.list"], host.integration.list),