core: close SSE stream when instance is disposed

2 files changed, 34 insertions, 9 deletions

diff --git a/packages/opencode/src/bus/index.ts b/packages/opencode/src/bus/index.ts
index aace87585..8a10c75d0 100644
--- a/packages/opencode/src/bus/index.ts
+++ b/packages/opencode/src/bus/index.ts
@@ -7,19 +7,35 @@ import { GlobalBus } from "./global"
 export namespace Bus {
   const log = Log.create({ service: "bus" })
   type Subscription = (event: any) => void
-
-  const state = Instance.state(() => {
-    const subscriptions = new Map<any, Subscription[]>()
-
-    return {
-      subscriptions,
-    }
-  })
+  const disposedEventType = "server.instance.disposed"
 
   export type EventDefinition = ReturnType<typeof event>
 
   const registry = new Map<string, EventDefinition>()
 
+  const state = Instance.state(
+    () => {
+      const subscriptions = new Map<any, Subscription[]>()
+
+      return {
+        subscriptions,
+      }
+    },
+    async (entry) => {
+      const wildcard = entry.subscriptions.get("*")
+      if (!wildcard) return
+      const event = {
+        type: disposedEventType,
+        properties: {
+          directory: Instance.directory,
+        },
+      }
+      for (const sub of [...wildcard]) {
+        sub(event)
+      }
+    },
+  )
+
   export function event<Type extends string, Properties extends ZodType>(type: Type, properties: Properties) {
     const result = {
       type,
@@ -29,6 +45,13 @@ export namespace Bus {
     return result
   }
 
+  export const InstanceDisposed = event(
+    disposedEventType,
+    z.object({
+      directory: z.string(),
+    }),
+  )
+
   export function payloads() {
     return z
       .discriminatedUnion(
diff --git a/packages/opencode/src/server/server.ts b/packages/opencode/src/server/server.ts
index 7743e3dbb..fe4ad195a 100644
--- a/packages/opencode/src/server/server.ts
+++ b/packages/opencode/src/server/server.ts
@@ -43,7 +43,6 @@ import { Snapshot } from "@/snapshot"
 import { SessionSummary } from "@/session/summary"
 import { GlobalBus } from "@/bus/global"
 import { SessionStatus } from "@/session/status"
-import { ShareNext } from "@/share/share-next"
 
 // @ts-ignore This global is needed to prevent ai-sdk from logging warnings to stdout https://github.com/vercel/ai/blob/2dc67e0ef538307f21368db32d5a12345d98831b/packages/ai/src/logger/log-warnings.ts#L85
 globalThis.AI_SDK_LOG_WARNINGS = false
@@ -2037,6 +2036,9 @@ export namespace Server {
               await stream.writeSSE({
                 data: JSON.stringify(event),
               })
+              if (event.type === Bus.InstanceDisposed.type) {
+                stream.close()
+              }
             })
             await new Promise<void>((resolve) => {
               stream.onAbort(() => {