From d8317d876fa090731db62201993af67f0e4ea5f3 Mon Sep 17 00:00:00 2001
From: ginuerzh <ginuerzh@gmail.com>
Date: Tue, 30 Jun 2026 20:46:12 +0800
Subject: [PATCH] fix(forwarder): append SSE \n\n terminator after end phase
 data, guard against double delim

eventsource_stream requires \n\n to release each SSE event. The end phase
(EOF and scanner-error paths) wrote rewritten data without the terminator,
so response.completed was buffered and never dispatched. Adds writeSSEStreamDelim
helper which only appends \n\n when the data doesn't already end with it.
Adds scanner error end phase path that was missing entirely.
---
 internal/util/forwarder/sniffer_sse.go  | 38 +++++++++++++++++++++++--
 internal/util/forwarder/sniffer_test.go |  4 +--
 2 files changed, 38 insertions(+), 4 deletions(-)

diff --git a/internal/util/forwarder/sniffer_sse.go b/internal/util/forwarder/sniffer_sse.go
index 8f233af..79e0a0d 100644
--- a/internal/util/forwarder/sniffer_sse.go
+++ b/internal/util/forwarder/sniffer_sse.go
@@ -69,6 +69,7 @@ type rewriteBody struct {
 	// SSE stream lifecycle state.
 	eventIndex int  // incremented per SSE event for the "event" phase
 	ended      bool // true after the stream-end phase has been emitted
+	scannerErr error // non-nil when scanner terminated with an error
 
 	sid string // session ID, cached from context at construction time
 }
@@ -290,6 +291,9 @@ func (b *rewriteBody) Read(p []byte) (n int, err error) {
 			err = b.scanner.Err()
 			if err == nil {
 				if b.ended {
+					if b.scannerErr != nil {
+						return 0, b.scannerErr
+					}
 					return 0, io.EOF
 				}
 				// End of upstream SSE events — emit stream end phase.
@@ -305,9 +309,32 @@ func (b *rewriteBody) Read(p []byte) (n int, err error) {
 				if endErr != nil {
 					return 0, endErr
 				}
-				b.buf.Write(rewritten)
+				if len(rewritten) > 0 {
+					b.buf.Write(rewritten)
+					writeSSEStreamDelim(&b.buf, rewritten)
+				}
 				return b.buf.Read(p)
 			}
+			// Scanner error (e.g. upstream connection reset): emit the end
+			// phase so Rewriter plugins can send a terminal event
+			// (response.failed / message_stop), then return the error
+			// after the buffer drains.
+			if !b.ended {
+				b.ended = true
+				b.scannerErr = err
+				md := map[string]any{
+					"sid":          b.sid,
+					"sse_phase":    "end",
+					"event_index":  b.eventIndex,
+					"stream_error": err.Error(),
+				}
+				rewritten, _ := b.apply(nil, rewriter.MetadataRewriteOption(md))
+				if len(rewritten) > 0 {
+					b.buf.Write(rewritten)
+					writeSSEStreamDelim(&b.buf, rewritten)
+					return b.buf.Read(p)
+				}
+			}
 			return 0, err
 		}
 
@@ -336,7 +363,7 @@ func (b *rewriteBody) Read(p []byte) (n int, err error) {
 		}
 		if len(rewritten) > 0 {
 			b.buf.Write(rewritten)
-			b.buf.Write([]byte("\n\n"))
+			writeSSEStreamDelim(&b.buf, rewritten)
 		}
 		// Loop to flush the buffer or scan the next event.
 	}
@@ -357,6 +384,13 @@ func sidMetadata(sid string) []rewriter.RewriteOption {
 	}
 }
 
+// ponytail: guard against plugin already having added \n\n delimiter
+func writeSSEStreamDelim(buf *bytes.Buffer, data []byte) {
+	if len(data) < 2 || data[len(data)-1] != '\n' || data[len(data)-2] != '\n' {
+		buf.Write([]byte("\n\n"))
+	}
+}
+
 func decompressBody(data []byte, encoding string) ([]byte, error) {
 	switch encoding {
 	case "gzip":
diff --git a/internal/util/forwarder/sniffer_test.go b/internal/util/forwarder/sniffer_test.go
index 85216ad..e64a889 100644
--- a/internal/util/forwarder/sniffer_test.go
+++ b/internal/util/forwarder/sniffer_test.go
@@ -1895,8 +1895,8 @@ func TestNewRewriteBody(t *testing.T) {
 
 		got, _ := io.ReadAll(body)
 		body.Close()
-		if string(got) != "data: rewritten\n\ndata: rewritten" {
-			t.Errorf("body = %q, want %q", string(got), "data: rewritten\n\ndata: rewritten")
+		if string(got) != "data: rewritten\n\ndata: rewritten\n\n" {
+			t.Errorf("body = %q, want %q", string(got), "data: rewritten\n\ndata: rewritten\n\n")
 		}
 	})