From 7aa8456b1b9f6935920bd1396d14728be5e0732b Mon Sep 17 00:00:00 2001
From: Jerry Yan <792602257@qq.com>
Date: Mon, 4 May 2026 19:32:24 +0800
Subject: [PATCH] fix(chat): NewStreamerHub accepts ConversationService
 interface; SSE handler drains ready events before ctx

---
 internal/chat/sse_test.go |  8 ++++++--
 internal/chat/streamer.go | 13 +++++++++++--
 2 files changed, 17 insertions(+), 4 deletions(-)

diff --git a/internal/chat/sse_test.go b/internal/chat/sse_test.go
index 928d29e..f1a8398 100644
--- a/internal/chat/sse_test.go
+++ b/internal/chat/sse_test.go
@@ -183,8 +183,13 @@ func TestStreamMessages_StopsWhenClientClosesContext(t *testing.T) {
 		r.ServeHTTP(w, req)
 	}()
 
-	// Send one event, then cancel.
+	// Send one event and give the handler a moment to drain it before cancelling,
+	// so the body assertion is deterministic. The cancel itself is what we're
+	// verifying — that the handler exits even though the channel never closes.
 	ch <- SSEEvent{ID: 1, Event: "text_delta", Data: "hi"}
+	require.Eventually(t, func() bool {
+		return strings.Contains(w.Body.String(), "text_delta")
+	}, time.Second, 10*time.Millisecond, "event was not flushed before cancel")
 	cancel()
 
 	// Handler must return promptly after context cancellation.
@@ -195,7 +200,6 @@ func TestStreamMessages_StopsWhenClientClosesContext(t *testing.T) {
 		t.Fatal("handler did not stop after context cancellation")
 	}
 
-	// The one event before cancel should have been written.
 	assert.Contains(t, w.Body.String(), "text_delta")
 }
 
diff --git a/internal/chat/streamer.go b/internal/chat/streamer.go
index 6d76f39..2d7a1b4 100644
--- a/internal/chat/streamer.go
+++ b/internal/chat/streamer.go
@@ -107,23 +107,32 @@ type streamerHub struct {
 }
 
 // NewStreamerHub constructs a StreamerHub.
+// convSvc must be the concrete *conversationService produced by NewConversationService;
+// the type assertion lets callers in other packages pass the interface without needing
+// to import the unexported concrete type. Pass nil to skip title generation.
 func NewStreamerHub(
 	repo Repository,
 	registry clientGetter,
 	dispatch notifyDispatcher,
 	auditor audit.Recorder,
-	convSvc *conversationService,
+	convSvc ConversationService,
 	log *slog.Logger,
 	retention time.Duration,
 	safetyPct float64,
 ) StreamerHub {
+	var concrete *conversationService
+	if convSvc != nil {
+		if c, ok := convSvc.(*conversationService); ok {
+			concrete = c
+		}
+	}
 	return &streamerHub{
 		byMsgID:   make(map[int64]*streamer),
 		repo:      repo,
 		registry:  registry,
 		dispatch:  dispatch,
 		auditor:   auditor,
-		convSvc:   convSvc,
+		convSvc:   concrete,
 		log:       log,
 		retention: retention,
 		safety:    safetyPct,