Fix: Forward-proxy streams SSE responses without a StreamingResponder

[huang195]

Summary

Fixes #642 — the forward proxy blocked SSE responses from a generic (e.g. MCP Streamable HTTP) upstream: an agent connecting to an SSE-based MCP server over the forward proxy received no events until it timed out and closed the connection.

Root cause

In forwardproxy/server.go serveOutbound, the frame-by-frame flushing SSE path (handleStreamingResponse) was gated on HasStreamingResponders():

if isEventStream(resp.Header.Get("Content-Type")) &&
    s.OutboundPipeline.HasStreamingResponders() &&   // ← too narrow
    resp.Body != nil { ... }

A plain pipeline (the reporter's is just token-exchange — not a StreamingResponder) failed that gate and fell through to the unflushed io.Copy(w, resp.Body). Go's default ResponseWriter buffers small writes, so intermittent SSE events were never pushed until the handler returned / the upstream closed — the agent timed out.

Fix

Relax the gate and add a dedicated byte-faithful flushing passthrough for the no-responder case:

• WritesBody → buffered fallback (unchanged — a body mutator can't stream).
• HasStreamingResponders → handleStreamingResponse (unchanged — parse + per-frame dispatch for inference/a2a parsers).
• otherwise → new streamPassthrough: copies raw chunks and Flush()es each write, relaying the exact upstream bytes so event: / id: / retry: / comment lines survive. Re-framing through sseframe/writeSSEFrame would drop them, which generic SSE clients like MCP Streamable HTTP depend on.

streamPassthrough runs the response-phase pipeline (RunResponse) before the first byte and honors a deny. This keeps header/status-based response gates working on streamed responses — e.g. opa's response-phase deny and litellm-budgettrack's cost accounting — rather than silently bypassing them. It's safe specifically here: this path is reached only when no StreamingResponder is configured, so RunResponse cannot double-dispatch a plugin that also implements OnResponseFrame (the reason handleStreamingResponse skips it), and every plugin reachable here uses status/headers only, never pctx.ResponseBody. Body-level response inspection on a stream still requires implementing StreamingResponder.

streamPassthrough reuses the existing idleReader wedge-guard, streamReadIdleTimeout, streamFallbackBuffered (no-http.Flusher fallback), and recordOutboundResponseEvent.

Testing

New tests in mcp_sse_stream_test.go:

• TestForwardProxy_SSE_StreamsWithoutResponder — upstream flushes one event then holds the connection open (the shape the existing mcp_sse_repro_test.go misses); asserts the event reaches the client within 2s while the upstream is still blocked, and that event: / id: / data: survive verbatim. Before the fix this times out.
• TestForwardProxy_SSE_ResponsePhaseDenyShortCircuits — a plugin that denies in OnResponse short-circuits before any SSE byte is written (403, no body).
• TestForwardProxy_SSE_HeaderOnResponseRuns — a non-denying header-level OnResponse still runs while the stream is delivered verbatim.

go test ./listener/forwardproxy/... ./listener/internal/sseframe/... all pass, including the unchanged WritesBody guard (writesse_test.go) and responder path (mcp_sse_repro_test.go). go vet ./... (GOWORK=off) and gofmt clean.

Scope / notes

Forward proxy only. The reverse proxy already streams SSE (#657). No pipeline/API changes.

One pre-existing limitation is left as-is: a mixed pipeline that does contain a StreamingResponder (e.g. [opa, inference-parser]) still routes to handleStreamingResponse, which skips RunResponse — so a response-phase opa deny is not enforced there. That predates this PR and applies regardless of #642; aligning both streaming paths would require selective per-plugin response dispatch and is out of scope here.

Assisted-By: Claude (Anthropic AI) noreply@anthropic.com

Summary by CodeRabbit

• New Features

  • SSE responses now stream to clients immediately, even without a streaming responder configured.
  • SSE data is relayed byte-for-byte with timely flushing while the connection remains open.
  • Response-phase processing continues to run before streamed content is delivered.

• Bug Fixes

  • Prevented SSE streams from being buffered indefinitely.
  • Ensured response-phase denials stop SSE output before any bytes are sent.
  • Added safeguards and warnings for incompatible response-body processing.

[coderabbitai]

📝 Walkthrough

Walkthrough

The forward proxy adds incremental SSE passthrough for outbound pipelines without streaming responders. It preserves response-phase hooks, flushes upstream bytes as they arrive, warns about unsupported body-reading plugins, and adds regression coverage for streaming and denial behavior.

Changes

SSE passthrough streaming

Layer / File(s)	Summary
Outbound SSE passthrough authbridge/authlib/listener/forwardproxy/server.go	serveOutbound selects streamPassthrough for eligible SSE responses; the helper runs response hooks, forwards and flushes bytes incrementally, removes Content-Length, and records the response event.
SSE response-phase behavior authbridge/authlib/listener/forwardproxy/mcp_sse_stream_test.go	Tests verify response-phase denial, successful OnResponse execution, verbatim stream delivery, and warnings for ReadsBody plugins without streaming responders.
Immediate SSE delivery regression test authbridge/authlib/listener/forwardproxy/mcp_sse_stream_test.go	A blocked upstream SSE handler and empty outbound pipeline verify that the first complete frame arrives before the upstream connection is released.

Estimated code review effort: 3 (Moderate) | ~25 minutes

Sequence Diagram(s)

sequenceDiagram
  participant UpstreamSSE
  participant serveOutbound
  participant ResponsePlugin
  participant DownstreamClient
  UpstreamSSE->>serveOutbound: Send text/event-stream response
  serveOutbound->>ResponsePlugin: Run OnResponse
  ResponsePlugin-->>serveOutbound: Allow or deny response
  serveOutbound->>DownstreamClient: Write SSE bytes
  serveOutbound->>DownstreamClient: Flush each write
  DownstreamClient-->>serveOutbound: Read streamed frame

Loading

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly summarizes the main change: SSE forwarding without requiring a StreamingResponder.
Linked Issues check	✅ Passed	The code and tests implement the requested immediate SSE passthrough while preserving response-phase handling and buffered behavior where needed.
Out of Scope Changes check	✅ Passed	The changes stay focused on forward-proxy SSE streaming behavior and its tests, with no obvious unrelated additions.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

Warning

There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure.

🔧 golangci-lint (2.12.2)

level=error msg="[linters_context] typechecking error: pattern ./...: directory prefix . does not contain main module or its selected dependencies"

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[pdettori]

A focused, well-reasoned fix for #642: SSE responses over the forward proxy now stream byte-for-byte with per-write flushing when no StreamingResponder is configured, instead of falling through to an unflushed io.Copy.

Verified against source:

• No double-dispatch — streamPassthrough runs RunResponse only on the flushing branch; the no-flusher branch delegates to streamFallbackBuffered, which runs its own RunResponse. Runs exactly once either way.
• Deny recording consistency — a response-phase Reject returns before the deferred recordOutboundResponseEvent, matching the existing buffered path.
• WritesBody fallthrough unchanged (buffered + warning).

Tests are strong: the regression test holds the upstream connection open (the shape the old repro missed), plus deny short-circuit and header-only response-phase coverage.

One non-blocking suggestion inline re: the ReadsBody guard asymmetry. Not a blocker — no current plugin hits it and CI is green.

Approving.

Assisted-By: Claude Code

[pdettori]

suggestion (non-blocking): this branch guards WritesBody() but not ReadsBody(). Since NeedsBody() = ReadsBody || WritesBody, a plugin with ReadsBody:true, WritesBody:false that is not a StreamingResponder now falls into streamPassthrough, and its OnResponse sees an empty pctx.ResponseBody — whereas before this PR that plugin+SSE combination went through the buffered path and saw the full body.

The PR prose documents this invariant ("plugins reachable here do not read pctx.ResponseBody"), but the code doesn't enforce or warn on it, so it's a latent footgun for a future response-body-reading plugin.

Buffering isn't the fix here (it would reintroduce the #642 timeout for that plugin on a truly live stream — such a plugin really should implement StreamingResponder). But a slog.Warn mirroring the existing WritesBody fallback message would surface the misconfiguration instead of silently starving the plugin of the body.

[huang195]

Added the warning in e2314be. Since WritesBody() is already false in this else branch, NeedsBody() here implies a ReadsBody-only, non-StreamingResponder plugin — so the guard is just if s.OutboundPipeline.NeedsBody(), logging a slog.Warn that mirrors the WritesBody fallback message above it. No buffering (that would reintroduce the #642 timeout on a live stream); such a plugin should implement StreamingResponder.

New test TestForwardProxy_SSE_ReadsBodyPluginWarnsAndStreams locks it in: a ReadsBody-only probe on an SSE response still gets the stream verbatim, its OnResponse sees an empty body, and the warning fires.

[huang195]

Author self-review — GitHub blocks self-approval, so this is a COMMENT-event review; treat as LGTM.

Tightly-scoped, well-reasoned fix for #642: forward-proxy SSE responses now stream with per-write flushing when no StreamingResponder is configured, instead of falling through to an unflushed io.Copy.

I verified the load-bearing claims against the full source:

• handleStreamingResponse genuinely skips RunResponse (server.go:570-574) — dispatch is per-frame via RunResponseFrame.
• streamFallbackBuffered runs its own RunResponse, so streamPassthrough delegating to it before running its own RunResponse avoids double-dispatch.
• Skipping the response event on a response-phase deny is consistent across all three paths (buffered, fallback, passthrough), and RunFinish still fires via the early defer.
• The raw header-copy loop matches the file's existing convention (buffered path + handleStreamingResponse do the same).
• Transfer-Encoding isn't in resp.Header for a Go-client response, so Del("Content-Length") is sufficient.

New tests are assertive and cover the real regression shape (upstream holds the connection open past the first flush), the response-phase deny short-circuit, and a non-denying header-level OnResponse. One non-blocking doc suggestion inline.

LGTM. Assisted-By: Claude Code

[huang195]

Good call running RunResponse here so header/status response gates aren't silently bypassed on the passthrough path.

One nuance worth a half-sentence in the doc comment: pctx.ResponseHeaders is a .Clone() of resp.Header (server.go:369), and this path writes headers from resp.Header. So a response-phase plugin that adds a header via pctx.ResponseHeaders would have it dropped — same as the buffered path, so not a regression, and no in-tree plugin does this today (opa / litellm-budgettrack only read status + headers). Since the doc comment cites cost accounting to justify running RunResponse, it'd help future plugin authors to note that only status-based deny + read-only header inspection propagate on this path; header mutations don't survive on any streaming path.

Non-blocking.

[coderabbitai]

🧹 Nitpick comments (1)

authbridge/authlib/listener/forwardproxy/mcp_sse_stream_test.go (1)

231-231: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Prefer strings.NewReader over bytes.NewReader([]byte(...)).

Using strings.NewReader with string literals avoids the unnecessary allocation of converting a string to a byte slice. This is a common and idiomatic optimization in Go tests.

• authbridge/authlib/listener/forwardproxy/mcp_sse_stream_test.go#L231-L231: Replace bytes.NewReader([]byte(...)) with strings.NewReader(...).
• authbridge/authlib/listener/forwardproxy/mcp_sse_stream_test.go#L258-L258: Replace bytes.NewReader([]byte(...)) with strings.NewReader(...).
• authbridge/authlib/listener/forwardproxy/mcp_sse_stream_test.go#L297-L297: Replace bytes.NewReader([]byte(...)) with strings.NewReader(...).

♻️ Proposed refactor for the anchor site

-	req, _ := http.NewRequest("POST", url, bytes.NewReader([]byte(`{"jsonrpc":"2.0","id":1,"method":"tools/list"}`)))
+	req, _ := http.NewRequest("POST", url, strings.NewReader(`{"jsonrpc":"2.0","id":1,"method":"tools/list"}`))

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@authbridge/authlib/listener/forwardproxy/mcp_sse_stream_test.go` at line 231,
Replace bytes.NewReader([]byte(...)) with strings.NewReader(...) for the request
bodies at authbridge/authlib/listener/forwardproxy/mcp_sse_stream_test.go lines
231-231, 258-258, and 297-297, and update imports as needed while preserving the
existing request behavior.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@authbridge/authlib/listener/forwardproxy/mcp_sse_stream_test.go`:
- Line 231: Replace bytes.NewReader([]byte(...)) with strings.NewReader(...) for
the request bodies at
authbridge/authlib/listener/forwardproxy/mcp_sse_stream_test.go lines 231-231,
258-258, and 297-297, and update imports as needed while preserving the existing
request behavior.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: edcb81a3-b46e-42df-9090-defd155b82a1

📥 Commits

Reviewing files that changed from the base of the PR and between ec2fb7e and e2314be.

📒 Files selected for processing (2)

• authbridge/authlib/listener/forwardproxy/mcp_sse_stream_test.go
• authbridge/authlib/listener/forwardproxy/server.go

🚧 Files skipped from review as they are similar to previous changes (1)

• authbridge/authlib/listener/forwardproxy/server.go