From b468520ca7bcc1de88c04cedc60e4024c25d4477 Mon Sep 17 00:00:00 2001 From: Ben Vargas Date: Thu, 11 Jun 2026 09:59:27 -0600 Subject: [PATCH 1/2] fix: cancel prompt and session-abort requests on stream close; wire doGenerate abort signal Follow-up to #30, which fixed the leaked GET /event SSE connection that kept the Node event loop alive after streaming finished. This extends the same per-stream AbortController to the remaining unanchored requests: - session.prompt and session.abort in doStream now receive the stream's abort signal, so a prompt the server never completes (e.g. after an abort race) cannot pin the event loop, and dispose() tears these requests down along with the SSE subscription. Prompt results that arrive after an intentional close are ignored instead of being mis-reported as empty-response errors (the SDK resolves with { error } rather than rejecting on fetch abort). - doGenerate 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 surfaces as AbortError (previously a generic empty-response error) and best-effort aborts the server-side session to stop generation. Adds the regression tests #30 lacked: SSE signal cancellation on stream close (including abort-before-iterator.return ordering), prompt-fetch cancellation on the abort and completion paths, doGenerate mid-flight abort semantics, and manager dispose() aborting registered controllers. Also adds the missing CHANGELOG entries for #30 and this change. --- CHANGELOG.md | 8 ++ src/opencode-client-manager.test.ts | 16 +++ src/opencode-language-model.test.ts | 149 ++++++++++++++++++++++++++++ src/opencode-language-model.ts | 83 +++++++++++----- 4 files changed, 233 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 21ab3e1..c487ba7 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/opencode-client-manager.test.ts b/src/opencode-client-manager.test.ts index 4ec2f11..2891dbe 100644 --- a/src/opencode-client-manager.test.ts +++ b/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(); diff --git a/src/opencode-language-model.test.ts b/src/opencode-language-model.test.ts index 3337965..9c167bc 100644 --- a/src/opencode-language-model.test.ts +++ b/src/opencode-language-model.test.ts @@ -372,6 +372,38 @@ 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 include response error details when response data is missing", async () => { mockClient.session.prompt.mockResolvedValueOnce({ data: undefined, @@ -1143,9 +1175,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 = { + [Symbol.asyncIterator]() { + return { + next() { + return new Promise>(() => { + // 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>(() => { + // 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", () => { diff --git a/src/opencode-language-model.ts b/src/opencode-language-model.ts index 3884aa3..367ace5 100644 --- a/src/opencode-language-model.ts +++ b/src/opencode-language-model.ts @@ -287,10 +287,29 @@ export class OpencodeLanguageModel implements LanguageModelV3 { messageID, ); - const result = await client.session.prompt(requestBody); + const abortSignal = options.abortSignal; + const result = abortSignal + ? await client.session.prompt(requestBody, { signal: abortSignal }) + : await client.session.prompt(requestBody); 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) { + const directory = this.getRequestDirectory(); + try { + await client.session.abort({ + sessionID: sessionId, + ...(directory ? { directory } : {}), + }); + } catch { + // ignore abort errors + } + const abortError = new Error("Request aborted"); + abortError.name = "AbortError"; + throw abortError; + } throw createEmptyResponseDataError(responseError, { sessionId, modelId: this.modelId, @@ -402,18 +421,19 @@ export class OpencodeLanguageModel implements LanguageModelV3 { 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 +491,31 @@ export class OpencodeLanguageModel implements LanguageModelV3 { 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 +529,7 @@ export class OpencodeLanguageModel implements LanguageModelV3 { // 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 +557,13 @@ export class OpencodeLanguageModel implements LanguageModelV3 { 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 +630,7 @@ export class OpencodeLanguageModel implements LanguageModelV3 { controller.enqueue({ type: "error", error: wrappedError }); } } finally { - eventAbortController.abort(); + requestAbortController.abort(); unregisterEventSubscription(); controller.close(); } From 4e35c4dfa7e30983f7773df7386323a9497581c6 Mon Sep 17 00:00:00 2001 From: Ben Vargas Date: Thu, 11 Jun 2026 10:17:18 -0600 Subject: [PATCH 2/2] fix: abort server session when throwOnError client rejects on doGenerate abort Codex review on #31 flagged that clients configured with the supported clientOptions.throwOnError passthrough reject the session.prompt await on fetch abort instead of resolving a fields-style { error } result, so the best-effort session.abort in the empty-data path never ran and the server kept generating after generateText was cancelled. Hoist the server-side abort into a helper and issue it from both the rejection path and the { error } path before rethrowing. --- src/opencode-language-model.test.ts | 32 ++++++++++++++++++++++++ src/opencode-language-model.ts | 38 ++++++++++++++++++++--------- 2 files changed, 58 insertions(+), 12 deletions(-) diff --git a/src/opencode-language-model.test.ts b/src/opencode-language-model.test.ts index 9c167bc..ab9e001 100644 --- a/src/opencode-language-model.test.ts +++ b/src/opencode-language-model.test.ts @@ -404,6 +404,38 @@ describe("opencode-language-model", () => { ); }); + 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, diff --git a/src/opencode-language-model.ts b/src/opencode-language-model.ts index 367ace5..4b12945 100644 --- a/src/opencode-language-model.ts +++ b/src/opencode-language-model.ts @@ -288,24 +288,38 @@ export class OpencodeLanguageModel implements LanguageModelV3 { ); const abortSignal = options.abortSignal; - const result = abortSignal - ? await client.session.prompt(requestBody, { signal: abortSignal }) - : await client.session.prompt(requestBody); + 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) { - const directory = this.getRequestDirectory(); - try { - await client.session.abort({ - sessionID: sessionId, - ...(directory ? { directory } : {}), - }); - } catch { - // ignore abort errors - } + await abortServerSession(); const abortError = new Error("Request aborted"); abortError.name = "AbortError"; throw abortError;