fix(server): add backpressure to SSE event queues

packages/opencode/src/server/routes/event.ts

@@ -32,7 +32,7 @@ export const EventRoutes = () =>
       c.header("X-Accel-Buffering", "no")
       c.header("X-Content-Type-Options", "nosniff")
       return streamSSE(c, async (stream) => {
-        const q = new AsyncQueue<string | null>()
+        const q = new AsyncQueue<string | null>(1024)
         let done = false
 
         q.push(

packages/opencode/src/server/routes/global.ts

@@ -19,7 +19,7 @@ export const GlobalDisposedEvent = BusEvent.define("global.disposed", z.object({
 
 async function streamEvents(c: Context, subscribe: (q: AsyncQueue<string | null>) => () => void) {
   return streamSSE(c, async (stream) => {
-    const q = new AsyncQueue<string | null>()
+    const q = new AsyncQueue<string | null>(1024)
     let done = false
 
     q.push(

packages/opencode/src/util/queue.ts

@@ -1,11 +1,17 @@
 export class AsyncQueue<T> implements AsyncIterable<T> {
   private queue: T[] = []
   private resolvers: ((value: T) => void)[] = []
+  private readonly capacity: number | undefined
+
+  constructor(capacity?: number) {
+    this.capacity = capacity
+  }
 
   push(item: T) {
     const resolve = this.resolvers.shift()
-    if (resolve) resolve(item)
-    else this.queue.push(item)
+    if (resolve) return resolve(item)
+    this.queue.push(item)
+    if (this.capacity !== undefined) while (this.queue.length > this.capacity) this.queue.shift()
   }
 
   async next(): Promise<T> {