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

- Add dispose() to Share, ShareNext, Plugin, and Format namespaces
- Add cleanupSession() and dispose() to ACP Agent with AbortControllers
- Add Bus._getSubscriptionCount() test helpers
- Add memory tests to verify cleanup works correctly

Supersedes anomalyco#7032
Fixes anomalyco#3013

Author: Mark Henderson

packages/opencode/src/acp/agent.ts

@@ -102,4 +102,19 @@ 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
+    for (const subs of state().subscriptions.values()) {
+      total += subs.length
+    }
+    return total
+  }
 }

packages/opencode/src/bus/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,14 @@ export namespace Format {
         }
       }
     })
+    unsubscribers.push(unsub)
+  }
+
+  export function dispose() {
+    for (const unsub of unsubscribers) {
+      unsub()
+    }
+    unsubscribers.length = 0
+    log.info("disposed format subscriptions")
   }
 }

packages/opencode/src/format/index.ts

@@ -99,20 +99,34 @@ 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() {
+    for (const unsub of unsubscribers) {
+      unsub()
+    }
+    unsubscribers.length = 0
+    log.info("disposed plugin subscriptions")
   }
 }

packages/opencode/src/plugin/index.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,39 @@ 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
+    for (const unsub of unsubscribers) {
+      unsub()
+    }
+    unsubscribers.length = 0
+    // Clear pending timeouts
+    for (const entry of queue.values()) {
+      clearTimeout(entry.timeout)
+    }
+    queue.clear()
+    log.info("disposed share-next subscriptions")
   }
 
   export async function create(sessionID: string) {
@@ -191,4 +218,16 @@ 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>()
+    const timeout = setTimeout(() => {}, 10000)
+    queue.set(sessionID, { timeout, data: dataMap })
+  }
 }

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

@@ -9,8 +9,14 @@ 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
+    if (disposed) return
     const [root, ...splits] = key.split("/")
     if (root !== "session") return
     const [sub, sessionID] = splits
@@ -21,6 +27,8 @@ export namespace Share {
     pending.set(key, content)
     queue = queue
       .then(async () => {
+        // Check if disposed before processing
+        if (disposed) return
         const content = pending.get(key)
         if (content === undefined) return
         pending.delete(key)
@@ -46,13 +54,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 +75,18 @@ export namespace Share {
         evt.properties.part,
       )
     })
+    unsubscribers.push(unsub1, unsub2, unsub3)
+  }
+
+  export function dispose() {
+    disposed = true
+    for (const unsub of unsubscribers) {
+      unsub()
+    }
+    unsubscribers.length = 0
+    pending.clear()
+    queue = Promise.resolve()
+    log.info("disposed share subscriptions")
   }
 
   export const URL =