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..ab9e001 100644 --- a/src/opencode-language-model.test.ts +++ b/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 = { + [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..4b12945 100644 --- a/src/opencode-language-model.ts +++ b/src/opencode-language-model.ts @@ -287,10 +287,43 @@ export class OpencodeLanguageModel implements LanguageModelV3 { 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, @@ -402,18 +435,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 +505,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 +543,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 +571,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 +644,7 @@ export class OpencodeLanguageModel implements LanguageModelV3 { controller.enqueue({ type: "error", error: wrappedError }); } } finally { - eventAbortController.abort(); + requestAbortController.abort(); unregisterEventSubscription(); controller.close(); }