fix(core): add dispose functions to prevent subscription memory leaks

packages/opencode/src/bus/index.ts

@@ -102,4 +102,20 @@ export namespace Bus {
       match.splice(index, 1)
     }
   }
+
+  /** @internal Test helper to get subscription count for a specific event type */
+  export function _getSubscriptionCount(type: string): number {
+    const match = state().subscriptions.get(type)
+    return match?.length ?? 0
+  }
+
+  /** @internal Test helper to get total subscription count across all event types */
+  export function _getTotalSubscriptionCount(): number {
+    let total = 0
+    // Use Array.from to snapshot values in case of concurrent modification
+    for (const subs of Array.from(state().subscriptions.values())) {
+      total += subs.length
+    }
+    return total
+  }
 }

packages/opencode/src/format/index.ts

@@ -100,9 +100,14 @@ export namespace Format {
     return result
   }
 
+  // Store unsubscribe functions for cleanup
+  const unsubscribers: Array<() => void> = []
+
   export function init() {
     log.info("init")
-    Bus.subscribe(File.Event.Edited, async (payload) => {
+    // Clean up any existing subscriptions before adding new ones
+    dispose()
+    const unsub = Bus.subscribe(File.Event.Edited, async (payload) => {
       const file = payload.properties.file
       log.info("formatting", { file })
       const ext = path.extname(file)
@@ -133,5 +138,18 @@ export namespace Format {
         }
       }
     })
+    unsubscribers.push(unsub)
+  }
+
+  export function dispose() {
+    const toUnsubscribe = unsubscribers.splice(0)
+    for (const unsub of toUnsubscribe) {
+      try {
+        unsub()
+      } catch (error) {
+        log.error("failed to unsubscribe format handler", { error })
+      }
+    }
+    log.info("disposed format subscriptions")
   }
 }

packages/opencode/src/plugin/index.ts

@@ -99,20 +99,38 @@ export namespace Plugin {
     return state().then((x) => x.hooks)
   }
 
+  // Store unsubscribe functions for cleanup
+  const unsubscribers: Array<() => void> = []
+
   export async function init() {
+    // Clean up any existing subscriptions before adding new ones
+    dispose()
     const hooks = await state().then((x) => x.hooks)
     const config = await Config.get()
     for (const hook of hooks) {
       // @ts-expect-error this is because we haven't moved plugin to sdk v2
       await hook.config?.(config)
     }
-    Bus.subscribeAll(async (input) => {
+    const unsub = Bus.subscribeAll(async (input) => {
       const hooks = await state().then((x) => x.hooks)
       for (const hook of hooks) {
         hook["event"]?.({
           event: input,
         })
       }
     })
+    unsubscribers.push(unsub)
+  }
+
+  export function dispose() {
+    const toUnsubscribe = unsubscribers.splice(0)
+    for (const unsub of toUnsubscribe) {
+      try {
+        unsub()
+      } catch (error) {
+        log.error("failed to unsubscribe plugin handler", { error })
+      }
+    }
+    log.info("disposed plugin subscriptions")
   }
 }

packages/opencode/src/share/share-next.ts

@@ -10,21 +10,31 @@ import type * as SDK from "@opencode-ai/sdk/v2"
 
 export namespace ShareNext {
   const log = Log.create({ service: "share-next" })
+  let disposed = false
+
+  // Store unsubscribe functions for cleanup
+  const unsubscribers: Array<() => void> = []
 
   async function url() {
     return Config.get().then((x) => x.enterprise?.url ?? "https://opncd.ai")
   }
 
   export async function init() {
-    Bus.subscribe(Session.Event.Updated, async (evt) => {
+    // Clean up any existing subscriptions before adding new ones
+    dispose()
+    disposed = false
+
+    const unsub1 = Bus.subscribe(Session.Event.Updated, async (evt) => {
+      if (disposed) return
       await sync(evt.properties.info.id, [
         {
           type: "session",
           data: evt.properties.info,
         },
       ])
     })
-    Bus.subscribe(MessageV2.Event.Updated, async (evt) => {
+    const unsub2 = Bus.subscribe(MessageV2.Event.Updated, async (evt) => {
+      if (disposed) return
       await sync(evt.properties.info.sessionID, [
         {
           type: "message",
@@ -44,22 +54,44 @@ export namespace ShareNext {
         ])
       }
     })
-    Bus.subscribe(MessageV2.Event.PartUpdated, async (evt) => {
+    const unsub3 = Bus.subscribe(MessageV2.Event.PartUpdated, async (evt) => {
+      if (disposed) return
       await sync(evt.properties.part.sessionID, [
         {
           type: "part",
           data: evt.properties.part,
         },
       ])
     })
-    Bus.subscribe(Session.Event.Diff, async (evt) => {
+    const unsub4 = Bus.subscribe(Session.Event.Diff, async (evt) => {
+      if (disposed) return
       await sync(evt.properties.sessionID, [
         {
           type: "session_diff",
           data: evt.properties.diff,
         },
       ])
     })
+    unsubscribers.push(unsub1, unsub2, unsub3, unsub4)
+  }
+
+  export function dispose() {
+    disposed = true
+    const toUnsubscribe = unsubscribers.splice(0)
+    for (const unsub of toUnsubscribe) {
+      try {
+        unsub()
+      } catch (error) {
+        log.error("failed to unsubscribe", { error })
+      }
+    }
+    // Hardened: snapshot and clear atomically to avoid race during iteration
+    const pending = Array.from(queue.values())
+    queue.clear()
+    for (const entry of pending) {
+      clearTimeout(entry.timeout)
+    }
+    log.info("disposed share-next subscriptions")
   }
 
   export async function create(sessionID: string) {
@@ -191,4 +223,17 @@ export namespace ShareNext {
       },
     ])
   }
+
+  /** @internal Test helper to get queue size */
+  export function _getQueueSize(): number {
+    return queue.size
+  }
+
+  /** @internal Test helper to add items to queue for testing dispose cleanup */
+  export function _addToQueueForTesting(sessionID: string) {
+    const dataMap = new Map<string, Data>()
+    // Use short timeout for tests - this is a no-op callback that won't cause issues if it fires
+    const timeout = setTimeout(() => {}, 100)
+    queue.set(sessionID, { timeout, data: dataMap })
+  }
 }

packages/opencode/src/share/share.ts

@@ -9,21 +9,33 @@ export namespace Share {
 
   let queue: Promise<void> = Promise.resolve()
   const pending = new Map<string, any>()
+  let disposed = false
+
+  // Store unsubscribe functions for cleanup
+  const unsubscribers: Array<() => void> = []
 
   export async function sync(key: string, content: any) {
+    // Skip if disposed - check at entry point
+    if (disposed) return
     const [root, ...splits] = key.split("/")
     if (root !== "session") return
     const [sub, sessionID] = splits
     if (sub === "share") return
     const share = await Session.getShare(sessionID).catch(() => {})
     if (!share) return
+    // Re-check disposed after async operation
+    if (disposed) return
     const { secret } = share
     pending.set(key, content)
     queue = queue
       .then(async () => {
+        // Check disposed at start of queued operation
+        if (disposed) return
         const content = pending.get(key)
         if (content === undefined) return
         pending.delete(key)
+        // Final check before network request
+        if (disposed) return
 
         return fetch(`${URL}/share_sync`, {
           method: "POST",
@@ -46,13 +58,17 @@ export namespace Share {
   }
 
   export function init() {
-    Bus.subscribe(Session.Event.Updated, async (evt) => {
+    // Clean up any existing subscriptions before adding new ones
+    dispose()
+    disposed = false
+
+    const unsub1 = Bus.subscribe(Session.Event.Updated, async (evt) => {
       await sync("session/info/" + evt.properties.info.id, evt.properties.info)
     })
-    Bus.subscribe(MessageV2.Event.Updated, async (evt) => {
+    const unsub2 = Bus.subscribe(MessageV2.Event.Updated, async (evt) => {
       await sync("session/message/" + evt.properties.info.sessionID + "/" + evt.properties.info.id, evt.properties.info)
     })
-    Bus.subscribe(MessageV2.Event.PartUpdated, async (evt) => {
+    const unsub3 = Bus.subscribe(MessageV2.Event.PartUpdated, async (evt) => {
       await sync(
         "session/part/" +
           evt.properties.part.sessionID +
@@ -63,6 +79,22 @@ export namespace Share {
         evt.properties.part,
       )
     })
+    unsubscribers.push(unsub1, unsub2, unsub3)
+  }
+
+  export function dispose() {
+    disposed = true
+    const toUnsubscribe = unsubscribers.splice(0)
+    for (const unsub of toUnsubscribe) {
+      try {
+        unsub()
+      } catch (error) {
+        log.error("failed to unsubscribe", { error })
+      }
+    }
+    pending.clear()
+    queue = Promise.resolve()
+    log.info("disposed share subscriptions")
   }
 
   export const URL =