fix: consolidate memory leak fixes from 23+ community PRs

binarydoubling · claude · rubend056 · commit 191dd2fbddd3 · 2026-03-18T08:13:55.000-04:00
Addresses the remaining memory leaks identified in anomalyco#16697 by consolidating the best fixes from 23+ open community PRs into a single coherent changeset. Fixes consolidated from PRs: anomalyco#16695, anomalyco#16346, anomalyco#14650, anomalyco#15646, - Plugin subscriber stacking: unsub before re-subscribing in init() - Subagent deallocation: Session.remove() after task completion - SSE stream cleanup: centralized cleanup with done guard (3 endpoints) - Compaction data trimming: clear output/attachments on prune - Process exit cleanup: Instance.disposeAll() with 5s timeout - Serve cmd: graceful shutdown instead of blocking forever - Bash tool: ring buffer with 10MB cap instead of O(n²) concat - LSP index teardown: clear clients/broken/spawning on dispose - LSP open-files cap: evict oldest when >1000 tracked files - Format subscription: store and cleanup unsub handle - Permission/Question clearSession: reject pending on session delete - Session.remove() cleanup chain: FileTime, Permission, Question - ShareNext subscription cleanup: store unsub handles, cleanup on dispose - OAuth transport: close existing before replacing Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
diff --git a/packages/opencode/src/cli/cmd/serve.ts b/packages/opencode/src/cli/cmd/serve.ts
@@ -5,6 +5,7 @@ import { Flag } from "../../flag/flag"
 import { Workspace } from "../../control-plane/workspace"
 import { Project } from "../../project/project"
 import { Installation } from "../../installation"
+import { Instance } from "../../project/instance"
 
 export const ServeCommand = cmd({
   command: "serve",
@@ -18,7 +19,13 @@ export const ServeCommand = cmd({
     const server = Server.listen(opts)
     console.log(`opencode server listening on http://${server.hostname}:${server.port}`)
 
-    await new Promise(() => {})
+    // Wait for termination signal instead of blocking forever
+    await new Promise<void>((resolve) => {
+      const shutdown = () => resolve()
+      process.on("SIGTERM", shutdown)
+      process.on("SIGINT", shutdown)
+    })
+    await Instance.disposeAll()
     await server.stop()
   },
 })
diff --git a/packages/opencode/src/control-plane/workspace-server/routes.ts b/packages/opencode/src/control-plane/workspace-server/routes.ts
@@ -7,10 +7,25 @@ export function WorkspaceServerRoutes() {
     c.header("X-Accel-Buffering", "no")
     c.header("X-Content-Type-Options", "nosniff")
     return streamSSE(c, async (stream) => {
+      let done = false
+      let resolveStream: (() => void) | undefined
+
+      const cleanup = () => {
+        if (done) return
+        done = true
+        clearInterval(heartbeat)
+        GlobalBus.off("event", handler)
+        resolveStream?.()
+      }
+
       const send = async (event: unknown) => {
-        await stream.writeSSE({
-          data: JSON.stringify(event),
-        })
+        try {
+          await stream.writeSSE({
+            data: JSON.stringify(event),
+          })
+        } catch {
+          cleanup()
+        }
       }
       const handler = async (event: { directory?: string; payload: unknown }) => {
         await send(event.payload)
@@ -22,11 +37,8 @@ export function WorkspaceServerRoutes() {
       }, 10_000)
 
       await new Promise<void>((resolve) => {
-        stream.onAbort(() => {
-          clearInterval(heartbeat)
-          GlobalBus.off("event", handler)
-          resolve()
-        })
+        resolveStream = resolve
+        stream.onAbort(cleanup)
       })
     })
   })
diff --git a/packages/opencode/src/index.ts b/packages/opencode/src/index.ts
@@ -205,6 +205,10 @@ try {
   }
   process.exitCode = 1
 } finally {
+  // Dispose all instances (LSP, MCP, PTY child processes) to prevent zombies.
+  // Race with a 5-second timeout so we don't hang on unresponsive subprocesses.
+  const { Instance } = await import("./project/instance")
+  await Promise.race([Instance.disposeAll(), new Promise((r) => setTimeout(r, 5000))]).catch(() => {})
   // Some subprocesses don't react properly to SIGTERM and similar signals.
   // Most notably, some docker-container-based MCP servers don't handle such signals unless
   // run using `docker run --init`.
diff --git a/packages/opencode/src/lsp/client.ts b/packages/opencode/src/lsp/client.ts
@@ -156,6 +156,7 @@ export namespace LSPClient {
       })
     }
 
+    const MAX_OPEN_FILES = 1000
     const files: {
       [path: string]: number
     } = {}
@@ -224,6 +225,12 @@ export namespace LSPClient {
             },
           })
           files[input.path] = 0
+          // Evict oldest file if we exceed the limit
+          const keys = Object.keys(files)
+          if (keys.length > MAX_OPEN_FILES) {
+            const oldest = keys[0]
+            delete files[oldest]
+          }
           return
         },
       },
@@ -263,6 +270,7 @@ export namespace LSPClient {
         l.info("shutting down")
         diagnostics.clear()
         diagnosticOrder.length = 0
+        for (const key of Object.keys(files)) delete files[key]
         connection.end()
         connection.dispose()
         input.server.process.kill()
diff --git a/packages/opencode/src/lsp/index.ts b/packages/opencode/src/lsp/index.ts
@@ -141,6 +141,9 @@ export namespace LSP {
     },
     async (state) => {
       await Promise.all(state.clients.map((client) => client.shutdown()))
+      state.clients.length = 0
+      state.broken.clear()
+      state.spawning.clear()
     },
   )
 
diff --git a/packages/opencode/src/mcp/index.ts b/packages/opencode/src/mcp/index.ts
@@ -420,7 +420,9 @@ export namespace MCP {
                 duration: 8000,
               }).catch((e) => log.debug("failed to show toast", { error: e }))
             } else {
-              // Store transport for later finishAuth call
+              // Close any existing pending transport before storing the new one
+              const existing = pendingOAuthTransports.get(key)
+              if (existing) existing.close?.().catch(() => {})
               pendingOAuthTransports.set(key, transport)
               status = { status: "needs_auth" as const }
               // Show toast for needs_auth
@@ -942,6 +944,8 @@ export namespace MCP {
   export async function removeAuth(mcpName: string): Promise<void> {
     await McpAuth.remove(mcpName)
     McpOAuthCallback.cancelPending(mcpName)
+    const transport = pendingOAuthTransports.get(mcpName)
+    if (transport) transport.close?.().catch(() => {})
     pendingOAuthTransports.delete(mcpName)
     await McpAuth.clearOAuthState(mcpName)
     log.info("removed oauth credentials", { mcpName })
diff --git a/packages/opencode/src/plugin/index.ts b/packages/opencode/src/plugin/index.ts
@@ -130,14 +130,18 @@ 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) => {
+    // Unsubscribe previous wildcard subscriber to prevent stacking on re-init
+    unsub?.()
+    unsub = Bus.subscribeAll(async (input) => {
       const hooks = await state().then((x) => x.hooks)
       for (const hook of hooks) {
         hook["event"]?.({
diff --git a/packages/opencode/src/question/index.ts b/packages/opencode/src/question/index.ts
@@ -33,7 +33,28 @@ export namespace Question {
     return runPromiseInstance(S.QuestionService.use((service) => service.reject(requestID)))
   }
 
+<<<<<<< HEAD
   export async function list(): Promise<Request[]> {
     return runPromiseInstance(S.QuestionService.use((service) => service.list()))
   }
+  }
+
+  export async function clearSession(sessionID: string) {
+    const s = await state()
+    for (const [id, pending] of Object.entries(s.pending)) {
+      if (pending.info.sessionID === sessionID) {
+        delete s.pending[id]
+        Bus.publish(Event.Rejected, {
+          sessionID: pending.info.sessionID,
+          requestID: pending.info.id,
+        })
+        pending.reject(new RejectedError())
+      }
+    }
+  }
+
+  export async function list() {
+    return state().then((x) => Array.from(x.pending.values(), (x) => x.info))
+>>>>>>> 5124d2ab6 (fix: consolidate memory leak fixes from 23+ community PRs)
+  }
 }
diff --git a/packages/opencode/src/server/routes/global.ts b/packages/opencode/src/server/routes/global.ts
@@ -69,40 +69,51 @@ export const GlobalRoutes = lazy(() =>
         c.header("X-Accel-Buffering", "no")
         c.header("X-Content-Type-Options", "nosniff")
         return streamSSE(c, async (stream) => {
-          stream.writeSSE({
-            data: JSON.stringify({
-              payload: {
-                type: "server.connected",
-                properties: {},
-              },
-            }),
-          })
+          let done = false
+          let resolveStream: (() => void) | undefined
+
+          const cleanup = () => {
+            if (done) return
+            done = true
+            clearInterval(heartbeat)
+            GlobalBus.off("event", handler)
+            resolveStream?.()
+            log.info("global event disconnected")
+          }
+
+          const writeSSE = async (data: string) => {
+            try {
+              await stream.writeSSE({ data })
+            } catch {
+              cleanup()
+            }
+          }
+
+          await writeSSE(JSON.stringify({
+            payload: {
+              type: "server.connected",
+              properties: {},
+            },
+          }))
+
           async function handler(event: any) {
-            await stream.writeSSE({
-              data: JSON.stringify(event),
-            })
+            await writeSSE(JSON.stringify(event))
           }
           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: {},
-                },
-              }),
-            })
+            void writeSSE(JSON.stringify({
+              payload: {
+                type: "server.heartbeat",
+                properties: {},
+              },
+            }))
           }, 10_000)
 
           await new Promise<void>((resolve) => {
-            stream.onAbort(() => {
-              clearInterval(heartbeat)
-              GlobalBus.off("event", handler)
-              resolve()
-              log.info("global event disconnected")
-            })
+            resolveStream = resolve
+            stream.onAbort(cleanup)
           })
         })
       },
diff --git a/packages/opencode/src/server/server.ts b/packages/opencode/src/server/server.ts
@@ -514,42 +514,54 @@ export namespace Server {
             },
           },
         }),
-        async (c) => {
-          log.info("event connected")
-          c.header("X-Accel-Buffering", "no")
-          c.header("X-Content-Type-Options", "nosniff")
-          return streamSSE(c, async (stream) => {
-            stream.writeSSE({
-              data: JSON.stringify({
+          async (c) => {
+            log.info("event connected")
+            c.header("X-Accel-Buffering", "no")
+            c.header("X-Content-Type-Options", "nosniff")
+            return streamSSE(c, async (stream) => {
+              let done = false
+              let resolveStream: (() => void) | undefined
+
+              const cleanup = () => {
+                if (done) return
+                done = true
+                clearInterval(heartbeat)
+                unsub()
+                resolveStream?.()
+                log.info("event disconnected")
+              }
+
+              const writeSSE = async (data: string) => {
+                try {
+                  await stream.writeSSE({ data })
+                } catch {
+                  cleanup()
+                }
+              }
+
+              await writeSSE(JSON.stringify({
                 type: "server.connected",
                 properties: {},
-              }),
-            })
-            const unsub = Bus.subscribeAll(async (event) => {
-              await stream.writeSSE({
-                data: JSON.stringify(event),
+              }))
+
+              const unsub = Bus.subscribeAll(async (event) => {
+                await writeSSE(JSON.stringify(event))
+                if (event.type === Bus.InstanceDisposed.type) {
+                  stream.close()
+                }
               })
-              if (event.type === Bus.InstanceDisposed.type) {
-                stream.close()
-              }
-            })
 
-            // Send heartbeat every 10s to prevent stalled proxy streams.
-            const heartbeat = setInterval(() => {
-              stream.writeSSE({
-                data: JSON.stringify({
+              // Send heartbeat every 10s to prevent stalled proxy streams.
+              const heartbeat = setInterval(() => {
+                void writeSSE(JSON.stringify({
                   type: "server.heartbeat",
                   properties: {},
-                }),
-              })
-            }, 10_000)
+                }))
+              }, 10_000)
 
-            await new Promise<void>((resolve) => {
-              stream.onAbort(() => {
-                clearInterval(heartbeat)
-                unsub()
-                resolve()
-                log.info("event disconnected")
+              await new Promise<void>((resolve) => {
+                resolveStream = resolve
+                stream.onAbort(cleanup)
               })
             })
           })
diff --git a/packages/opencode/src/session/compaction.ts b/packages/opencode/src/session/compaction.ts
@@ -92,6 +92,10 @@ export namespace SessionCompaction {
       for (const part of toPrune) {
         if (part.state.status === "completed") {
           part.state.time.compacted = Date.now()
+          // Clear output and attachments to free memory — compacted parts
+          // are already replaced with placeholder text in toModelMessages
+          part.state.output = ""
+          part.state.attachments = undefined
           await Session.updatePart(part)
         }
       }
diff --git a/packages/opencode/src/session/index.ts b/packages/opencode/src/session/index.ts
@@ -29,6 +29,8 @@ import { SessionID, MessageID, PartID } from "./schema"
 import type { Provider } from "@/provider/provider"
 import { ModelID, ProviderID } from "@/provider/schema"
 import { PermissionNext } from "@/permission"
+import { Question } from "@/question"
+import { FileTime } from "@/file/time"
 import { Global } from "@/global"
 import type { LanguageModelV2Usage } from "@ai-sdk/provider"
 import { iife } from "@/util/iife"
@@ -668,6 +670,11 @@ export namespace Session {
       for (const child of await children(sessionID)) {
         await remove(child.id)
       }
+      // Clean up per-session state before deleting
+      await PermissionNext.clearSession(sessionID)
+      await Question.clearSession(sessionID)
+      const ft = FileTime.state()
+      delete ft.read[sessionID]
       await unshare(sessionID).catch(() => {})
       // CASCADE delete handles messages and parts automatically
       Database.use((db) => {
diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts
diff --git a/packages/opencode/src/tool/bash.ts b/packages/opencode/src/tool/bash.ts
diff --git a/packages/opencode/src/tool/task.ts b/packages/opencode/src/tool/task.ts

Original file line number	Diff line number	Diff line change
`@@ -141,6 +141,9 @@ export namespace LSP {`
`141`	`141`	`},`
`142`	`142`	`async (state) => {`
`143`	`143`	`await Promise.all(state.clients.map((client) => client.shutdown()))`
	`144`	`+ state.clients.length = 0`
	`145`	`+ state.broken.clear()`
	`146`	`+ state.spawning.clear()`
`144`	`147`	`},`
`145`	`148`	`)`
`146`	`149`