fix: prevent SSE stream memory leaks on client disconnect

- Add cleanup() guard with done flag to prevent double-cleanup
- await stream.writeSSE() calls and catch errors to trigger cleanup
- Unsubscribe Bus and GlobalBus listeners on abort or write failure
- Clear heartbeat interval in all exit paths
- Add Bus.debug() subscription count introspection
- Add GET /debug/memory endpoint for runtime memory diagnostics

Author: Brendan DeBeasi

packages/opencode/src/bus/index.ts

@@ -102,4 +102,12 @@ export namespace Bus {
       match.splice(index, 1)
     }
   }
+
+  export function debug() {
+    const counts: Record<string, number> = {}
+    for (const [type, subs] of state().subscriptions) {
+      counts[type] = subs.length
+    }
+    return { subscriptions: counts }
+  }
 }

packages/opencode/src/server/routes/global.ts

@@ -69,40 +69,59 @@ export const GlobalRoutes = lazy(() =>
         c.header("X-Accel-Buffering", "no")
         c.header("X-Content-Type-Options", "nosniff")
         return streamSSE(c, async (stream) => {
-          stream.writeSSE({
+          await stream.writeSSE({
             data: JSON.stringify({
               payload: {
                 type: "server.connected",
                 properties: {},
               },
             }),
           })
+
+          let done = false
+          let resolveWait: () => void = () => {}
+          let heartbeat: ReturnType<typeof setInterval> | undefined
+
+          const cleanup = () => {
+            if (done) return
+            done = true
+            clearInterval(heartbeat)
+            GlobalBus.off("event", handler)
+            resolveWait()
+            log.info("global event disconnected")
+          }
+
           async function handler(event: any) {
-            await stream.writeSSE({
-              data: JSON.stringify(event),
-            })
+            if (done) return
+            try {
+              await stream.writeSSE({ data: JSON.stringify(event) })
+            } catch {
+              cleanup()
+            }
           }
+
           GlobalBus.on("event", handler)
 
           // Send heartbeat every 10s to prevent stalled proxy streams.
-          const heartbeat = setInterval(() => {
-            stream.writeSSE({
-              data: JSON.stringify({
-                payload: {
-                  type: "server.heartbeat",
-                  properties: {},
-                },
-              }),
-            })
+          heartbeat = setInterval(async () => {
+            if (done) return
+            try {
+              await stream.writeSSE({
+                data: JSON.stringify({
+                  payload: {
+                    type: "server.heartbeat",
+                    properties: {},
+                  },
+                }),
+              })
+            } catch {
+              cleanup()
+            }
           }, 10_000)
 
           await new Promise<void>((resolve) => {
-            stream.onAbort(() => {
-              clearInterval(heartbeat)
-              GlobalBus.off("event", handler)
-              resolve()
-              log.info("global event disconnected")
-            })
+            resolveWait = resolve
+            stream.onAbort(cleanup)
           })
         })
       },

packages/opencode/src/server/server.ts

@@ -1,5 +1,6 @@
 import { BusEvent } from "@/bus/bus-event"
 import { Bus } from "@/bus"
+import { GlobalBus } from "@/bus/global"
 import { Log } from "../util/log"
 import { describeRoute, generateSpecs, validator, resolver, openAPIRouteHandler } from "hono-openapi"
 import { Hono } from "hono"
@@ -516,38 +517,58 @@ export namespace Server {
           c.header("X-Accel-Buffering", "no")
           c.header("X-Content-Type-Options", "nosniff")
           return streamSSE(c, async (stream) => {
-            stream.writeSSE({
+            await stream.writeSSE({
               data: JSON.stringify({
                 type: "server.connected",
                 properties: {},
               }),
             })
-            const unsub = Bus.subscribeAll(async (event) => {
-              await stream.writeSSE({
-                data: JSON.stringify(event),
-              })
-              if (event.type === Bus.InstanceDisposed.type) {
-                stream.close()
+
+            let done = false
+            let unsub: () => void = () => {}
+            let resolveWait: () => void = () => {}
+            let heartbeat: ReturnType<typeof setInterval> | undefined
+
+            const cleanup = () => {
+              if (done) return
+              done = true
+              clearInterval(heartbeat)
+              unsub()
+              resolveWait()
+              log.info("event disconnected")
+            }
+
+            unsub = Bus.subscribeAll(async (event) => {
+              if (done) return
+              try {
+                await stream.writeSSE({ data: JSON.stringify(event) })
+                if (event.type === Bus.InstanceDisposed.type) {
+                  cleanup()
+                  stream.close()
+                }
+              } catch {
+                cleanup()
               }
             })
 
             // Send heartbeat every 10s to prevent stalled proxy streams.
-            const heartbeat = setInterval(() => {
-              stream.writeSSE({
-                data: JSON.stringify({
-                  type: "server.heartbeat",
-                  properties: {},
-                }),
-              })
+            heartbeat = setInterval(async () => {
+              if (done) return
+              try {
+                await stream.writeSSE({
+                  data: JSON.stringify({
+                    type: "server.heartbeat",
+                    properties: {},
+                  }),
+                })
+              } catch {
+                cleanup()
+              }
             }, 10_000)
 
             await new Promise<void>((resolve) => {
-              stream.onAbort(() => {
-                clearInterval(heartbeat)
-                unsub()
-                resolve()
-                log.info("event disconnected")
-              })
+              resolveWait = resolve
+              stream.onAbort(cleanup)
             })
           })
         },