From b055a3d50717c86a84ac8c814e50602fff8aa4ac Mon Sep 17 00:00:00 2001
From: Dave Jong <dave.jong@patchstack.com>
Date: Wed, 12 Aug 2026 13:10:32 +0200
Subject: [PATCH] fix(protect): don't buffer live streams (SSE) in response
 screening
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Response screening buffers the whole body before deciding, so a Server-Sent-Events / token
stream — the common shape for LLM output in AI-built apps — was withheld from the client until
the stream *ended* (readTextResponse awaits the reader to `done`). Exclude `text/event-stream`
from screening on both paths (the fetch `readTextResponse` and the node `isTextCT` gate): a live
stream passes through unbuffered and streams normally, unscreened.

This removes the last blocker to enabling response screening by default on streaming stacks.
+2 tests (a never-closing SSE stream returns promptly & untouched; a normal text response is
still screened); 628 pass; typecheck clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 src/protect/runtime.js                   |  7 ++++
 tests/protect/response-streaming.test.ts | 48 ++++++++++++++++++++++++
 2 files changed, 55 insertions(+)
 create mode 100644 tests/protect/response-streaming.test.ts

diff --git a/src/protect/runtime.js b/src/protect/runtime.js
index b6128c8..20cdc2f 100644
--- a/src/protect/runtime.js
+++ b/src/protect/runtime.js
@@ -172,6 +172,10 @@ export async function createProtection(options = {}) {
   // block mode (dry-run records via onDetect but returns 'pass').
   const isTextCT = (ct) => {
     ct = (ct || '').toLowerCase();
+    // Exclude live streams: a Server-Sent-Events / token stream must pass through unbuffered.
+    // Screening buffers the whole body, so it would withhold every chunk until the stream ends —
+    // breaking incremental LLM streaming, which AI-built apps lean on heavily.
+    if (ct.includes('event-stream')) return false;
     return ct === '' || /(json|text|xml|html|javascript|csv|yaml|x-www-form-urlencoded)/.test(ct);
   };
   const screenText = (text, meta, reqCtx) => {
@@ -585,6 +589,9 @@ function byPhase(rules, phase) {
 async function readTextResponse(response, cap = DEFAULT_SCREEN_CAP) {
   if (!response || typeof response.clone !== 'function') return null;
   const ct = (response.headers?.get?.('content-type') || '').toLowerCase();
+  // Live stream (SSE / token stream): never buffer it — reading to completion would withhold the
+  // response until the stream ends, breaking incremental streaming. Pass it through unscreened.
+  if (ct.includes('event-stream')) return null;
   const isText = ct === '' || /(json|text|xml|html|javascript|csv|yaml|x-www-form-urlencoded)/.test(ct);
   if (!isText) return null;
   const len = Number(response.headers?.get?.('content-length') || 0);
diff --git a/tests/protect/response-streaming.test.ts b/tests/protect/response-streaming.test.ts
new file mode 100644
index 0000000..a6c7aae
--- /dev/null
+++ b/tests/protect/response-streaming.test.ts
@@ -0,0 +1,48 @@
+import { describe, expect, it } from 'vitest';
+import { createProtection } from '../../src/protect/runtime.js';
+
+// Response screening buffers the whole body, which would withhold a live stream (SSE / LLM tokens)
+// until it ends — breaking incremental streaming. `text/event-stream` responses must pass through
+// unbuffered.
+
+describe('response phase — live streams pass through unbuffered', () => {
+  it('returns an SSE response immediately without consuming the (never-ending) stream', async () => {
+    const p: any = await createProtection({ mode: 'block' }); // default response rules active
+
+    // A stream that emits one event and never closes — modelling a long-lived token stream.
+    const stream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode('data: hello\n\n'));
+        // deliberately never call controller.close()
+      },
+    });
+    const sse = new Response(stream, { status: 200, headers: { 'content-type': 'text/event-stream' } });
+
+    // If screening tried to buffer this, it would await `done` forever and this would time out.
+    const out: any = await Promise.race([
+      p.screenResponse(sse),
+      new Promise((r) => setTimeout(() => r('TIMEOUT'), 500)),
+    ]);
+
+    expect(out).not.toBe('TIMEOUT'); // returned promptly
+    expect(out).toBe(sse); // the ORIGINAL response, untouched (not rebuilt/screened)
+  });
+
+  it('still screens a normal (non-stream) text response', async () => {
+    const rule = {
+      phase: 'response',
+      category: 'x',
+      action: 'redact',
+      rule_v2: [{ parameter: 'response.body', match: { type: 'regex', value: '/topsecret/' } }],
+    };
+    const p: any = await createProtection({
+      rules: { firewall: [], whitelists: [], whitelist_keys: {} },
+      responseRules: [rule],
+      mode: 'block',
+    });
+    const out = await p.screenResponse(
+      new Response('a topsecret b', { status: 200, headers: { 'content-type': 'text/plain' } }),
+    );
+    expect(/topsecret/.test(await out.text())).toBe(false); // non-stream text is still screened
+  });
+});