From 89c51a86bd96dd514c73ea26f69f59ce94ee1f36 Mon Sep 17 00:00:00 2001
From: Aiden Cline <aidenpcline@gmail.com>
Date: Wed, 13 May 2026 20:16:20 -0500
Subject: [PATCH] fix(httpapi): provide instance context to event stream
 subscription
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The /event SSE handler called bus.subscribeAll() which evaluates its
inner Effect (InstanceState.get → InstanceRef/Instance.current lookup)
inside the body-stream consumer fiber. That fiber does not carry the
request handler's ALS/Effect context, so the lookup failed and the
stream halted right after server.connected — no message deltas,
permission asks, or heartbeats could reach clients.

Capture InstanceState.context and InstanceState.workspaceID at handler
time and provide them to the subscription stream via Stream.provideService.

Fixes #27391.

Co-authored-by: James Long <longster@gmail.com>
---
 .../server/routes/instance/httpapi/event.ts   | 17 +++++++++--
 .../test/server/httpapi-event.test.ts         | 28 +++++++++++++++++++
 2 files changed, 42 insertions(+), 3 deletions(-)

diff --git a/packages/opencode/src/server/routes/instance/httpapi/event.ts b/packages/opencode/src/server/routes/instance/httpapi/event.ts
index 8113c76f51..7756cbf201 100644
--- a/packages/opencode/src/server/routes/instance/httpapi/event.ts
+++ b/packages/opencode/src/server/routes/instance/httpapi/event.ts
@@ -1,4 +1,8 @@
 import { Bus } from "@/bus"
+import type { WorkspaceID } from "@/control-plane/schema"
+import { InstanceRef, WorkspaceRef } from "@/effect/instance-ref"
+import { InstanceState } from "@/effect/instance-state"
+import type { InstanceContext } from "@/project/instance"
 import * as Log from "@opencode-ai/core/util/log"
 import { Effect, Schema } from "effect"
 import * as Stream from "effect/Stream"
@@ -39,8 +43,12 @@ function eventData(data: unknown): Sse.Event {
   }
 }
 
-function eventResponse(bus: Bus.Interface) {
-  const events = bus.subscribeAll().pipe(Stream.takeUntil((event) => event.type === Bus.InstanceDisposed.type))
+function eventResponse(bus: Bus.Interface, refs: { instance: InstanceContext; workspace?: WorkspaceID }) {
+  const events = bus.subscribeAll().pipe(
+    Stream.provideService(InstanceRef, refs.instance),
+    Stream.provideService(WorkspaceRef, refs.workspace),
+    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: {} })),
@@ -72,7 +80,10 @@ export const eventHandlers = HttpApiBuilder.group(EventApi, "event", (handlers)
     return handlers.handleRaw(
       "subscribe",
       Effect.fn("EventHttpApi.subscribe")(function* () {
-        return eventResponse(bus)
+        return eventResponse(bus, {
+          instance: yield* InstanceState.context,
+          workspace: yield* InstanceState.workspaceID,
+        })
       }),
     )
   }),
diff --git a/packages/opencode/test/server/httpapi-event.test.ts b/packages/opencode/test/server/httpapi-event.test.ts
index df716ed096..077dcb461d 100644
--- a/packages/opencode/test/server/httpapi-event.test.ts
+++ b/packages/opencode/test/server/httpapi-event.test.ts
@@ -31,6 +31,18 @@ async function readFirstEvent(response: Response) {
   }
 }
 
+async function readEvent(reader: ReadableStreamDefaultReader<Uint8Array>) {
+  const result = await Promise.race([
+    reader.read(),
+    new Promise<never>((_, reject) => setTimeout(() => reject(new Error("timed out waiting for event")), 5_000)),
+  ])
+  return JSON.parse(new TextDecoder().decode(result.value).replace(/^data: /, "")) as {
+    id?: string
+    type: string
+    properties: Record<string, unknown>
+  }
+}
+
 afterEach(async () => {
   await disposeAllInstances()
   await resetDatabase()
@@ -56,4 +68,20 @@ describe("event HttpApi", () => {
 
     expect(await readFirstEvent(response)).toMatchObject({ type: "server.connected", properties: {} })
   })
+
+  test("keeps the event stream open after the initial event", async () => {
+    await using tmp = await tmpdir({ git: true, config: { formatter: false, lsp: false } })
+    const response = await app().request(EventPaths.event, { headers: { "x-opencode-directory": tmp.path } })
+    if (!response.body) throw new Error("missing response body")
+
+    const reader = response.body.getReader()
+    expect(await readEvent(reader)).toMatchObject({ type: "server.connected", properties: {} })
+    const next = await Promise.race([
+      reader.read().then((result) => (result.done ? "closed" : "event")),
+      new Promise<"open">((resolve) => setTimeout(() => resolve("open"), 250)),
+    ])
+    await reader.cancel()
+
+    expect(next).toBe("open")
+  })
 })