fix(client): cancel the SSE stream on teardown to avoid leaking connections

`readFrom` (used by `parseSseStream` for both the JSON-RPC and REST client
transports) only called `reader.releaseLock()` in its `finally`. When a
consumer stops iterating early — a `break`, a `throw`, or a REST transport
that throws on an `error` event — the async generator's `return()`/`throw()`
runs that `finally`, which detaches the reader but never cancels the
underlying `ReadableStream`. The fetch body (and its socket) is left open,
so repeated early terminations leak connections.

Cancel the reader on teardown so cancellation propagates to the response
body. On normal completion the stream is already closed and `cancel()` is a
no-op; on an errored stream `cancel()` rejects with the same error, which is
ignored so the original error still surfaces.

Adds regression tests asserting the underlying stream is canceled when the
consumer breaks early and when it throws mid-iteration.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: RenKoya1

src/sse_utils.ts

@@ -144,6 +144,10 @@ async function* readFrom(stream: ReadableStream<string>): AsyncGenerator<string,
       yield value;
     }
   } finally {
+    // `releaseLock()` alone leaves the body un-cancelled, leaking the
+    // connection when a consumer breaks/throws early. `.catch()` swallows the
+    // rejection cancel() produces on an already-errored stream.
+    await reader.cancel().catch(() => {});
     reader.releaseLock();
   }
 }

test/sse_utils.spec.ts

@@ -177,6 +177,60 @@ describe('SSE Utils', () => {
     });
   });
 
+  describe('parseSseStream teardown', () => {
+    it('cancels the underlying stream when the consumer stops early', async () => {
+      // An early break must cancel the response body, not just release the lock.
+      let sourceCancelled = false;
+      const stream = new ReadableStream<Uint8Array>({
+        start(controller) {
+          // One event, then stay open — a long-lived SSE connection.
+          controller.enqueue(new TextEncoder().encode('data: {"id":1}\n\n'));
+        },
+        cancel() {
+          sourceCancelled = true;
+        },
+      });
+      const response = new Response(stream, {
+        headers: { 'Content-Type': 'text/event-stream' },
+      });
+
+      const seen: SseEvent[] = [];
+      for await (const event of parseSseStream(response)) {
+        seen.push(event);
+        break;
+      }
+
+      expect(seen).toHaveLength(1);
+      expect(sourceCancelled).toBe(true);
+    });
+
+    it('cancels the underlying stream when the consumer throws', async () => {
+      let sourceCancelled = false;
+      const stream = new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.enqueue(new TextEncoder().encode('data: {"id":1}\n\n'));
+        },
+        cancel() {
+          sourceCancelled = true;
+        },
+      });
+      const response = new Response(stream, {
+        headers: { 'Content-Type': 'text/event-stream' },
+      });
+
+      await expect(
+        (async () => {
+          for await (const event of parseSseStream(response)) {
+            expect(event.data).toBe('{"id":1}');
+            throw new Error('consumer boom');
+          }
+        })()
+      ).rejects.toThrow('consumer boom');
+
+      expect(sourceCancelled).toBe(true);
+    });
+  });
+
   describe('Symmetry: parser understands formatter output', () => {
     it('should parse what formatSSEEvent produces', async () => {
       const originalData = { kind: 'task', id: '123', status: 'completed' };