fix: cancel prompt and session-abort requests on stream close; wire doGenerate abort signal

CHANGELOG.md

@@ -5,6 +5,14 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Fixed
+
+- **Process hang from leaked `/event` SSE connection** ([#30](https://github.com/ben-vargas/ai-sdk-provider-opencode-sdk/pull/30)) - `doStream` passes a per-stream `AbortController` signal to `client.event.subscribe` and aborts it when the stream closes (before `iterator.return()`, which could otherwise block until the next server event). Previously the SSE iterator's `return()` only released its reader lock without cancelling the underlying fetch, so the `GET /event` connection stayed open and kept the Node event loop alive after streaming finished — `examples/abort-signal.ts` intermittently never exited after printing "Done.". The client manager tracks these controllers (`registerEventSubscription`) and aborts any still open during `dispose()`.
+- **Streaming `session.prompt` and `session.abort` requests cancelled on stream close** - The per-stream abort signal from [#30](https://github.com/ben-vargas/ai-sdk-provider-opencode-sdk/pull/30) is now also passed to the `session.prompt` and `session.abort` requests, so a prompt the server never completes (e.g. after an abort race) cannot pin the event loop, and `dispose()` tears these requests down too. Prompt results that arrive after an intentional close are ignored instead of being mis-reported as empty-response errors.
+- **`doGenerate` abort signal** - The non-streaming path now forwards `options.abortSignal` to the prompt request, so aborting `generateText` cancels the underlying HTTP request instead of leaving it pending. A caller-initiated abort now surfaces as an `AbortError` (previously a generic empty-response error) and best-effort aborts the server-side session to stop generation.
+
 ## [3.0.5] - 2026-06-11
 
 ### Added

src/opencode-client-manager.test.ts

@@ -528,6 +528,22 @@ describe("opencode-client-manager", () => {
       // Should not throw
     });
 
+    it("should abort registered event subscription controllers", async () => {
+      const instance = OpencodeClientManager.getInstance();
+
+      const tracked = new AbortController();
+      instance.registerEventSubscription(tracked);
+
+      const unregistered = new AbortController();
+      const unregister = instance.registerEventSubscription(unregistered);
+      unregister();
+
+      await instance.dispose();
+
+      expect(tracked.signal.aborted).toBe(true);
+      expect(unregistered.signal.aborted).toBe(false);
+    });
+
     it("should clear client reference", async () => {
       const instance = OpencodeClientManager.getInstance();
       await instance.getClient();

src/opencode-language-model.test.ts

@@ -372,6 +372,70 @@ describe("opencode-language-model", () => {
       });
     });
 
+    it("should cancel the prompt request and throw AbortError when aborted mid-flight", async () => {
+      mockClient.session.prompt.mockImplementationOnce(
+        (_body: unknown, options?: { signal?: AbortSignal }) =>
+          new Promise((resolve) => {
+            // Emulate fetch abort semantics: the SDK resolves with { error }
+            // instead of rejecting when the request signal is aborted.
+            options?.signal?.addEventListener("abort", () => {
+              resolve({
+                error: Object.assign(new Error("aborted"), {
+                  name: "AbortError",
+                }),
+              });
+            });
+          }),
+      );
+
+      const abortController = new AbortController();
+      const resultPromise = model.doGenerate({
+        prompt: basicPrompt,
+        abortSignal: abortController.signal,
+      });
+
+      setTimeout(() => abortController.abort(), 10);
+
+      await expect(resultPromise).rejects.toMatchObject({
+        name: "AbortError",
+      });
+      expect(mockClient.session.abort).toHaveBeenCalledWith(
+        expect.objectContaining({ sessionID: "session-123" }),
+      );
+    });
+
+    it("should abort the server session when a throwOnError client rejects on abort", async () => {
+      mockClient.session.prompt.mockImplementationOnce(
+        (_body: unknown, options?: { signal?: AbortSignal }) =>
+          new Promise((_resolve, reject) => {
+            // Emulate a client configured with throwOnError: an aborted
+            // fetch rejects instead of resolving with { error }.
+            options?.signal?.addEventListener("abort", () => {
+              reject(
+                Object.assign(new Error("This operation was aborted"), {
+                  name: "AbortError",
+                }),
+              );
+            });
+          }),
+      );
+
+      const abortController = new AbortController();
+      const resultPromise = model.doGenerate({
+        prompt: basicPrompt,
+        abortSignal: abortController.signal,
+      });
+
+      setTimeout(() => abortController.abort(), 10);
+
+      await expect(resultPromise).rejects.toMatchObject({
+        name: "AbortError",
+      });
+      expect(mockClient.session.abort).toHaveBeenCalledWith(
+        expect.objectContaining({ sessionID: "session-123" }),
+      );
+    });
+
     it("should include response error details when response data is missing", async () => {
       mockClient.session.prompt.mockResolvedValueOnce({
         data: undefined,
@@ -1143,9 +1207,126 @@ describe("opencode-language-model", () => {
       expect(parts[0]).toMatchObject({ type: "stream-start" });
       expect(mockClient.session.abort).toHaveBeenCalledWith(
         expect.objectContaining({ sessionID: "session-123" }),
+        expect.objectContaining({ signal: expect.any(AbortSignal) }),
       );
       expect(iteratorReturn).toHaveBeenCalled();
     });
+
+    it("should cancel the in-flight prompt request when the stream is aborted", async () => {
+      const hangingStream: AsyncIterable<unknown> = {
+        [Symbol.asyncIterator]() {
+          return {
+            next() {
+              return new Promise<IteratorResult<unknown>>(() => {
+                // Intentionally never resolves to emulate an idle stream.
+              });
+            },
+            return: () =>
+              Promise.resolve({ done: true as const, value: undefined }),
+          };
+        },
+      };
+
+      mockClient.event.subscribe.mockResolvedValueOnce({
+        stream: hangingStream,
+      });
+
+      let promptOptions: { signal?: AbortSignal } | undefined;
+      mockClient.session.prompt.mockImplementationOnce(
+        (_body: unknown, options?: { signal?: AbortSignal }) => {
+          promptOptions = options;
+          // Emulate a prompt request the server never completes.
+          return new Promise(() => {});
+        },
+      );
+
+      const abortController = new AbortController();
+      const result = await model.doStream({
+        prompt: basicPrompt,
+        abortSignal: abortController.signal,
+      });
+
+      const reader = result.stream.getReader();
+      setTimeout(() => abortController.abort(), 10);
+      while (true) {
+        const { done } = await reader.read();
+        if (done) break;
+      }
+
+      expect(promptOptions?.signal).toBeInstanceOf(AbortSignal);
+      expect(promptOptions?.signal?.aborted).toBe(true);
+    });
+
+    it("should cancel the SSE event subscription when the stream closes", async () => {
+      let subscribeOptions: { signal?: AbortSignal } | undefined;
+      let signalAbortedAtReturn: boolean | undefined;
+
+      mockClient.event.subscribe.mockImplementationOnce(
+        (_params: unknown, options?: { signal?: AbortSignal }) => {
+          subscribeOptions = options;
+          return Promise.resolve({
+            stream: {
+              [Symbol.asyncIterator]() {
+                return {
+                  next: () =>
+                    new Promise<IteratorResult<unknown>>(() => {
+                      // Never resolves: no events arrive for this session.
+                    }),
+                  return: () => {
+                    signalAbortedAtReturn = subscribeOptions?.signal?.aborted;
+                    return Promise.resolve({
+                      done: true as const,
+                      value: undefined,
+                    });
+                  },
+                };
+              },
+            },
+          });
+        },
+      );
+      mockClient.session.prompt.mockImplementationOnce(
+        () => new Promise(() => {}),
+      );
+
+      const abortController = new AbortController();
+      const result = await model.doStream({
+        prompt: basicPrompt,
+        abortSignal: abortController.signal,
+      });
+
+      const reader = result.stream.getReader();
+      setTimeout(() => abortController.abort(), 10);
+      while (true) {
+        const { done } = await reader.read();
+        if (done) break;
+      }
+
+      expect(subscribeOptions?.signal).toBeInstanceOf(AbortSignal);
+      expect(subscribeOptions?.signal?.aborted).toBe(true);
+      // The signal must abort before iterator.return(): the SDK's SSE
+      // generator cancels its underlying fetch only through the signal, and
+      // return() can block until the next event while a read is pending.
+      expect(signalAbortedAtReturn).toBe(true);
+    });
+
+    it("should abort the prompt request signal after the stream completes normally", async () => {
+      const result = await model.doStream({
+        prompt: basicPrompt,
+      });
+
+      const reader = result.stream.getReader();
+      while (true) {
+        const { done } = await reader.read();
+        if (done) break;
+      }
+
+      const promptOptions = mockClient.session.prompt.mock.calls[0]?.[1] as
+        | { signal?: AbortSignal }
+        | undefined;
+      expect(promptOptions?.signal).toBeInstanceOf(AbortSignal);
+      expect(promptOptions?.signal?.aborted).toBe(true);
+    });
   });
 
   describe("getSessionId", () => {

src/opencode-language-model.ts

@@ -287,10 +287,43 @@
         messageID,
       );
 
-      const result = await client.session.prompt(requestBody);
+      const abortSignal = options.abortSignal;
+      const abortServerSession = async () => {
+        const directory = this.getRequestDirectory();
+        try {
+          await client.session.abort({
+            sessionID: sessionId,
+            ...(directory ? { directory } : {}),
+          });
+        } catch {
+          // ignore abort errors
+        }
+      };
+
+      let result: unknown;
+      try {
+        result = abortSignal
+          ? await client.session.prompt(requestBody, { signal: abortSignal })
+          : await client.session.prompt(requestBody);
+      } catch (error) {
+        // Clients configured with throwOnError reject on fetch abort instead
+        // of resolving { error }; still stop server-side generation.
+        if (isAbortError(error) || abortSignal?.aborted) {
+          await abortServerSession();
+        }
+        throw error;
+      }
 
       const { data, error: responseError } = extractSdkResult(result);
       if (!data) {
+        // The SDK surfaces fetch aborts as { error } instead of rejecting, so
+        // map a caller-initiated abort to AbortError and stop server-side work.
+        if (abortSignal?.aborted) {
+          await abortServerSession();
+          const abortError = new Error("Request aborted");
+          abortError.name = "AbortError";
+          throw abortError;
+        }
         throw createEmptyResponseDataError(responseError, {
           sessionId,
           modelId: this.modelId,
@@ -303,11 +336,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);
@@ -402,18 +435,19 @@
       start: async (controller) => {
         const streamWarnings = [...warnings];
         let streamStartEmitted = false;
-
-        // The SDK's SSE generator only cancels its fetch reader from an
+        // Cancels the stream's in-flight HTTP requests (SSE event
+        // subscription, prompt, session abort) when the stream closes. 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 requestAbortController = new AbortController();
         const unregisterEventSubscription =
-          this.clientManager.registerEventSubscription(eventAbortController);
+          this.clientManager.registerEventSubscription(requestAbortController);
 
         try {
           const eventsResult = await client.event.subscribe(
             directory ? { directory } : undefined,
-            { signal: eventAbortController.signal },
+            { signal: requestAbortController.signal },
           );
 
           const eventStream = eventsResult.stream;
@@ -471,17 +505,31 @@
             resolvePromptFailed?.();
           };
 
-          client.session.prompt(requestBody).then((result) => {
-            const { data, error: responseError } = extractSdkResult(result);
-            if (!data) {
-              handlePromptFailure(
-                createEmptyResponseDataError(responseError, {
-                  sessionId,
-                  modelId: this.modelId,
-                }),
-              );
-            }
-          }, handlePromptFailure);
+          client.session
+            .prompt(requestBody, { signal: requestAbortController.signal })
+            .then(
+              (result) => {
+                if (requestAbortController.signal.aborted) {
+                  // The stream already closed; the prompt outcome is irrelevant.
+                  return;
+                }
+                const { data, error: responseError } = extractSdkResult(result);
+                if (!data) {
+                  handlePromptFailure(
+                    createEmptyResponseDataError(responseError, {
+                      sessionId,
+                      modelId: this.modelId,
+                    }),
+                  );
+                }
+              },
+              (error) => {
+                if (requestAbortController.signal.aborted) {
+                  return;
+                }
+                handlePromptFailure(error);
+              },
+            );
 
           const state = createStreamState();
           let lastMessageInfo: Message | undefined;
@@ -495,7 +543,7 @@
             // 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();
+            requestAbortController.abort();
             if (typeof iterator.return === "function") {
               try {
                 await iterator.return(undefined);
@@ -523,10 +571,13 @@
 
               if (result.type === "aborted") {
                 try {
-                  await client.session.abort({
-                    sessionID: sessionId,
-                    ...(directory ? { directory } : {}),
-                  });
+                  await client.session.abort(
+                    {
+                      sessionID: sessionId,
+                      ...(directory ? { directory } : {}),
+                    },
+                    { signal: requestAbortController.signal },
+                  );
                 } catch {
                   // ignore abort errors
                 }
@@ -593,7 +644,7 @@
             controller.enqueue({ type: "error", error: wrappedError });
           }
         } finally {
-          eventAbortController.abort();
+          requestAbortController.abort();
           unregisterEventSubscription();
           controller.close();
         }