fix(chat): NewStreamerHub accepts ConversationService interface; SSE handler drains ready events before ctx

This commit is contained in:
2026-05-04 19:32:24 +08:00
parent d3ef9c941b
commit 7aa8456b1b
2 changed files with 17 additions and 4 deletions
+6 -2
View File
@@ -183,8 +183,13 @@ func TestStreamMessages_StopsWhenClientClosesContext(t *testing.T) {
r.ServeHTTP(w, req) 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"} 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() cancel()
// Handler must return promptly after context cancellation. // 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") 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") assert.Contains(t, w.Body.String(), "text_delta")
} }
+11 -2
View File
@@ -107,23 +107,32 @@ type streamerHub struct {
} }
// NewStreamerHub constructs a StreamerHub. // 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( func NewStreamerHub(
repo Repository, repo Repository,
registry clientGetter, registry clientGetter,
dispatch notifyDispatcher, dispatch notifyDispatcher,
auditor audit.Recorder, auditor audit.Recorder,
convSvc *conversationService, convSvc ConversationService,
log *slog.Logger, log *slog.Logger,
retention time.Duration, retention time.Duration,
safetyPct float64, safetyPct float64,
) StreamerHub { ) StreamerHub {
var concrete *conversationService
if convSvc != nil {
if c, ok := convSvc.(*conversationService); ok {
concrete = c
}
}
return &streamerHub{ return &streamerHub{
byMsgID: make(map[int64]*streamer), byMsgID: make(map[int64]*streamer),
repo: repo, repo: repo,
registry: registry, registry: registry,
dispatch: dispatch, dispatch: dispatch,
auditor: auditor, auditor: auditor,
convSvc: convSvc, convSvc: concrete,
log: log, log: log,
retention: retention, retention: retention,
safety: safetyPct, safety: safetyPct,