fix: additional memory leak fixes from upstream PR anomalyco#10914

- AsyncQueue: add close()/drain() methods, iterator exits on close
- Bash tool: use Buffer[] ring buffer with 10MB cap instead of unbounded concat
- LSP client: clear diagnostics map on shutdown, cap at 5000 entries
- Bus: clear subscriptions map on instance dispose
- PTY: use Buffer[] instead of string concat for session buffer

Author: kiennq

packages/opencode/src/bus/index.ts

@@ -25,16 +25,18 @@ export namespace Bus {
     },
     async (entry) => {
       const wildcard = entry.subscriptions.get("*")
-      if (!wildcard) return
-      const event = {
-        type: InstanceDisposed.type,
-        properties: {
-          directory: Instance.directory,
-        },
-      }
-      for (const sub of [...wildcard]) {
-        sub(event)
+      if (wildcard) {
+        const event = {
+          type: InstanceDisposed.type,
+          properties: {
+            directory: Instance.directory,
+          },
+        }
+        for (const sub of [...wildcard]) {
+          sub(event)
+        }
       }
+      entry.subscriptions.clear()
     },
   )
 

packages/opencode/src/lsp/client.ts

@@ -48,6 +48,7 @@ export namespace LSPClient {
       new StreamMessageWriter(input.server.process.stdin as any),
     )
 
+    const MAX_DIAGNOSTICS_FILES = 5000
     const diagnostics = new Map<string, Diagnostic[]>()
     connection.onNotification("textDocument/publishDiagnostics", (params) => {
       const filePath = Filesystem.normalizePath(fileURLToPath(params.uri))
@@ -56,6 +57,11 @@ export namespace LSPClient {
         count: params.diagnostics.length,
       })
       const exists = diagnostics.has(filePath)
+      // Evict oldest entries if map grows too large
+      if (!exists && diagnostics.size >= MAX_DIAGNOSTICS_FILES) {
+        const oldest = diagnostics.keys().next().value
+        if (oldest !== undefined) diagnostics.delete(oldest)
+      }
       diagnostics.set(filePath, params.diagnostics)
       if (!exists && input.serverID === "typescript") return
       Bus.publish(Event.Diagnostics, { path: filePath, serverID: input.serverID })
@@ -240,6 +246,7 @@ export namespace LSPClient {
         files.clear()
         connection.end()
         connection.dispose()
+        diagnostics.clear()
         input.server.process.kill()
         l.info("shutdown")
       },

packages/opencode/src/pty/index.ts

@@ -67,7 +67,8 @@ export namespace Pty {
   interface ActiveSession {
     info: Info
     process: IPty
-    buffer: string
+    bufferChunks: Buffer[]
+    bufferSize: number
     subscribers: Set<WSContext>
   }
 
@@ -138,7 +139,8 @@ export namespace Pty {
     const session: ActiveSession = {
       info,
       process: ptyProcess,
-      buffer: "",
+      bufferChunks: [],
+      bufferSize: 0,
       subscribers: new Set(),
     }
     state().set(id, session)
@@ -153,9 +155,14 @@ export namespace Pty {
         ws.send(data)
       }
       if (open) return
-      session.buffer += data
-      if (session.buffer.length <= BUFFER_LIMIT) return
-      session.buffer = session.buffer.slice(-BUFFER_LIMIT)
+      const chunk = Buffer.from(data)
+      session.bufferChunks.push(chunk)
+      session.bufferSize += chunk.length
+      // Trim oldest chunks when exceeding limit
+      while (session.bufferSize > BUFFER_LIMIT && session.bufferChunks.length > 1) {
+        const dropped = session.bufferChunks.shift()!
+        session.bufferSize -= dropped.length
+      }
     })
     ptyProcess.onExit(({ exitCode }) => {
       log.info("session exited", { id, exitCode })
@@ -223,16 +230,19 @@ export namespace Pty {
     }
     log.info("client connected to session", { id })
     session.subscribers.add(ws)
-    if (session.buffer) {
-      const buffer = session.buffer.length <= BUFFER_LIMIT ? session.buffer : session.buffer.slice(-BUFFER_LIMIT)
-      session.buffer = ""
+    if (session.bufferChunks.length > 0) {
+      const buffer = Buffer.concat(session.bufferChunks).toString()
+      session.bufferChunks.length = 0
+      session.bufferSize = 0
       try {
         for (let i = 0; i < buffer.length; i += BUFFER_CHUNK) {
           ws.send(buffer.slice(i, i + BUFFER_CHUNK))
         }
       } catch {
         session.subscribers.delete(ws)
-        session.buffer = buffer
+        const chunk = Buffer.from(buffer)
+        session.bufferChunks.push(chunk)
+        session.bufferSize = chunk.length
         ws.close()
         return
       }

packages/opencode/src/tool/bash.ts

@@ -175,7 +175,9 @@ export const BashTool = Tool.define("bash", async () => {
         detached: process.platform !== "win32",
       })
 
-      const chunks: Buffer[] = []
+      const MAX_OUTPUT_BYTES = 10 * 1024 * 1024 // 10MB
+      const outputChunks: Buffer[] = []
+      let outputSize = 0
 
       // Initialize metadata with empty output
       ctx.metadata({
@@ -186,12 +188,17 @@ export const BashTool = Tool.define("bash", async () => {
       })
 
       const append = (chunk: Buffer) => {
-        chunks.push(chunk)
-        const output = Buffer.concat(chunks).toString()
+        outputChunks.push(chunk)
+        outputSize += chunk.length
+        // Ring buffer: drop earliest chunks when exceeding limit
+        while (outputSize > MAX_OUTPUT_BYTES && outputChunks.length > 1) {
+          const dropped = outputChunks.shift()!
+          outputSize -= dropped.length
+        }
+        const preview = Buffer.concat(outputChunks).toString("utf-8")
         ctx.metadata({
           metadata: {
-            // truncate the metadata to avoid GIANT blobs of data (has nothing to do w/ what agent can access)
-            output: output.length > MAX_METADATA_LENGTH ? output.slice(0, MAX_METADATA_LENGTH) + "\n\n..." : output,
+            output: preview.length > MAX_METADATA_LENGTH ? preview.slice(0, MAX_METADATA_LENGTH) + "\n\n..." : preview,
             description: params.description,
           },
         })
@@ -242,6 +249,10 @@ export const BashTool = Tool.define("bash", async () => {
         })
       })
 
+      let output = Buffer.concat(outputChunks).toString("utf-8")
+      outputChunks.length = 0
+      outputSize = 0
+
       const resultMetadata: string[] = []
 
       if (timedOut) {
@@ -252,7 +263,6 @@ export const BashTool = Tool.define("bash", async () => {
         resultMetadata.push("User aborted the command")
       }
 
-      let output = Buffer.concat(chunks).toString()
       if (resultMetadata.length > 0) {
         output += "\n\n<bash_metadata>\n" + resultMetadata.join("\n") + "\n</bash_metadata>"
       }

packages/opencode/src/util/queue.ts

@@ -1,20 +1,40 @@
 export class AsyncQueue<T> implements AsyncIterable<T> {
   private queue: T[] = []
-  private resolvers: ((value: T) => void)[] = []
+  private resolvers: ((value: T | undefined) => void)[] = []
+  private closed = false
 
   push(item: T) {
+    if (this.closed) return
     const resolve = this.resolvers.shift()
     if (resolve) resolve(item)
     else this.queue.push(item)
   }
 
-  async next(): Promise<T> {
+  async next(): Promise<T | undefined> {
     if (this.queue.length > 0) return this.queue.shift()!
+    if (this.closed) return undefined
     return new Promise((resolve) => this.resolvers.push(resolve))
   }
 
+  close() {
+    this.closed = true
+    for (const resolve of this.resolvers) {
+      resolve(undefined)
+    }
+    this.resolvers.length = 0
+  }
+
+  drain() {
+    this.close()
+    this.queue.length = 0
+  }
+
   async *[Symbol.asyncIterator]() {
-    while (true) yield await this.next()
+    while (true) {
+      const value = await this.next()
+      if (value === undefined) return
+      yield value
+    }
   }
 }