From 027b5d5ebaae2f922796be1d272070080093ee55 Mon Sep 17 00:00:00 2001
From: Jerry Yan <792602257@qq.com>
Date: Mon, 4 May 2026 09:35:19 +0800
Subject: [PATCH] fix(infra/llm): anthropic stream goroutine respects ctx
 cancel on send

---
 internal/infra/llm/anthropic.go | 29 +++++++++++++++++++++++------
 1 file changed, 23 insertions(+), 6 deletions(-)

diff --git a/internal/infra/llm/anthropic.go b/internal/infra/llm/anthropic.go
index bde46f7..2cf4e71 100644
--- a/internal/infra/llm/anthropic.go
+++ b/internal/infra/llm/anthropic.go
@@ -77,6 +77,15 @@ func (c *AnthropicClient) Stream(ctx context.Context, modelID string, req Reques
 	stream := c.sdk.Messages.NewStreaming(ctx, params)
 	out := make(chan StreamEvent, 16)
 
+	// send 在 ctx 取消时立即放弃发送，防止下游 reader 已退出导致 goroutine 永久阻塞。
+	send := func(ev StreamEvent) bool {
+		select {
+		case <-ctx.Done():
+			return false
+		case out <- ev:
+			return true
+		}
+	}
 	go func() {
 		defer close(out)
 		var inputTokens int
@@ -93,25 +102,33 @@ func (c *AnthropicClient) Stream(ctx context.Context, modelID string, req Reques
 				switch ev.Delta.Type {
 				case "text_delta":
 					if text := ev.Delta.AsTextDelta().Text; text != "" {
-						out <- StreamEvent{Type: EventTextDelta, Text: text}
+						if !send(StreamEvent{Type: EventTextDelta, Text: text}) {
+							return
+						}
 					}
 				case "thinking_delta":
 					if thinking := ev.Delta.AsThinkingDelta().Thinking; thinking != "" {
-						out <- StreamEvent{Type: EventThinkingDelta, Text: thinking}
+						if !send(StreamEvent{Type: EventThinkingDelta, Text: thinking}) {
+							return
+						}
 					}
 				}
 			case "message_delta":
 				ev := event.AsMessageDelta()
-				out <- StreamEvent{Type: EventUsage, Usage: &Usage{
+				if !send(StreamEvent{Type: EventUsage, Usage: &Usage{
 					PromptTokens:     inputTokens,
 					CompletionTokens: int(ev.Usage.OutputTokens),
-				}}
+				}}) {
+					return
+				}
 			case "message_stop":
-				out <- StreamEvent{Type: EventDone, Reason: "end_turn"}
+				if !send(StreamEvent{Type: EventDone, Reason: "end_turn"}) {
+					return
+				}
 			}
 		}
 		if err := stream.Err(); err != nil {
-			out <- StreamEvent{Type: EventError, Err: fmt.Errorf("anthropic stream: %w", err)}
+			send(StreamEvent{Type: EventError, Err: fmt.Errorf("anthropic stream: %w", err)})
 		}
 	}()
 	return out, nil