feat(core): restore Modal model discovery

packages/core/src/plugin/provider.ts

@@ -16,6 +16,7 @@ import { GroqPlugin } from "./provider/groq.js"
 import { KiloPlugin } from "./provider/kilo.js"
 import { LLMGatewayPlugin } from "./provider/llmgateway.js"
 import { MistralPlugin } from "./provider/mistral.js"
+import { ModalPlugin } from "./provider/modal.js"
 import { NvidiaPlugin } from "./provider/nvidia.js"
 import { OpenAIPlugin } from "./provider/openai.js"
 import { SnowflakeCortexPlugin } from "./provider/snowflake-cortex.js"
@@ -49,6 +50,7 @@ export const ProviderPlugins: PluginInternal.InternalPlugin[] = [
   KiloPlugin,
   LLMGatewayPlugin,
   MistralPlugin,
+  ModalPlugin,
   NvidiaPlugin,
   OpencodePlugin,
   SnowflakeCortexPlugin,

packages/core/src/plugin/provider/modal.ts

@@ -0,0 +1,176 @@
+import { Money } from "@opencode-ai/schema/money"
+import { define } from "@opencode-ai/plugin/effect/plugin"
+import { Effect, Schema, Semaphore, Stream } from "effect"
+import { Bus } from "../../bus.js"
+import { Catalog } from "../../catalog.js"
+import { Integration } from "../../integration.js"
+import { Model } from "../../model.js"
+import { Provider } from "../../provider.js"
+
+const providerID = Provider.ID.make("modal")
+
+const ReasoningOption = Schema.Struct({
+  type: Schema.Literal("effort"),
+  values: Schema.Array(Schema.NullOr(Schema.String)),
+})
+
+const Response = Schema.Struct({
+  data: Schema.Array(
+    Schema.Struct({
+      id: Schema.String,
+      base_model_id: Schema.optional(Schema.String),
+      hugging_face_id: Schema.optional(Schema.String),
+      name: Schema.optional(Schema.String),
+      input_modalities: Schema.optional(Schema.Array(Schema.String)),
+      output_modalities: Schema.optional(Schema.Array(Schema.String)),
+      context_length: Schema.optional(Schema.Number),
+      max_output_length: Schema.optional(Schema.Number),
+      pricing: Schema.optional(
+        Schema.Struct({
+          prompt: Schema.optional(Schema.Union([Schema.String, Schema.Number])),
+          completion: Schema.optional(Schema.Union([Schema.String, Schema.Number])),
+          input_cache_read: Schema.optional(Schema.Union([Schema.String, Schema.Number])),
+        }),
+      ),
+      supported_features: Schema.optional(Schema.Array(Schema.String)),
+      reasoning_options: Schema.optional(Schema.Array(ReasoningOption)),
+      interleaved: Schema.optional(
+        Schema.Union([
+          Schema.Boolean,
+          Schema.Struct({ field: Schema.Literals(["reasoning", "reasoning_content", "reasoning_details"]) }),
+        ]),
+      ),
+    }),
+  ),
+})
+
+const decode = Schema.decodeUnknownSync(Response)
+
+export const ModalPlugin = define({
+  id: "opencode.provider.modal",
+  effect: Effect.fn(function* (ctx) {
+    const bus = yield* Bus.Service
+    const catalog = yield* Catalog.Service
+    const loading = Semaphore.makeUnsafe(1)
+    let templates: Map<Model.ID, Model.Info> | undefined
+    let models: Map<Model.ID, Model.Info> | undefined
+
+    const load = Effect.fn("ModalPlugin.load")(function* () {
+      const existing =
+        templates ??
+        new Map(
+          (yield* catalog.model.all())
+            .filter((model) => model.providerID === providerID)
+            .map((model) => [model.id, model]),
+        )
+      templates = existing
+      const connection = yield* ctx.integration.connection.active("modal")
+      const credential = connection
+        ? yield* ctx.integration.connection.resolve(connection).pipe(Effect.catch(() => Effect.succeed(undefined)))
+        : undefined
+      const provider = yield* catalog.provider.get(providerID)
+      const baseURL = typeof provider?.settings?.baseURL === "string" ? provider.settings.baseURL : undefined
+      if (credential?.type !== "key" || !baseURL) {
+        models = new Map()
+        return
+      }
+
+      models = yield* Effect.tryPromise({
+        try: () => discover(baseURL, credential.key, existing),
+        catch: (cause) => cause,
+      }).pipe(
+        Effect.catch((cause) =>
+          Effect.logWarning("failed to sync Modal models", { cause }).pipe(Effect.as(new Map<Model.ID, Model.Info>())),
+        ),
+      )
+    })
+
+    yield* ctx.catalog.transform((draft) => {
+      if (!models) return
+      const provider = draft.provider.get(providerID)
+      if (!provider) return
+      for (const id of provider.models.keys()) {
+        if (!models.has(Model.ID.make(id))) draft.model.remove(providerID, Model.ID.make(id))
+      }
+      for (const [id, model] of models) {
+        draft.model.update(providerID, id, (item) => Object.assign(item, structuredClone(model)))
+      }
+    })
+    const refresh = () => loading.withPermit(load().pipe(Effect.andThen(ctx.catalog.reload())))
+    yield* bus.subscribe(Integration.Event.ConnectionUpdated).pipe(
+      Stream.filter((event) => event.data.integrationID === Integration.ID.make("modal")),
+      Stream.runForEach(refresh),
+      Effect.forkScoped({ startImmediately: true }),
+    )
+    yield* refresh().pipe(Effect.forkScoped)
+  }),
+})
+
+async function discover(baseURL: string, apiKey: string, templates: ReadonlyMap<Model.ID, Model.Info>) {
+  const response = await fetch(`${baseURL.replace(/\/+$/, "")}/models`, {
+    headers: { Authorization: `Bearer ${apiKey}` },
+    signal: AbortSignal.timeout(3_000),
+  })
+  if (!response.ok) throw new Error(`Failed to fetch Modal models: ${response.status}`)
+
+  return new Map(
+    decode(await response.json()).data.map((item) => {
+      const id = Model.ID.make(item.id)
+      const template = templates.get(Model.ID.make(item.base_model_id ?? item.hugging_face_id ?? item.id))
+      return [id, build(id, item, baseURL, template)]
+    }),
+  )
+}
+
+function build(id: Model.ID, item: (typeof Response.Type)["data"][number], baseURL: string, template?: Model.Info) {
+  const fallback: Model.Info = template ?? Model.Info.make(Model.Info.default(providerID, id))
+  const baseCost = fallback.cost[0]
+  const variants = item.reasoning_options?.flatMap((option) =>
+    option.values.map((value) => {
+      const effort = value ?? "none"
+      return {
+        id: Model.VariantID.make(effort),
+        settings: { reasoningEffort: effort },
+      }
+    }),
+  )
+  return Model.Info.make({
+    ...structuredClone(fallback),
+    id,
+    modelID: id,
+    providerID,
+    name: item.name ?? fallback.name,
+    compatibility: Model.compatibility(item.interleaved) ?? fallback.compatibility,
+    package: fallback.package ?? Provider.aisdk("@ai-sdk/openai-compatible"),
+    settings: Provider.mergeOverlay(fallback.settings, { baseURL }),
+    capabilities: {
+      tools: item.supported_features?.includes("tools") ?? fallback.capabilities.tools,
+      input: item.input_modalities ? [...item.input_modalities] : [...fallback.capabilities.input],
+      output: item.output_modalities ? [...item.output_modalities] : [...fallback.capabilities.output],
+    },
+    variants: variants ?? [...fallback.variants],
+    cost: [
+      {
+        input: price(item.pricing?.prompt, baseCost?.input ?? Money.USDPerMillionTokens.zero),
+        output: price(item.pricing?.completion, baseCost?.output ?? Money.USDPerMillionTokens.zero),
+        cache: {
+          read: price(item.pricing?.input_cache_read, baseCost?.cache.read ?? Money.USDPerMillionTokens.zero),
+          write: baseCost?.cache.write ?? Money.USDPerMillionTokens.zero,
+        },
+      },
+    ],
+    limit: {
+      context: item.context_length ?? fallback.limit.context,
+      input: fallback.limit.input,
+      output: item.max_output_length ?? fallback.limit.output,
+    },
+    status: fallback.status,
+    enabled: fallback.enabled,
+  })
+}
+
+function price(value: string | number | undefined, fallback: number) {
+  if (value === undefined) return Money.USDPerMillionTokens.make(fallback)
+  const parsed = Number(value)
+  return Money.USDPerMillionTokens.make(Number.isFinite(parsed) ? parsed * 1_000_000 : fallback)
+}

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

@@ -150,6 +150,10 @@ 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,8 +1,7 @@
 import { describe, expect } from "bun:test"
 import { Message, SystemPart } from "@opencode-ai/ai"
-import { DateTime, Deferred, Effect, Schema, Stream } from "effect"
+import { DateTime, Effect, Schema } 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"
@@ -19,7 +18,6 @@ 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"
@@ -439,170 +437,6 @@ 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/plugin/provider-modal.test.ts

@@ -0,0 +1,143 @@
+import { Catalog } from "@opencode-ai/core/catalog"
+import { Integration } from "@opencode-ai/core/integration"
+import { Model } from "@opencode-ai/core/model"
+import { Plugin } from "@opencode-ai/core/plugin"
+import { PluginHost } from "@opencode-ai/core/plugin/host"
+import { ModalPlugin } from "@opencode-ai/core/plugin/provider/modal"
+import { Provider } from "@opencode-ai/core/provider"
+import { State } from "@opencode-ai/core/state"
+import { Money } from "@opencode-ai/schema/money"
+import { expect } from "bun:test"
+import { Effect } from "effect"
+import { testEffect } from "../lib/effect"
+import { PluginTestLayer } from "./fixture"
+
+const it = testEffect(PluginTestLayer)
+const providerID = Provider.ID.make("modal")
+const integrationID = Integration.ID.make("modal")
+const baseModelID = Model.ID.make("thinkingmachines/Inkling-NVFP4")
+const runtimeModelID = Model.ID.make("workspace--inkling.us-west.modal.direct")
+
+function eventually<A>(
+  effect: Effect.Effect<A>,
+  predicate: (value: A) => boolean,
+  remaining = 1000,
+): Effect.Effect<A, Error> {
+  return Effect.gen(function* () {
+    const value = yield* effect
+    if (predicate(value)) return value
+    if (remaining === 0) return yield* Effect.fail(new Error("Timed out waiting for value"))
+    yield* Effect.promise(() => Bun.sleep(1))
+    return yield* eventually(effect, predicate, remaining - 1)
+  })
+}
+
+const setup = Effect.fn(function* (baseURL: string, key?: string) {
+  const catalog = yield* Catalog.Service
+  const integrations = yield* Integration.Service
+  yield* integrations.transform((draft) => {
+    draft.method.update({ integrationID, method: { type: "key" } })
+  })
+  if (key) yield* integrations.connection.key({ integrationID, key })
+  yield* State.batch(
+    Effect.gen(function* () {
+      yield* catalog.transform((draft) => {
+        draft.provider.update(providerID, (provider) => {
+          provider.name = "Modal"
+          provider.package = Provider.aisdk("@ai-sdk/openai-compatible")
+          provider.settings = { baseURL }
+          provider.integrationID = integrationID
+        })
+        draft.model.update(providerID, baseModelID, (model) => {
+          model.name = "Inkling"
+          model.family = Model.Family.make("ling")
+          model.compatibility = { reasoningField: "reasoning_content" }
+          model.capabilities = { tools: true, input: ["text", "image", "audio"], output: ["text"] }
+          model.variants = [{ id: Model.VariantID.make("fallback"), settings: { reasoningEffort: "fallback" } }]
+          model.cost = [
+            {
+              input: Money.USDPerMillionTokens.make(1),
+              output: Money.USDPerMillionTokens.make(4),
+              cache: {
+                read: Money.USDPerMillionTokens.make(0.2),
+                write: Money.USDPerMillionTokens.zero,
+              },
+            },
+          ]
+          model.limit = { context: 128_000, output: 8_192 }
+          model.time = { released: Date.parse("2026-07-15") }
+        })
+      })
+      yield* ModalPlugin.effect(yield* PluginHost.make(yield* Plugin.Service))
+    }),
+  )
+})
+
+it.live("discovers Modal workspace models", () =>
+  Effect.gen(function* () {
+    const requests: Array<{ authorization: string | null; path: string }> = []
+    using server = Bun.serve({
+      port: 0,
+      fetch(request) {
+        requests.push({ authorization: request.headers.get("authorization"), path: new URL(request.url).pathname })
+        return Response.json({
+          data: [
+            {
+              id: runtimeModelID,
+              base_model_id: baseModelID,
+              name: "Thinking Machines: Inkling",
+              input_modalities: ["text", "image", "audio"],
+              output_modalities: ["text"],
+              context_length: 1_048_576,
+              max_output_length: 262_144,
+              pricing: { prompt: "0.0000012", completion: "0.000005", input_cache_read: "0.00000027" },
+              supported_features: ["tools", "reasoning"],
+              reasoning_options: [{ type: "effort", values: ["none", "low", "high"] }],
+              interleaved: { field: "reasoning_content" },
+            },
+          ],
+        })
+      },
+    })
+    yield* setup(`${server.url}v1`, "test-token")
+
+    const models = yield* eventually(
+      (yield* Catalog.Service).model
+        .all()
+        .pipe(Effect.map((models) => models.filter((model) => model.providerID === providerID))),
+      (models) => models.some((model) => model.id === runtimeModelID),
+    )
+    expect(requests).toEqual([{ authorization: "Bearer test-token", path: "/v1/models" }])
+    expect(models).toHaveLength(1)
+    expect(models[0]).toMatchObject({
+      id: runtimeModelID,
+      modelID: runtimeModelID,
+      name: "Thinking Machines: Inkling",
+      family: "ling",
+      compatibility: { reasoningField: "reasoning_content" },
+      settings: { baseURL: `${server.url}v1` },
+      capabilities: { tools: true, input: ["text", "image", "audio"], output: ["text"] },
+      variants: [
+        { id: "none", settings: { reasoningEffort: "none" } },
+        { id: "low", settings: { reasoningEffort: "low" } },
+        { id: "high", settings: { reasoningEffort: "high" } },
+      ],
+      cost: [{ input: 1.2, output: 5, cache: { read: 0.27, write: 0 } }],
+      limit: { context: 1_048_576, output: 262_144 },
+    })
+  }),
+)
+
+it.live("hides static Modal models when discovery fails", () =>
+  Effect.gen(function* () {
+    using server = Bun.serve({ port: 0, fetch: () => new Response(null, { status: 503 }) })
+    yield* setup(`${server.url}v1`, "test-token")
+    const models = yield* eventually(
+      (yield* Catalog.Service).model
+        .all()
+        .pipe(Effect.map((models) => models.filter((model) => model.providerID === providerID))),
+      (models) => models.length === 0,
+    )
+    expect(models).toEqual([])
+  }),
+)

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

@@ -217,16 +217,13 @@ it.effect("batches text deltas and flushes pending text before the terminal even
       { discard: true },
     )
 
-    expect(published.filter((event) => event.type === "session.text.delta").map((event) => event.data)).toMatchObject([
-      { delta: "one" },
-    ])
+    expect(published.filter((event) => event.type === "session.text.delta")).toHaveLength(0)
     yield* TestClock.adjust("99 millis")
-    expect(published.filter((event) => event.type === "session.text.delta")).toHaveLength(1)
+    expect(published.filter((event) => event.type === "session.text.delta")).toHaveLength(0)
     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" },
-      { delta: " two three four" },
+      { delta: "one two three four" },
     ])
 
     yield* publisher.publish(LLMEvent.textDelta({ id: "text", text: " five" }))
@@ -253,7 +250,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" }, { delta: " two three" }])
+    ).toMatchObject([{ delta: "one 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(2), Stream.runCollect, Effect.forkScoped)
+      ? yield* bus.subscribe(fixture.delta).pipe(Stream.take(1), 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(2)
+      expect(streamed).toHaveLength(1)
       expect(
         streamed
           .map((event) => {

packages/plugin/src/promise/adapter.ts

@@ -1,5 +1,5 @@
 import { Tool } from "@opencode-ai/schema/tool"
-import { Cause, Effect, Exit, Queue, Schema, SchemaAST, Scope, Stream } from "effect"
+import { Effect, 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,55 +149,13 @@ export function fromPromise(plugin: Plugin) {
             reload: () => run(host.command.reload()),
           },
           event: {
-            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
-                },
-              }
-            },
+            subscribe: () =>
+              Stream.toAsyncIterable(
+                host.event.subscribe().pipe(
+                  Stream.mapEffect((event) => Schema.encodeUnknownEffect(OpenCodeEvent)(event)),
+                  Stream.map((event) => event as unknown as PromiseEvent),
+                ),
+              ),
           },
           integration: {
             list: adaptApiMethod(IntegrationEndpoints["integration.list"], host.integration.list),