Fix beta integration

Co-authored-by: Luke Parker <10430890+Hona@users.noreply.github.com>

.github/workflows/publish.yml

@@ -466,6 +466,7 @@ jobs:
           VITE_SENTRY_ENVIRONMENT: ${{ (github.ref_name == 'beta' && 'beta') || 'production' }}
           VITE_SENTRY_RELEASE: desktop@${{ needs.version.outputs.version }}
           OPENCODE_CLI_TARGET: ${{ matrix.settings.target }}
+          OPENCODE_CLI_DIST: ${{ github.workspace }}/packages/cli/dist
 
       - name: Package
         if: needs.version.outputs.release

packages/app/src/pages/new-session/new-session-view.tsx

@@ -60,7 +60,11 @@ export function NewSessionView(props: {
                   <Show
                     when={props.workspace.bar.visible()}
                     fallback={
-                      <PromptGitStatus branch={props.workspace.bar.branch()} noGit={!props.workspace.project.git()} />
+                      <PromptGitStatus
+                        branch={props.workspace.bar.branch()}
+                        noGit={!props.workspace.project.git()}
+                        class="ms-1"
+                      />
                     }
                   >
                     <PromptWorkspaceSelector

packages/cli/bin/opencode2.cjs

@@ -5,7 +5,8 @@ const fs = require("fs")
 const path = require("path")
 const os = require("os")
 
-const forwardedSignals = ["SIGINT", "SIGTERM", "SIGHUP"]
+const forwardedSignals =
+  process.platform === "win32" ? ["SIGINT", "SIGTERM", "SIGHUP"] : ["SIGINT", "SIGTERM", "SIGHUP", "SIGUSR1"]
 
 function run(target) {
   const child = childProcess.spawn(target, process.argv.slice(2), { stdio: "inherit" })

packages/cli/package.json

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

packages/cli/script/build.ts

@@ -117,9 +117,9 @@ for (const item of targets) {
       autoloadTsconfig: true,
       autoloadPackageJson: true,
       target: target.replace(binary, "bun") as Bun.Build.CompileTarget,
-      executablePath,
+      ...(executablePath ? { executablePath } : {}),
       outfile: path.join(outdir, name, "bin", binary),
-      execArgv: [`--user-agent=${binary}/${Script.version}`, "--use-system-ca", "--"],
+      execArgv: [`--user-agent=${binary}/${Script.version}`, "--use-system-ca", "--no-warnings", "--"],
       windows: {},
     },
     define: {

packages/cli/src/commands/commands.ts

@@ -1,5 +1,6 @@
-import { Argument, Flag } from "effect/unstable/cli"
+import { Argument, Command, Flag } from "effect/unstable/cli"
 import { Spec } from "../framework/spec"
+import { GlobalFlags } from "./global-flags"
 
 declare const OPENCODE_CLI_NAME: string | undefined
 
@@ -26,7 +27,7 @@ const PermissionParams = {
   ),
 }
 
-export const Commands = Spec.make(typeof OPENCODE_CLI_NAME === "string" ? OPENCODE_CLI_NAME : "opencode", {
+const Root = Spec.make(typeof OPENCODE_CLI_NAME === "string" ? OPENCODE_CLI_NAME : "opencode", {
   description: "OpenCode 2.0 preview command line interface",
   params: {
     ...ServerParams,
@@ -70,7 +71,7 @@ export const Commands = Spec.make(typeof OPENCODE_CLI_NAME === "string" ? OPENCO
       description: "Debugging and troubleshooting tools",
       commands: [
         Spec.make("agents", { description: "List all agents" }),
-        Spec.make("config", { description: "Show resolved configuration" }),
+        Spec.make("config", { description: "List configuration sources" }),
       ],
     }),
     Spec.make("console", {
@@ -277,3 +278,5 @@ export const Commands = Spec.make(typeof OPENCODE_CLI_NAME === "string" ? OPENCO
     }),
   ],
 })
+
+export const Commands = { ...Root, spec: Root.spec.pipe(Command.withGlobalFlags(GlobalFlags.all)) }

packages/cli/src/commands/global-flags.ts

@@ -0,0 +1,12 @@
+export * as GlobalFlags from "./global-flags"
+
+import { Flag, GlobalFlag } from "effect/unstable/cli"
+
+export const CpuProfile = GlobalFlag.setting("cpu-profile")({
+  flag: Flag.string("cpu-profile").pipe(
+    Flag.withDescription("Write a CPU profile to this path when the process stops"),
+    Flag.optional,
+  ),
+})
+
+export const all = [CpuProfile] as const

packages/cli/src/cpu-profile.ts

@@ -0,0 +1,45 @@
+export * as CpuProfile from "./cpu-profile"
+
+import { Effect, FileSystem } from "effect"
+import { Session } from "node:inspector"
+import path from "node:path"
+
+export function run<A, E, R>(file: string, effect: Effect.Effect<A, E, R>) {
+  const target = path.resolve(file)
+  return Effect.acquireUseRelease(
+    Effect.gen(function* () {
+      const fs = yield* FileSystem.FileSystem
+      yield* fs.makeDirectory(path.dirname(target), { recursive: true })
+      const session = new Session()
+      session.connect()
+      yield* command(session, "Profiler.enable")
+      yield* command(session, "Profiler.start")
+      yield* Effect.logInfo("CPU profile started", { path: target })
+      return session
+    }),
+    () => effect,
+    (session) =>
+      Effect.tryPromise(
+        () =>
+          new Promise<void>((resolve, reject) => {
+            session.post("Profiler.stop", (error, result) => {
+              session.disconnect()
+              if (error) return reject(error)
+              Bun.write(target, JSON.stringify(result.profile)).then(() => resolve(), reject)
+            })
+          }),
+      ).pipe(
+        Effect.andThen(Effect.logInfo("CPU profile written", { path: target })),
+        Effect.catchCause((cause) => Effect.logError("Failed to write CPU profile", { path: target, cause })),
+      ),
+  )
+}
+
+function command(session: Session, method: "Profiler.enable" | "Profiler.start") {
+  return Effect.tryPromise(
+    () =>
+      new Promise<void>((resolve, reject) => {
+        session.post(method, (error) => (error ? reject(error) : resolve()))
+      }),
+  )
+}

packages/cli/src/framework/runtime.ts

@@ -1,10 +1,13 @@
-import { Effect, FileSystem, Scope } from "effect"
+import { Effect, FileSystem, Option, Scope } from "effect"
 import { Command } from "effect/unstable/cli"
 import { Spec } from "./spec"
 import { Global } from "@opencode-ai/util/global"
 import { Updater } from "../services/updater"
 import { Config } from "../config"
 import { Npm } from "@opencode-ai/util/npm"
+import { GlobalFlags } from "../commands/global-flags"
+import { CpuProfile } from "../cpu-profile"
+import path from "node:path"
 
 export type Input<Value> =
   Value extends Spec.Node<infer _Name, infer Command, infer _Commands>
@@ -86,7 +89,22 @@ function provide(node: Spec.Any, handlers: ReadonlyArray<LazyHandler>): Provided
     ? node.spec.pipe(
         Command.withHandler((input) =>
           Effect.gen(function* () {
-            yield* Effect.flatMap(Effect.promise(handler.load), (module) => module.default(input))
+            const module = yield* Effect.promise(handler.load)
+            const cpuProfile = Option.getOrUndefined(yield* GlobalFlags.CpuProfile)
+            if (!cpuProfile) return yield* module.default(input)
+            const target = path.resolve(cpuProfile)
+            const previous = process.env.OPENCODE_CPU_PROFILE
+            process.env.OPENCODE_CPU_PROFILE = target
+            return yield* (
+              node.name === "serve" ? CpuProfile.run(target, module.default(input)) : module.default(input)
+            ).pipe(
+              Effect.ensuring(
+                Effect.sync(() => {
+                  if (previous === undefined) delete process.env.OPENCODE_CPU_PROFILE
+                  else process.env.OPENCODE_CPU_PROFILE = previous
+                }),
+              ),
+            )
           }),
         ),
       )

packages/cli/src/heap.ts

@@ -0,0 +1,37 @@
+import { Global } from "@opencode-ai/util/global"
+import { Effect, Queue } from "effect"
+import path from "node:path"
+
+export const listen = Effect.gen(function* () {
+  const global = yield* Global.Service
+  if (process.platform === "win32") return
+  const signals = yield* Queue.dropping<void>(1)
+  yield* Effect.acquireRelease(
+    Effect.sync(() => {
+      const handler = () => Queue.offerUnsafe(signals, undefined)
+      process.on("SIGUSR1", handler)
+      return handler
+    }),
+    (handler) => Effect.sync(() => process.off("SIGUSR1", handler)),
+  )
+  yield* Queue.take(signals).pipe(
+    Effect.andThen(
+      Effect.suspend(() => {
+        const file = path.join(
+          global.log,
+          `heap-${process.pid}-${new Date().toISOString().replace(/[:.]/g, "")}.heapsnapshot`,
+        )
+        return Effect.gen(function* () {
+          yield* Effect.logInfo("writing heap snapshot", { path: file })
+          const { writeHeapSnapshot } = yield* Effect.tryPromise(() => import("node:v8"))
+          yield* Effect.try(() => writeHeapSnapshot(file))
+          yield* Effect.logInfo("heap snapshot written", { path: file })
+        }).pipe(Effect.catchCause((cause) => Effect.logError("failed to write heap snapshot", { path: file, cause })))
+      }),
+    ),
+    Effect.forever,
+    Effect.forkScoped({ startImmediately: true }),
+  )
+})
+
+export * as Heap from "./heap"

packages/cli/src/index.ts

@@ -12,6 +12,7 @@ import { Global } from "@opencode-ai/util/global"
 import { AppProcess } from "@opencode-ai/util/process"
 import { Config } from "./config"
 import { Npm } from "@opencode-ai/util/npm"
+import { Heap } from "./heap"
 
 const Handlers = Runtime.handlers(Commands, {
   $: () => import("./commands/handlers/default"),
@@ -54,13 +55,16 @@ const Handlers = Runtime.handlers(Commands, {
   serve: () => import("./commands/handlers/serve"),
 })
 
-Effect.logInfo("cli starting", {
-  version: OPENCODE_VERSION,
-  channel: OPENCODE_CHANNEL,
-  local: OPENCODE_LOCAL,
-  args: process.argv.slice(2),
+Effect.gen(function* () {
+  yield* Heap.listen
+  yield* Effect.logInfo("cli starting", {
+    version: OPENCODE_VERSION,
+    channel: OPENCODE_CHANNEL,
+    local: OPENCODE_LOCAL,
+    args: process.argv.slice(2),
+  })
+  return yield* Runtime.run(Commands, Handlers, { version: OPENCODE_VERSION })
 }).pipe(
-  Effect.flatMap(() => Runtime.run(Commands, Handlers, { version: OPENCODE_VERSION })),
   Effect.annotateLogs({ role: "cli" }),
   Effect.provide(Config.layer),
   Effect.provide(Updater.layer),

packages/cli/src/services/service-config.ts

@@ -104,7 +104,12 @@ export const options = Effect.fnUntraced(function* (input: { readonly checkVersi
   return {
     file,
     version: input.checkVersion ? OPENCODE_VERSION : undefined,
-    command: [...selfCommand(), "serve", "--service"],
+    command: [
+      ...selfCommand(),
+      "serve",
+      "--service",
+      ...(process.env.OPENCODE_CPU_PROFILE ? ["--cpu-profile", process.env.OPENCODE_CPU_PROFILE] : []),
+    ],
   }
 })
 

packages/cli/test/debug-config.test.ts

@@ -10,9 +10,10 @@ describe("debug config command", () => {
 
     expect(debug.exitCode).toBe(0)
     expect(debug.stdout).toContain("config")
-    expect(debug.stdout).toContain("Show resolved configuration")
+    expect(debug.stdout).toContain("List configuration sources")
     expect(config.exitCode).toBe(0)
     expect(config.stdout).toContain("opencode debug config [flags]")
+    expect(config.stdout).toContain("List configuration sources")
   })
 
   test("prints config entries from the invoking directory without reordering permissions", async () => {

packages/cli/test/service.test.ts

@@ -19,6 +19,29 @@ test("managed service ports are stable per installation channel", () => {
   expect(ServiceConfig.defaultPort("preview-a")).not.toBe(ServiceConfig.defaultPort("preview-b"))
 })
 
+test("managed service forwards the CPU profile path to the server", async () => {
+  const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-profile-"))
+  const profile = path.join(root, "server.cpuprofile")
+  try {
+    const previous = process.env.OPENCODE_CPU_PROFILE
+    process.env.OPENCODE_CPU_PROFILE = profile
+    try {
+      const options = await Effect.runPromise(
+        ServiceConfig.options().pipe(
+          Effect.provide(Global.layerWith({ config: path.join(root, "config"), state: path.join(root, "state") })),
+          Effect.provide(NodeFileSystem.layer),
+        ),
+      )
+      expect(options.command.slice(-2)).toEqual(["--cpu-profile", profile])
+    } finally {
+      if (previous === undefined) delete process.env.OPENCODE_CPU_PROFILE
+      else process.env.OPENCODE_CPU_PROFILE = previous
+    }
+  } finally {
+    await fs.rm(root, { recursive: true, force: true })
+  }
+})
+
 test("local channel stores service config with the local service filename", async () => {
   const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-"))
   try {

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/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),