fix: close OpenCode event streams before ending requests

Completed parallel conversations could leave buffered event connections open.
Chat and native context compaction now cancel their own subscriptions before
ending SDK iteration, without cancelling another conversation or its Server.

Strengthen cleanup-order regressions and verify real OpenCode parallel runs,
cancellation and compaction. The remote Agent uses ACP stdio and transcript
polling rather than this SDK event stream and needs no corresponding change.

Validation: 99 focused tests passed; the final candidate passed 4089 tests
with 66 skipped, typecheck and lint. No external model calls were made.

src/main/agent/opencode-runtime-lifecycle.test.ts

@@ -15,7 +15,7 @@ const binaryPath = join(
 )
 
 it.skipIf(!existsSync(binaryPath))(
-  'closes real event streams after parallel runs without cancelling a peer',
+  'closes real event streams after parallel runs and compaction without cancelling a peer',
   async () => {
     const workspace = await mkdtemp(join(tmpdir(), 'goodbuddy-opencode-lifecycle-'))
     let cancelledController: AbortController | undefined
@@ -94,14 +94,18 @@ it.skipIf(!existsSync(binaryPath))(
         }
       })
     })
-    const run = async (controller = new AbortController(), cancel = false): Promise<string> => {
+    const run = async (
+      controller = new AbortController(),
+      cancel = false,
+      conversationId = crypto.randomUUID()
+    ): Promise<string> => {
       const requestId = crypto.randomUUID()
       const token = gateway.grant(requestId, [crypto.randomUUID()], controller.signal)!
       const deadline = setTimeout(() => controller.abort(), 30_000)
       let text = ''
       try {
         for await (const event of runtime.run({
-          requestId, conversationId: crypto.randomUUID(), workMode: 'ask',
+          requestId, conversationId, workMode: 'ask',
           prompt: cancel ? 'CANCEL_LIFECYCLE' : 'Reply LIFECYCLE_OK without tools.',
           knowledgeCapabilityToken: token
         }, controller.signal)) {
@@ -126,6 +130,18 @@ it.skipIf(!existsSync(binaryPath))(
         expect(run()).resolves.toBe('LIFECYCLE_OK')
       ])
       await expect.poll(() => openEventBodies).toBe(0)
+      const conversationId = crypto.randomUUID()
+      await expect(run(new AbortController(), false, conversationId)).resolves.toBe('LIFECYCLE_OK')
+      await expect(runtime.compactConversation({
+        requestId: crypto.randomUUID(),
+        conversationId,
+        runtimeSelection: { provider: 'opencode' },
+        history: [],
+        historyMessageIds: []
+      }, new AbortController().signal)).resolves.toMatchObject({
+        result: { compacted: true }
+      })
+      await expect.poll(() => openEventBodies).toBe(0)
     } finally {
       await runtime.dispose()
       await gateway.dispose()

src/main/agent/opencode-runtime.test.ts

@@ -1737,6 +1737,7 @@ describe("OpenCodeRuntime embedded launcher", () => {
       const setup = runClient([]);
       const controller = new AbortController();
       let subscriptionSignal: AbortSignal | undefined;
+      let abortedBeforeIteratorClose: boolean | undefined;
       const subscribe = setup.event.subscribe as unknown as ReturnType<
         typeof vi.fn
       >;
@@ -1745,15 +1746,19 @@ describe("OpenCodeRuntime embedded launcher", () => {
           subscriptionSignal = options.signal;
           return {
             stream: (async function* () {
-              if (outcome === "cancelled") {
-                controller.abort(new DOMException("Cancelled", "AbortError"));
-                return;
+              try {
+                if (outcome === "cancelled") {
+                  controller.abort(new DOMException("Cancelled", "AbortError"));
+                  return;
+                }
+                if (outcome === "failed") return;
+                yield {
+                  type: "session.idle",
+                  properties: { sessionID: "session-1" },
+                };
+              } finally {
+                abortedBeforeIteratorClose = subscriptionSignal?.aborted;
               }
-              if (outcome === "failed") return;
-              yield {
-                type: "session.idle",
-                properties: { sessionID: "session-1" },
-              };
             })(),
           };
         },
@@ -1784,6 +1789,9 @@ describe("OpenCodeRuntime embedded launcher", () => {
           await collect();
         }
         expect(subscriptionSignal?.aborted).toBe(true);
+        if (outcome !== "failed") {
+          expect(abortedBeforeIteratorClose).toBe(true);
+        }
         expect(controller.signal.aborted).toBe(outcome === "cancelled");
       } finally {
         await runtime.dispose();

src/main/agent/opencode-runtime.ts

@@ -295,6 +295,24 @@ function eventSubscriptionOptions(signal: AbortSignal) {
   };
 }
 
+async function* consumeEventSubscription<T>(
+  stream: AsyncGenerator<T, void, unknown>,
+  controller: AbortController,
+): AsyncGenerator<T, void, unknown> {
+  try {
+    while (true) {
+      const next = await stream.next();
+      if (next.done) return;
+      yield next.value;
+    }
+  } finally {
+    // The SDK removes its abort listener on iterator return without cancelling
+    // the reader. Abort first, while that listener can still close the body.
+    controller.abort();
+    await stream.return();
+  }
+}
+
 function awaitWithAbort<T>(
   operation: Promise<T>,
   signal: AbortSignal,
@@ -2466,7 +2484,10 @@ export class OpenCodeRuntime implements AgentRuntime {
         const reportedMessageIds = new Set<string>();
         const childTaskIdsBySession = new Map<string, string>();
         let waitingForRetry = false;
-        for await (const event of subscription.stream) {
+        for await (const event of consumeEventSubscription(
+          subscription.stream,
+          subscriptionController,
+        )) {
           const childProgress = subagentProgress.update(event);
           if (
             childProgress &&
@@ -3101,7 +3122,10 @@ export class OpenCodeRuntime implements AgentRuntime {
       const usageEvents: RuntimeModelUsageEvent[] = [];
       const reportedMessageIds = new Set<string>();
       const usageCapture = (async () => {
-        for await (const event of subscription.stream) {
+        for await (const event of consumeEventSubscription(
+          subscription.stream,
+          subscriptionController,
+        )) {
           if (
             event.type === "message.updated" &&
             event.properties.sessionID === sessionId &&