fix: preserve context for SDK event stream

packages/opencode/src/server/routes/instance/httpapi/event.ts

@@ -1,6 +1,6 @@
 import { Bus } from "@/bus"
 import * as Log from "@opencode-ai/core/util/log"
-import { Effect, Schema } from "effect"
+import { Effect, Queue, Schema } from "effect"
 import * as Stream from "effect/Stream"
 import { HttpServerResponse } from "effect/unstable/http"
 import { HttpApi, HttpApiBuilder, HttpApiEndpoint, HttpApiGroup, HttpApiSchema, OpenApi } from "effect/unstable/httpapi"
@@ -9,6 +9,12 @@ import { WorkspaceRoutingQuery } from "./middleware/workspace-routing"
 
 const log = Log.create({ service: "server" })
 
+type EventPayload = {
+  id: string
+  type: string
+  properties: Record<string, unknown>
+}
+
 export const EventPaths = {
   event: "/event",
 } as const
@@ -40,30 +46,39 @@ function eventData(data: unknown): Sse.Event {
 }
 
 function eventResponse(bus: Bus.Interface) {
-  const events = bus.subscribeAll().pipe(Stream.takeUntil((event) => event.type === Bus.InstanceDisposed.type))
-  const heartbeat = Stream.tick("10 seconds").pipe(
-    Stream.drop(1),
-    Stream.map(() => ({ id: Bus.createID(), type: "server.heartbeat", properties: {} })),
-  )
+  return Effect.gen(function* () {
+    const queue = yield* Queue.unbounded<EventPayload>()
+    const unsubscribe = yield* bus.subscribeAllCallback((event) => Queue.offerUnsafe(queue, event))
+    const events = Stream.fromQueue(queue).pipe(Stream.takeUntil((event) => event.type === Bus.InstanceDisposed.type))
+    const heartbeat = Stream.tick("10 seconds").pipe(
+      Stream.drop(1),
+      Stream.map(() => ({ id: Bus.createID(), type: "server.heartbeat", properties: {} })),
+    )
 
-  log.info("event connected")
-  return HttpServerResponse.stream(
-    Stream.make({ id: Bus.createID(), type: "server.connected", properties: {} }).pipe(
-      Stream.concat(events.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }))),
-      Stream.map(eventData),
-      Stream.pipeThroughChannel(Sse.encode()),
-      Stream.encodeText,
-      Stream.ensuring(Effect.sync(() => log.info("event disconnected"))),
-    ),
-    {
-      contentType: "text/event-stream",
-      headers: {
-        "Cache-Control": "no-cache, no-transform",
-        "X-Accel-Buffering": "no",
-        "X-Content-Type-Options": "nosniff",
+    log.info("event connected")
+    return HttpServerResponse.stream(
+      Stream.make({ id: Bus.createID(), type: "server.connected", properties: {} }).pipe(
+        Stream.concat(events.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }))),
+        Stream.map(eventData),
+        Stream.pipeThroughChannel(Sse.encode()),
+        Stream.encodeText,
+        Stream.ensuring(
+          Effect.sync(unsubscribe).pipe(
+            Effect.andThen(Queue.shutdown(queue)),
+            Effect.andThen(Effect.sync(() => log.info("event disconnected"))),
+          ),
+        ),
+      ),
+      {
+        contentType: "text/event-stream",
+        headers: {
+          "Cache-Control": "no-cache, no-transform",
+          "X-Accel-Buffering": "no",
+          "X-Content-Type-Options": "nosniff",
+        },
       },
-    },
-  )
+    )
+  })
 }
 
 export const eventHandlers = HttpApiBuilder.group(EventApi, "event", (handlers) =>
@@ -72,7 +87,7 @@ export const eventHandlers = HttpApiBuilder.group(EventApi, "event", (handlers)
     return handlers.handleRaw(
       "subscribe",
       Effect.fn("EventHttpApi.subscribe")(function* () {
-        return eventResponse(bus)
+        return yield* eventResponse(bus)
       }),
     )
   }),