Merge pull request #4142 from fengshao1227/fix/streaming-stop-reason-and-content-filter

fix(apicompat): 流式 Anthropic max_tokens 截断语义 + content_filter finish_reason

backend/internal/pkg/apicompat/anthropic_to_responses_response.go

@@ -164,6 +164,8 @@ type AnthropicEventToResponsesState struct {
 	OutputTokens             int
 	CacheReadInputTokens     int
 	CacheCreationInputTokens int
+
+	StopReason string
 }
 
 // NewAnthropicEventToResponsesState returns an initialised stream state.
@@ -405,7 +407,6 @@ func anthToResHandleContentBlockStop(evt *AnthropicStreamEvent, state *Anthropic
 }
 
 func anthToResHandleMessageDelta(evt *AnthropicStreamEvent, state *AnthropicEventToResponsesState) []ResponsesStreamEvent {
-	// Update usage
 	if evt.Usage != nil {
 		state.OutputTokens = evt.Usage.OutputTokens
 		if evt.Usage.InputTokens > 0 {
@@ -418,6 +419,9 @@ func anthToResHandleMessageDelta(evt *AnthropicStreamEvent, state *AnthropicEven
 			state.CacheCreationInputTokens = evt.Usage.CacheCreationInputTokens
 		}
 	}
+	if evt.Delta != nil && evt.Delta.StopReason != "" {
+		state.StopReason = evt.Delta.StopReason
+	}
 
 	return nil
 }
@@ -428,15 +432,15 @@ func anthToResHandleMessageStop(state *AnthropicEventToResponsesState) []Respons
 	}
 
 	var events []ResponsesStreamEvent
-
-	// Close any open item
 	events = append(events, closeCurrentResponsesItem(state)...)
 
-	// Determine status
 	status := "completed"
 	var incompleteDetails *ResponsesIncompleteDetails
+	if state.StopReason == "max_tokens" {
+		status = "incomplete"
+		incompleteDetails = &ResponsesIncompleteDetails{Reason: "max_output_tokens"}
+	}
 
-	// Emit response.completed
 	events = append(events, makeResponsesCompletedEvent(state, status, incompleteDetails))
 	state.CompletedSent = true
 	return events
@@ -509,15 +513,20 @@ func makeResponsesCompletedEvent(
 		}
 	}
 
+	eventType := "response.completed"
+	if status == "incomplete" {
+		eventType = "response.incomplete"
+	}
+
 	return ResponsesStreamEvent{
-		Type:           "response.completed",
+		Type:           eventType,
 		SequenceNumber: seq,
 		Response: &ResponsesResponse{
 			ID:                state.ResponseID,
 			Object:            "response",
 			Model:             state.Model,
 			Status:            status,
-			Output:            []ResponsesOutput{}, // Simplified; full output tracking would add complexity
+			Output:            []ResponsesOutput{},
 			Usage:             usage,
 			IncompleteDetails: incompleteDetails,
 		},

backend/internal/pkg/apicompat/responses_to_chatcompletions.go

@@ -89,8 +89,13 @@ func ResponsesToChatCompletions(resp *ResponsesResponse, model string) *ChatComp
 func responsesStatusToChatFinishReason(status string, details *ResponsesIncompleteDetails, toolCalls []ChatToolCall) string {
 	switch status {
 	case "incomplete":
-		if details != nil && details.Reason == "max_output_tokens" {
-			return "length"
+		if details != nil {
+			switch details.Reason {
+			case "max_output_tokens":
+				return "length"
+			case "content_filter":
+				return "content_filter"
+			}
 		}
 		return "stop"
 	case "completed":
@@ -299,8 +304,13 @@ func resToChatHandleCompleted(evt *ResponsesStreamEvent, state *ResponsesEventTo
 
 		switch evt.Response.Status {
 		case "incomplete":
-			if evt.Response.IncompleteDetails != nil && evt.Response.IncompleteDetails.Reason == "max_output_tokens" {
-				finishReason = "length"
+			if evt.Response.IncompleteDetails != nil {
+				switch evt.Response.IncompleteDetails.Reason {
+				case "max_output_tokens":
+					finishReason = "length"
+				case "content_filter":
+					finishReason = "content_filter"
+				}
 			}
 		case "completed":
 			if state.SawToolCall {

backend/internal/pkg/apicompat/streaming_stop_reason_test.go

@@ -0,0 +1,122 @@
+package apicompat
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAnthropicStreamingMaxTokens_MapsToIncomplete(t *testing.T) {
+	state := NewAnthropicEventToResponsesState()
+
+	AnthropicEventToResponsesEvents(&AnthropicStreamEvent{
+		Type:    "message_start",
+		Message: &AnthropicResponse{ID: "msg_test", Model: "claude-opus-4-6", Role: "assistant"},
+	}, state)
+
+	AnthropicEventToResponsesEvents(&AnthropicStreamEvent{
+		Type: "message_delta",
+		Delta: &AnthropicDelta{
+			StopReason: "max_tokens",
+		},
+		Usage: &AnthropicUsage{OutputTokens: 4096},
+	}, state)
+
+	require.Equal(t, "max_tokens", state.StopReason)
+
+	events := AnthropicEventToResponsesEvents(&AnthropicStreamEvent{
+		Type: "message_stop",
+	}, state)
+
+	var completed *ResponsesStreamEvent
+	for i := range events {
+		if events[i].Type == "response.completed" || events[i].Type == "response.incomplete" {
+			completed = &events[i]
+			break
+		}
+	}
+	require.NotNil(t, completed, "should have terminal event")
+	assert.Equal(t, "response.incomplete", completed.Type)
+	require.NotNil(t, completed.Response)
+	assert.Equal(t, "incomplete", completed.Response.Status)
+	require.NotNil(t, completed.Response.IncompleteDetails)
+	assert.Equal(t, "max_output_tokens", completed.Response.IncompleteDetails.Reason)
+}
+
+func TestAnthropicStreamingEndTurn_MapsToCompleted(t *testing.T) {
+	state := NewAnthropicEventToResponsesState()
+
+	AnthropicEventToResponsesEvents(&AnthropicStreamEvent{
+		Type:    "message_start",
+		Message: &AnthropicResponse{ID: "msg_test", Model: "claude-opus-4-6", Role: "assistant"},
+	}, state)
+
+	AnthropicEventToResponsesEvents(&AnthropicStreamEvent{
+		Type:  "message_delta",
+		Delta: &AnthropicDelta{StopReason: "end_turn"},
+		Usage: &AnthropicUsage{OutputTokens: 100},
+	}, state)
+
+	events := AnthropicEventToResponsesEvents(&AnthropicStreamEvent{
+		Type: "message_stop",
+	}, state)
+
+	var completed *ResponsesStreamEvent
+	for i := range events {
+		if events[i].Type == "response.completed" {
+			completed = &events[i]
+			break
+		}
+	}
+	require.NotNil(t, completed)
+	assert.Equal(t, "completed", completed.Response.Status)
+	assert.Nil(t, completed.Response.IncompleteDetails)
+}
+
+func TestResponsesToChatCompletions_ContentFilter(t *testing.T) {
+	resp := &ResponsesResponse{
+		ID:     "resp_cf",
+		Status: "incomplete",
+		IncompleteDetails: &ResponsesIncompleteDetails{
+			Reason: "content_filter",
+		},
+		Output: []ResponsesOutput{{
+			Type:    "message",
+			Content: []ResponsesContentPart{{Type: "output_text", Text: "partial"}},
+		}},
+		Usage: &ResponsesUsage{InputTokens: 10, OutputTokens: 5},
+	}
+
+	cc := ResponsesToChatCompletions(resp, "gpt-5.5")
+	require.Len(t, cc.Choices, 1)
+	assert.Equal(t, "content_filter", cc.Choices[0].FinishReason)
+}
+
+func TestResponsesToChatCompletionsStreaming_ContentFilter(t *testing.T) {
+	state := NewResponsesEventToChatState()
+	state.ID = "resp_cf"
+	state.Model = "gpt-5.5"
+	state.SentRole = true
+
+	events := ResponsesEventToChatChunks(&ResponsesStreamEvent{
+		Type: "response.completed",
+		Response: &ResponsesResponse{
+			ID:     "resp_cf",
+			Status: "incomplete",
+			IncompleteDetails: &ResponsesIncompleteDetails{
+				Reason: "content_filter",
+			},
+		},
+	}, state)
+
+	hasContentFilter := false
+	for _, chunk := range events {
+		for _, choice := range chunk.Choices {
+			if choice.FinishReason != nil && *choice.FinishReason == "content_filter" {
+				hasContentFilter = true
+			}
+		}
+	}
+	assert.True(t, hasContentFilter, "streaming content_filter should map to finish_reason content_filter")
+}