fix(responses): surface rewrite failures before tee cancellation

docs-site/src/content/docs/reference/proxy-formats.md

@@ -83,6 +83,11 @@ This applies to both tee inspection and eager relay, including Windows rewrite t
 even when the upstream read rejects before the response-body cancellation hook runs.
 A terminal captured during the bounded post-disconnect drain retains its actual outcome.
 
+If native passthrough rewriting fails, including when it exceeds the translation
+buffer budget, the relay reports the failure without waiting for upstream inspection
+to finish. It cancels the upstream work and emits `response.failed` followed by
+`data: [DONE]`; a budget overflow uses the `translation_buffer_limit` error code.
+
 Client-facing Responses SSE frames are limited to 4 MiB per frame, measured in raw bytes before the
 SSE block delimiter. On HTTP, an unterminated upstream frame that exceeds the limit fails closed
 with a synthetic `response.failed` event followed by `data: [DONE]`. On the Responses WebSocket

src/server/sse-payload-rewrite.ts

@@ -249,7 +249,9 @@ export function relaySseWithBlockRewrite(
       } catch (error) {
         releaseBuffer();
         disposeRewrite();
-        try { await reader.cancel(error); } catch { /* already closed */ }
+        // Cancelling one tee branch waits for its sibling. Surface the failure
+        // now so downstream can abort upstream and release the inspection branch.
+        void reader.cancel(error).catch(() => {});
         controller.error(error);
       }
     },

tests/responses/sse-payload-rewrite.test.ts

@@ -153,4 +153,82 @@ describe("SSE payload rewrite composition", () => {
     expect(budget.snapshot().currentBytes).toBe(0);
     budget.dispose();
   });
+
+  test.each(["resolve", "reject"] as const)(
+    "surfaces a rewrite failure before tee cancellation can %s",
+    async cancellationOutcome => {
+      const budget = createTestTranslatorBudget({ maxTurnBytes: 64 });
+      const upstream = new AbortController();
+      const cancellation = Promise.withResolvers<void>();
+      const cancellationError = new Error("upstream cancellation failed");
+      let cancelCalls = 0;
+      let disposeCalls = 0;
+      const source = new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.enqueue(new TextEncoder().encode("data: partial"));
+          controller.enqueue(new TextEncoder().encode("x".repeat(80)));
+          // Keep the source open after exhausting the rewrite budget.
+        },
+        cancel() {
+          cancelCalls += 1;
+          return cancellation.promise;
+        },
+      });
+      const [native, inspection] = source.tee();
+      const inspectionReader = inspection.getReader();
+      await inspectionReader.read();
+      await inspectionReader.read();
+      let inspectionSettled = false;
+      const pendingInspection = inspectionReader.read().then(() => { inspectionSettled = true; });
+      const rewrite = Object.assign((block: string) => [block], {
+        dispose() { disposeCalls += 1; },
+      });
+      const rewritten = relaySseWithBlockRewrite(native, rewrite, budget);
+      const client = relaySseWithFailedTail(rewritten, upstream);
+      const completion = readAll(client);
+      let deadline: ReturnType<typeof setTimeout> | undefined;
+
+      try {
+        const out = await Promise.race([
+          completion,
+          new Promise<never>((_, reject) => {
+            deadline = setTimeout(() => reject(new Error("rewrite failure waited for the inspection tee")), 1_000);
+          }),
+        ]);
+        expect(out.match(/event: response.failed/g)).toHaveLength(1);
+        expect(out).toContain('"code":"translation_buffer_limit"');
+        expect(out).toEndWith("data: [DONE]\n\n");
+        expect(upstream.signal.aborted).toBe(true);
+        expect(inspectionSettled).toBe(false);
+        expect(cancelCalls).toBe(0);
+        expect(disposeCalls).toBe(1);
+        expect(budget.snapshot().currentBytes).toBe(0);
+        expect(budget.snapshot().overflows).toBe(1);
+
+        // Releasing inspection settles both tee cancellation promises. A late
+        // rejection must be handled by the rewriter as well as this reader.
+        const siblingCancellation = inspectionReader.cancel("inspection cleanup");
+        expect(cancelCalls).toBe(1);
+        if (cancellationOutcome === "reject") {
+          cancellation.reject(cancellationError);
+          await expect(siblingCancellation).rejects.toBe(cancellationError);
+        } else {
+          cancellation.resolve();
+          await siblingCancellation;
+        }
+        await pendingInspection;
+        await Bun.sleep(0); // Let the runner observe any unhandled cancellation rejection.
+        expect(disposeCalls).toBe(1);
+      } finally {
+        clearTimeout(deadline);
+        const cleanup = inspectionReader.cancel().catch(() => {});
+        cancellation.resolve();
+        await cleanup;
+        await pendingInspection;
+        await completion.catch(() => {});
+        inspectionReader.releaseLock();
+        budget.dispose();
+      }
+    },
+  );
 });