fix(sdk): close abandoned SSE responses when a watcher stops

sdks/typescript/src/client.ts

@@ -962,72 +962,94 @@ export class TraceDecayClient {
         },
         options.signal,
       );
-      if (!response.ok || response.body === null) {
-        await this.readJson(response);
-        throw new TraceDecayMalformedResponseError(
-          "the daemon did not open an SSE stream",
-          { status: response.status },
-        );
-      }
-      if (!hasMediaType(response, "text/event-stream")) {
-        throw new TraceDecayMalformedResponseError(
-          "the daemon opened an operation stream with an invalid media type",
-          { status: response.status },
-        );
-      }
-
+      let streamFailed = false;
       try {
-        for await (const frame of parseSse(response.body)) {
-          if (frame.retryDelayMs !== undefined) {
-            reconnectDelayMs = frame.retryDelayMs;
-          }
-          const decoded = decodeSse(frame, response.status, operationId);
-          if (decoded.frontier !== undefined) {
-            nextSequence = decoded.frontier.nextSequence;
-            resumeToken = decoded.frontier.resumeToken;
-          }
-          if (decoded.sequence !== undefined) {
-            if (
-              lastSequence !== undefined &&
-              BigInt(decoded.sequence) <= BigInt(lastSequence)
-            ) {
-              continue;
+        if (!response.ok || response.body === null) {
+          await this.readJson(response);
+          throw new TraceDecayMalformedResponseError(
+            "the daemon did not open an SSE stream",
+            { status: response.status },
+          );
+        }
+        if (!hasMediaType(response, "text/event-stream")) {
+          throw new TraceDecayMalformedResponseError(
+            "the daemon opened an operation stream with an invalid media type",
+            { status: response.status },
+          );
+        }
+
+        try {
+          for await (const frame of parseSse(response.body)) {
+            if (frame.retryDelayMs !== undefined) {
+              reconnectDelayMs = frame.retryDelayMs;
+            }
+            const decoded = decodeSse(frame, response.status, operationId);
+            if (decoded.frontier !== undefined) {
+              nextSequence = decoded.frontier.nextSequence;
+              resumeToken = decoded.frontier.resumeToken;
             }
-            const expected =
-              lastSequence === undefined ? undefined : incrementDecimal(lastSequence);
-            if (expected !== undefined && decoded.sequence !== expected) {
-              throw new TraceDecayMalformedResponseError(
-                "the daemon returned a non-contiguous SSE sequence",
-                { status: response.status, payload: decoded.event },
-              );
+            if (decoded.sequence !== undefined) {
+              if (
+                lastSequence !== undefined &&
+                BigInt(decoded.sequence) <= BigInt(lastSequence)
+              ) {
+                continue;
+              }
+              const expected =
+                lastSequence === undefined ? undefined : incrementDecimal(lastSequence);
+              if (expected !== undefined && decoded.sequence !== expected) {
+                throw new TraceDecayMalformedResponseError(
+                  "the daemon returned a non-contiguous SSE sequence",
+                  { status: response.status, payload: decoded.event },
+                );
+              }
+              lastSequence = decoded.lastSequence ?? decoded.sequence;
+              nextSequence = incrementDecimal(lastSequence);
             }
-            lastSequence = decoded.lastSequence ?? decoded.sequence;
-            nextSequence = incrementDecimal(lastSequence);
+            yield {
+              event: frame.event,
+              id: decoded.sequence ?? null,
+              data: decoded.event,
+            };
+            if (TERMINAL_EVENTS.has(frame.event)) {
+              return;
+            }
+          }
+        } catch (cause) {
+          streamFailed = true;
+          if (cause instanceof TraceDecayProtocolError) {
+            throw cause;
           }
-          yield {
-            event: frame.event,
-            id: decoded.sequence ?? null,
-            data: decoded.event,
-          };
-          if (TERMINAL_EVENTS.has(frame.event)) {
-            return;
+          if (options.signal?.aborted) {
+            throw new TraceDecayAbortError(options.signal.reason ?? cause);
+          }
+          if (reconnects >= maxReconnects) {
+            if (cause instanceof TraceDecayDisconnectedError) {
+              throw cause;
+            }
+            throw new TraceDecayDisconnectedError(
+              `operation stream disconnected after ${reconnects} reconnects`,
+              cause,
+            );
           }
         }
       } catch (cause) {
-        if (cause instanceof TraceDecayProtocolError) {
-          throw cause;
-        }
-        if (options.signal?.aborted) {
-          throw new TraceDecayAbortError(options.signal.reason ?? cause);
-        }
-        if (reconnects >= maxReconnects) {
-          if (cause instanceof TraceDecayDisconnectedError) {
-            throw cause;
+        streamFailed = true;
+        throw cause;
+      } finally {
+        // Closing a watcher must close its HTTP body without cancelling the operation.
+        // parseSse has already released its reader when this outer cleanup runs.
+        try {
+          await response.body?.cancel();
+        } catch (cause) {
+          // Do not mask a primary failure (including a retryable disconnect).
+          // Otherwise disposal must report that its transport did not close.
+          if (!streamFailed) {
+            throw new TraceDecayDisconnectedError(
+              "failed to close operation stream transport",
+              cause,
+            );
           }
-          throw new TraceDecayDisconnectedError(
-            `operation stream disconnected after ${reconnects} reconnects`,
-            cause,
-          );
         }
       }
 

sdks/typescript/test/client.test.ts

@@ -1798,6 +1798,91 @@ describe("TraceDecayClient transport envelopes", () => {
   });
 });
 
+describe("TraceDecayClient operation stream disposal", () => {
+  const open = [
+    "event: open",
+    'data: {"event":"open","data":{"correlation_id":"request.operation","frontier":{"next_sequence":0,"retained_from_sequence":0,"resume_token":"resume"}}}',
+    "",
+    "",
+  ].join("\n");
+
+  it("cancels a live HTTP body and releases its reader on early iterator return", async () => {
+    let cancelled = 0;
+    const response = new Response(new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode(open));
+      },
+      cancel() {
+        cancelled += 1;
+      },
+    }), { headers: { "content-type": "text/event-stream" } });
+    const client = createClient({
+      baseUrl: "http://127.0.0.1:4317",
+      projectId: "project.sdk",
+      token: "sdk-secret",
+      fetch: async () => response,
+    });
+    const iterator = client.streamOperation("request.operation");
+
+    expect((await iterator.next()).value?.event).toBe("open");
+    await iterator.return(undefined);
+
+    expect(cancelled).toBe(1);
+    expect(response.body?.locked).toBe(false);
+    expect(await iterator.next()).toEqual({ done: true, value: undefined });
+  });
+
+  it("closes a rejected non-SSE response without suppressing the protocol error", async () => {
+    let cancelled = 0;
+    const response = new Response(new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode(open));
+      },
+      cancel() {
+        cancelled += 1;
+      },
+    }), { headers: { "content-type": "text/plain" } });
+    const client = createClient({
+      baseUrl: "http://127.0.0.1:4317",
+      projectId: "project.sdk",
+      token: "sdk-secret",
+      fetch: async () => response,
+    });
+    await expect(client.streamOperation("request.operation").next()).rejects.toThrow(
+      "the daemon opened an operation stream with an invalid media type",
+    );
+    expect(cancelled).toBe(1);
+    expect(response.body?.locked).toBe(false);
+  });
+
+  it("closes a live HTTP watcher without posting an operation cancellation", async () => {
+    let close: Promise<void> | undefined;
+    await withServer(
+      [
+        (_request, response) => {
+          close = new Promise<void>((resolve) => response.once("close", resolve));
+          response.writeHead(200, { "content-type": "text/event-stream" });
+          response.write(open);
+        },
+      ],
+      async (baseUrl, requests) => {
+        const client = createClient({ baseUrl, projectId: "project.sdk", token: "sdk-secret" });
+        for await (const event of client.streamOperation("request.operation")) {
+          expect(event.event).toBe("open");
+          break;
+        }
+        expect(close).toBeDefined();
+        await Promise.race([
+          close,
+          new Promise<never>((_resolve, reject) =>
+            setTimeout(() => reject(new Error("watcher HTTP connection never closed")), 2000)),
+        ]);
+        expect(requests.map(request => request.method)).toEqual(["GET"]);
+      },
+    );
+  });
+});
+
 describe("TraceDecayClient operation lifecycle", () => {
   it("accepts the server media type, UTF-8 BOM, and inherited SSE event ID", async () => {
     await withServer(

sdks/typescript/test/stream-disposal-error.test.ts

@@ -0,0 +1,83 @@
+import { describe, expect, it, vi } from "vitest";
+
+import {
+  createClient,
+  TraceDecayDisconnectedError,
+  TraceDecayMalformedResponseError,
+} from "../src/client";
+
+const OPEN = [
+  "event: open",
+  'data: {"event":"open","data":{"correlation_id":"request.operation","frontier":{"next_sequence":0,"retained_from_sequence":0,"resume_token":"resume"}}}',
+  "",
+  "",
+].join("\n");
+
+function rejectingCleanupFixture(content = OPEN, contentType = "text/event-stream") {
+  const cleanupError = new Error("underlying transport could not close");
+  const cancel = vi.fn(async () => { throw cleanupError; });
+  const response = new Response(new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(new TextEncoder().encode(content));
+    },
+    cancel,
+  }), { headers: { "content-type": contentType } });
+  const fetch = vi.fn(async () => response);
+  const client = createClient({
+    baseUrl: "http://127.0.0.1:4317",
+    projectId: "project.sdk",
+    token: "sdk-secret",
+    fetch,
+  });
+  return { client, response, cancel, cleanupError, fetch };
+}
+
+describe("operation stream cleanup failures", () => {
+  it("reports a typed transport failure when early return cannot close its body", async () => {
+    const fixture = rejectingCleanupFixture();
+    const iterator = fixture.client.streamOperation("request.operation");
+    expect((await iterator.next()).value?.event).toBe("open");
+
+    const failure = await iterator.return(undefined).catch((error: unknown) => error);
+    expect(failure).toBeInstanceOf(TraceDecayDisconnectedError);
+    expect((failure as TraceDecayDisconnectedError).cause).toBe(fixture.cleanupError);
+    expect(fixture.cancel).toHaveBeenCalledTimes(1);
+    expect(fixture.fetch).toHaveBeenCalledTimes(1);
+    expect(fixture.response.body?.locked).toBe(false);
+    expect(await iterator.next()).toEqual({ done: true, value: undefined });
+  });
+
+  it("propagates failed cleanup through a consumer's break", async () => {
+    const fixture = rejectingCleanupFixture();
+    const consume = async () => {
+      for await (const _event of fixture.client.streamOperation("request.operation")) {
+        break;
+      }
+    };
+    await expect(consume()).rejects.toBeInstanceOf(TraceDecayDisconnectedError);
+    expect(fixture.cancel).toHaveBeenCalledTimes(1);
+    expect(fixture.fetch).toHaveBeenCalledTimes(1);
+  });
+
+  it("does not replace a response validation error with a cleanup error", async () => {
+    const fixture = rejectingCleanupFixture(OPEN, "text/plain");
+    const failure = await fixture.client.streamOperation("request.operation")
+      .next().catch((error: unknown) => error);
+    expect(failure).toBeInstanceOf(TraceDecayMalformedResponseError);
+    expect((failure as Error).message).toBe(
+      "the daemon opened an operation stream with an invalid media type",
+    );
+    expect(fixture.cancel).toHaveBeenCalledTimes(1);
+    expect(fixture.response.body?.locked).toBe(false);
+  });
+
+  it("does not replace a malformed SSE payload with a cleanup error", async () => {
+    const fixture = rejectingCleanupFixture("event: open\ndata: {invalid\n\n");
+    const failure = await fixture.client.streamOperation("request.operation")
+      .next().catch((error: unknown) => error);
+    expect(failure).toBeInstanceOf(TraceDecayMalformedResponseError);
+    expect((failure as Error).message).toBe("the daemon returned non-JSON data");
+    expect(fixture.cancel).toHaveBeenCalledTimes(1);
+    expect(fixture.response.body?.locked).toBe(false);
+  });
+});