fix: abort SSE event subscription on stream close so process can exit

src/opencode-client-manager.ts

@@ -40,6 +40,7 @@ export class OpencodeClientManager {
   private logger: Logger;
   private initPromise: Promise<OpencodeClient> | null = null;
   private isDisposed = false;
+  private activeEventSubscriptions = new Set<AbortController>();
   private cleanupHandlersRegistered = false;
   private cleanupHandlers: {
     exit?: () => void;
@@ -321,6 +322,17 @@ export class OpencodeClientManager {
     return this.server !== null;
   }
 
+  /**
+   * Track an event-stream AbortController so dispose() can tear down SSE
+   * connections that are still open. Returns an unregister function.
+   */
+  registerEventSubscription(controller: AbortController): () => void {
+    this.activeEventSubscriptions.add(controller);
+    return () => {
+      this.activeEventSubscriptions.delete(controller);
+    };
+  }
+
   /**
    * Dispose of the client manager, stopping the server if managed.
    */
@@ -332,6 +344,13 @@ export class OpencodeClientManager {
     this.isDisposed = true;
     this.unregisterCleanupHandlers();
 
+    // Abort any SSE event streams still open; stopping a managed server does
+    // not close them, and for reused servers nothing else would.
+    for (const controller of this.activeEventSubscriptions) {
+      controller.abort();
+    }
+    this.activeEventSubscriptions.clear();
+
     // Release the singleton slot so a later getInstance() builds a fresh
     // manager instead of returning this disposed one.
     if (OpencodeClientManager.instance === this) {

src/opencode-language-model.test.ts

@@ -76,6 +76,7 @@ const mockClientManager = {
   getClient: vi.fn().mockResolvedValue(mockClient),
   dispose: vi.fn().mockResolvedValue(undefined),
   getServerUrl: vi.fn().mockReturnValue("http://127.0.0.1:4096"),
+  registerEventSubscription: vi.fn().mockReturnValue(() => {}),
 };
 
 describe("opencode-language-model", () => {

src/opencode-language-model.ts

@@ -303,11 +303,11 @@
       };
 
       this.logger.debug?.(
         `[doGenerate] raw parts from OpenCode: ${JSON.stringify(responseData.parts?.map((p) => ({ type: p.type, ...(p.type === "text" ? { text: (p as any).text?.slice(0, 200) } : {}), ...(p.type === "tool" ? { tool: (p as any).tool, callID: (p as any).callID, status: (p as any).state?.status, input: (p as any).state?.input } : {}) })))}`,
       );
       const content = this.extractContentFromParts(responseData.parts ?? []);
       this.logger.debug?.(
         `[doGenerate] extracted content: ${JSON.stringify(content.map((c) => ({ type: c.type, ...(c.type === "text" ? { text: (c as any).text?.slice(0, 200) } : {}), ...(c.type === "tool-call" ? { toolName: (c as any).toolName } : {}) })))}`,
       );
       const usage = this.extractUsageFromParts(responseData.parts ?? []);
       const finishReason = mapOpencodeFinishReason(responseData.info);
@@ -403,9 +403,17 @@
         const streamWarnings = [...warnings];
         let streamStartEmitted = false;
 
+        // The SDK's SSE generator only cancels its fetch reader from an
+        // abort-signal handler; closing the iterator alone leaves the socket
+        // open and keeps the Node event loop alive.
+        const eventAbortController = new AbortController();
+        const unregisterEventSubscription =
+          this.clientManager.registerEventSubscription(eventAbortController);
+
         try {
           const eventsResult = await client.event.subscribe(
             directory ? { directory } : undefined,
+            { signal: eventAbortController.signal },
           );
 
           const eventStream = eventsResult.stream;
@@ -484,6 +492,10 @@
               return;
             }
             iteratorClosed = true;
+            // Abort before iterator.return(): the abort handler cancels the
+            // SSE reader, which also unblocks a pending iterator.next() the
+            // return() call would otherwise wait on.
+            eventAbortController.abort();
             if (typeof iterator.return === "function") {
               try {
                 await iterator.return(undefined);
@@ -581,6 +593,8 @@
             controller.enqueue({ type: "error", error: wrappedError });
           }
         } finally {
+          eventAbortController.abort();
+          unregisterEventSubscription();
           controller.close();
         }
       },

src/opencode-provider.test.ts

@@ -26,13 +26,15 @@ vi.mock("./opencode-client-manager.js", async (importOriginal) => {
         }),
         dispose: vi.fn().mockResolvedValue(undefined),
         getServerUrl: vi.fn().mockReturnValue("http://127.0.0.1:4096"),
+        registerEventSubscription: vi.fn().mockReturnValue(() => {}),
       }),
       resetInstance: vi.fn(),
     },
     createClientManagerFromSettings: vi.fn().mockReturnValue({
       getClient: vi.fn().mockResolvedValue({}),
       dispose: vi.fn().mockResolvedValue(undefined),
       getServerUrl: vi.fn().mockReturnValue("http://127.0.0.1:4096"),
+      registerEventSubscription: vi.fn().mockReturnValue(() => {}),
     }),
   };
 });