fix: resolve multiple memory leaks in bus subscriptions, compaction, and state cleanup

- Return unsub from Bus.once() so callers can clean up
- Store and call unsubscribe handles in Share, ShareNext, Format, Plugin, Bootstrap
- Return unsub from subscribeSessionEvents in github command and call it in finally block
- Actually clear tool output and attachments during compaction instead of just marking timestamp
- Add dispose callback to FileTime state so it gets cleaned on instance disposal
- Clear ShareNext queue timeouts on dispose and catch failed sync fetches

Co-Authored-By: Claude Opus 4.6 <[email protected]>

Author: sebishogun

packages/opencode/src/bus/index.ts

@@ -80,6 +80,7 @@ export namespace Bus {
     const unsub = subscribe(def, (event) => {
       if (callback(event)) unsub()
     })
+    return unsub
   }
 
   export function subscribeAll(callback: (event: any) => void) {

packages/opencode/src/cli/cmd/github.ts

@@ -468,6 +468,7 @@ export const GithubRunCommand = cmd({
       let session: { id: string; title: string; version: string }
       let shareId: string | undefined
       let exitCode = 0
+      let unsubSessionEvents: (() => void) | undefined
       type PromptFiles = Awaited<ReturnType<typeof getUserPrompt>>["promptFiles"]
       const triggerCommentId = isCommentEvent
         ? (payload as IssueCommentEvent | PullRequestReviewCommentEvent).comment.id
@@ -518,7 +519,7 @@ export const GithubRunCommand = cmd({
             },
           ],
         })
-        subscribeSessionEvents()
+        unsubSessionEvents = subscribeSessionEvents()
         shareId = await (async () => {
           if (share === false) return
           if (!share && repoData.data.private) return
@@ -633,6 +634,7 @@ export const GithubRunCommand = cmd({
         // Also output the clean error message for the action to capture
         //core.setOutput("prepare_error", e.message);
       } finally {
+        unsubSessionEvents?.()
         if (!useGithubToken) {
           await restoreGitConfig()
           await revokeAppToken()
@@ -829,7 +831,7 @@ export const GithubRunCommand = cmd({
         }
 
         let text = ""
-        Bus.subscribe(MessageV2.Event.PartUpdated, async (evt) => {
+        return Bus.subscribe(MessageV2.Event.PartUpdated, async (evt) => {
           if (evt.properties.part.sessionID !== session.id) return
           //if (evt.properties.part.messageID === messageID) return
           const part = evt.properties.part

packages/opencode/src/file/time.ts

@@ -8,18 +8,26 @@ export namespace FileTime {
   // All tools that overwrite existing files should run their
   // assert/read/write/update sequence inside withLock(filepath, ...)
   // so concurrent writes to the same file are serialized.
-  export const state = Instance.state(() => {
-    const read: {
-      [sessionID: string]: {
-        [path: string]: Date | undefined
+  export const state = Instance.state(
+    () => {
+      const read: {
+        [sessionID: string]: {
+          [path: string]: Date | undefined
+        }
+      } = {}
+      const locks = new Map<string, Promise<void>>()
+      return {
+        read,
+        locks,
       }
-    } = {}
-    const locks = new Map<string, Promise<void>>()
-    return {
-      read,
-      locks,
-    }
-  })
+    },
+    async (current) => {
+      for (const key of Object.keys(current.read)) {
+        delete current.read[key]
+      }
+      current.locks.clear()
+    },
+  )
 
   export function read(sessionID: string, file: string) {
     log.info("read", { sessionID, file })

packages/opencode/src/format/index.ts

@@ -100,9 +100,12 @@ export namespace Format {
     return result
   }
 
+  let unsub: (() => void) | undefined
+
   export function init() {
     log.info("init")
-    Bus.subscribe(File.Event.Edited, async (payload) => {
+    unsub?.()
+    unsub = Bus.subscribe(File.Event.Edited, async (payload) => {
       const file = payload.properties.file
       log.info("formatting", { file })
       const ext = path.extname(file)

packages/opencode/src/plugin/index.ts

@@ -119,14 +119,17 @@ export namespace Plugin {
     return state().then((x) => x.hooks)
   }
 
+  let unsub: (() => void) | undefined
+
   export async function init() {
     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) => {
+    unsub?.()
+    unsub = Bus.subscribeAll(async (input) => {
       const hooks = await state().then((x) => x.hooks)
       for (const hook of hooks) {
         hook["event"]?.({

packages/opencode/src/project/bootstrap.ts

@@ -13,6 +13,8 @@ import { ShareNext } from "@/share/share-next"
 import { Snapshot } from "../snapshot"
 import { Truncate } from "../tool/truncation"
 
+let unsub: (() => void) | undefined
+
 export async function InstanceBootstrap() {
   Log.Default.info("bootstrapping", { directory: Instance.directory })
   await Plugin.init()
@@ -25,7 +27,8 @@ export async function InstanceBootstrap() {
   Snapshot.init()
   Truncate.init()
 
-  Bus.subscribe(Command.Event.Executed, async (payload) => {
+  unsub?.()
+  unsub = Bus.subscribe(Command.Event.Executed, async (payload) => {
     if (payload.properties.name === Command.Default.INIT) {
       await Project.setInitialized(Instance.project.id)
     }

packages/opencode/src/session/compaction.ts

@@ -91,6 +91,8 @@ export namespace SessionCompaction {
       for (const part of toPrune) {
         if (part.state.status === "completed") {
           part.state.time.compacted = Date.now()
+          part.state.output = "[Old tool result content cleared]"
+          part.state.attachments = []
           await Session.updatePart(part)
         }
       }

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

@@ -18,52 +18,64 @@ export namespace ShareNext {
 
   const disabled = process.env["OPENCODE_DISABLE_SHARE"] === "true" || process.env["OPENCODE_DISABLE_SHARE"] === "1"
 
+  let unsubs: (() => void)[] = []
+
   export async function init() {
     if (disabled) return
-    Bus.subscribe(Session.Event.Updated, async (evt) => {
-      await sync(evt.properties.info.id, [
-        {
-          type: "session",
-          data: evt.properties.info,
-        },
-      ])
-    })
-    Bus.subscribe(MessageV2.Event.Updated, async (evt) => {
-      await sync(evt.properties.info.sessionID, [
-        {
-          type: "message",
-          data: evt.properties.info,
-        },
-      ])
-      if (evt.properties.info.role === "user") {
+    dispose()
+    unsubs.push(
+      Bus.subscribe(Session.Event.Updated, async (evt) => {
+        await sync(evt.properties.info.id, [
+          {
+            type: "session",
+            data: evt.properties.info,
+          },
+        ])
+      }),
+      Bus.subscribe(MessageV2.Event.Updated, async (evt) => {
         await sync(evt.properties.info.sessionID, [
           {
-            type: "model",
-            data: [
-              await Provider.getModel(evt.properties.info.model.providerID, evt.properties.info.model.modelID).then(
-                (m) => m,
-              ),
-            ],
+            type: "message",
+            data: evt.properties.info,
           },
         ])
-      }
-    })
-    Bus.subscribe(MessageV2.Event.PartUpdated, async (evt) => {
-      await sync(evt.properties.part.sessionID, [
-        {
-          type: "part",
-          data: evt.properties.part,
-        },
-      ])
-    })
-    Bus.subscribe(Session.Event.Diff, async (evt) => {
-      await sync(evt.properties.sessionID, [
-        {
-          type: "session_diff",
-          data: evt.properties.diff,
-        },
-      ])
-    })
+        if (evt.properties.info.role === "user") {
+          await sync(evt.properties.info.sessionID, [
+            {
+              type: "model",
+              data: [
+                await Provider.getModel(evt.properties.info.model.providerID, evt.properties.info.model.modelID).then(
+                  (m) => m,
+                ),
+              ],
+            },
+          ])
+        }
+      }),
+      Bus.subscribe(MessageV2.Event.PartUpdated, async (evt) => {
+        await sync(evt.properties.part.sessionID, [
+          {
+            type: "part",
+            data: evt.properties.part,
+          },
+        ])
+      }),
+      Bus.subscribe(Session.Event.Diff, async (evt) => {
+        await sync(evt.properties.sessionID, [
+          {
+            type: "session_diff",
+            data: evt.properties.diff,
+          },
+        ])
+      }),
+    )
+  }
+
+  export function dispose() {
+    for (const unsub of unsubs) unsub()
+    unsubs = []
+    for (const [, entry] of queue) clearTimeout(entry.timeout)
+    queue.clear()
   }
 
   export async function create(sessionID: string) {
@@ -154,7 +166,7 @@ export namespace ShareNext {
           secret: share.secret,
           data: Array.from(queued.data.values()),
         }),
-      })
+      }).catch(() => {})
     }, 1000)
     queue.set(sessionID, { timeout, data: dataMap })
   }