update

Signed-off-by: yxia216 <[email protected]>

Author: hustxiayang

cmd/aigw/run.go

@@ -136,14 +136,15 @@ func run(ctx context.Context, c cmdRun, o *runOpts, stdout, stderr io.Writer) er
 	// Do the translation of the given AI Gateway resources Yaml into Envoy Gateway resources and write them to the file.
 	resourcesBuf := &bytes.Buffer{}
 	runCtx := &runCmdContext{
-		isDebug:                  c.Debug,
-		envoyGatewayResourcesOut: resourcesBuf,
-		stderrLogger:             debugLogger,
-		stderr:                   stderr,
-		tmpdir:                   filepath.Dir(o.logPath), // runDir
-		udsPath:                  o.extprocUDSPath,
-		adminPort:                c.AdminPort,
-		extProcLauncher:          o.extProcLauncher,
+		isDebug:                        c.Debug,
+		envoyGatewayResourcesOut:       resourcesBuf,
+		stderrLogger:                   debugLogger,
+		stderr:                         stderr,
+		tmpdir:                         filepath.Dir(o.logPath), // runDir
+		udsPath:                        o.extprocUDSPath,
+		adminPort:                      c.AdminPort,
+		extProcLauncher:                o.extProcLauncher,
+		mcpSessionEncryptionIterations: c.MCPSessionEncryptionIterations,
 	}
 	// If any of the configured MCP servers is using stdio, set up the streamable HTTP proxies for them
 	if err = proxyStdioMCPServers(ctx, debugLogger, c.mcpConfig); err != nil {

internal/translator/openai_gcpanthropic.go

@@ -22,6 +22,7 @@ import (
 	openAIconstant "github.com/openai/openai-go/shared/constant"
 	"github.com/tidwall/sjson"
 
+	"github.com/envoyproxy/ai-gateway/internal/apischema/awsbedrock"
 	"github.com/envoyproxy/ai-gateway/internal/apischema/openai"
 	"github.com/envoyproxy/ai-gateway/internal/internalapi"
 	"github.com/envoyproxy/ai-gateway/internal/metrics"
@@ -859,15 +860,43 @@ func (o *openAIToGCPAnthropicTranslatorV1ChatCompletion) ResponseBody(_ map[stri
 
 	for i := range anthropicResp.Content { // NOTE: Content structure is massive, do not range over values.
 		output := &anthropicResp.Content[i]
-		if output.Type == string(constant.ValueOf[constant.ToolUse]()) && output.ID != "" {
-			toolCalls, toolErr := anthropicToolUseToOpenAICalls(output)
-			if toolErr != nil {
-				return nil, nil, metrics.TokenUsage{}, "", fmt.Errorf("failed to convert anthropic tool use to openai tool call: %w", toolErr)
+		switch output.Type {
+		case string(constant.ValueOf[constant.ToolUse]()):
+			if output.ID != "" {
+				toolCalls, toolErr := anthropicToolUseToOpenAICalls(output)
+				if toolErr != nil {
+					return nil, nil, metrics.TokenUsage{}, "", fmt.Errorf("failed to convert anthropic tool use to openai tool call: %w", toolErr)
+				}
+				choice.Message.ToolCalls = append(choice.Message.ToolCalls, toolCalls...)
+			}
+		case string(constant.ValueOf[constant.Text]()):
+			if output.Text != "" {
+				if choice.Message.Content == nil {
+					choice.Message.Content = &output.Text
+				}
 			}
-			choice.Message.ToolCalls = append(choice.Message.ToolCalls, toolCalls...)
-		} else if output.Type == string(constant.ValueOf[constant.Text]()) && output.Text != "" {
-			if choice.Message.Content == nil {
-				choice.Message.Content = &output.Text
+		case string(constant.ValueOf[constant.Thinking]()):
+			if output.Thinking != "" {
+				choice.Message.ReasoningContent = &openai.ReasoningContentUnion{
+					Value: &openai.ReasoningContent{
+						ReasoningContent: &awsbedrock.ReasoningContentBlock{
+							ReasoningText: &awsbedrock.ReasoningTextBlock{
+								Text:      output.Thinking,
+								Signature: output.Signature,
+							},
+						},
+					},
+				}
+			}
+		case string(constant.ValueOf[constant.RedactedThinking]()):
+			if output.Data != "" {
+				choice.Message.ReasoningContent = &openai.ReasoningContentUnion{
+					Value: &openai.ReasoningContent{
+						ReasoningContent: &awsbedrock.ReasoningContentBlock{
+							RedactedContent: []byte(output.Data),
+						},
+					},
+				}
 			}
 		}
 	}

internal/translator/openai_gcpanthropic_stream.go

@@ -14,18 +14,14 @@ import (
 
 	"github.com/anthropics/anthropic-sdk-go"
 	"github.com/anthropics/anthropic-sdk-go/shared/constant"
-	"k8s.io/utils/ptr"
 
 	"github.com/envoyproxy/ai-gateway/internal/apischema/openai"
 	"github.com/envoyproxy/ai-gateway/internal/internalapi"
 	"github.com/envoyproxy/ai-gateway/internal/metrics"
 	tracing "github.com/envoyproxy/ai-gateway/internal/tracing/api"
 )
 
-var (
-	sseEventPrefix = []byte("event:")
-	emptyStrPtr    = ptr.To("")
-)
+var sseEventPrefix = []byte("event:")
 
 // streamingToolCall holds the state for a single tool call that is being streamed.
 type streamingToolCall struct {
@@ -265,16 +261,7 @@ func (p *anthropicStreamParser) handleAnthropicStreamEvent(eventType []byte, dat
 			}
 			return p.constructOpenAIChatCompletionChunk(delta, ""), nil
 		}
-		if event.ContentBlock.Type == string(constant.ValueOf[constant.Thinking]()) {
-			delta := openai.ChatCompletionResponseChunkChoiceDelta{Content: emptyStrPtr}
-			return p.constructOpenAIChatCompletionChunk(delta, ""), nil
-		}
-
-		if event.ContentBlock.Type == string(constant.ValueOf[constant.RedactedThinking]()) {
-			// This is a latency-hiding event, ignore it.
-			return nil, nil
-		}
-
+		// do not need to return an empty str for thinking start block
 		return nil, nil
 
 	case string(constant.ValueOf[constant.MessageDelta]()):
@@ -304,10 +291,28 @@ func (p *anthropicStreamParser) handleAnthropicStreamEvent(eventType []byte, dat
 			return nil, fmt.Errorf("unmarshal content_block_delta: %w", err)
 		}
 		switch event.Delta.Type {
-		case string(constant.ValueOf[constant.TextDelta]()), string(constant.ValueOf[constant.ThinkingDelta]()):
-			// Treat thinking_delta just like a text_delta.
+		case string(constant.ValueOf[constant.TextDelta]()):
 			delta := openai.ChatCompletionResponseChunkChoiceDelta{Content: &event.Delta.Text}
 			return p.constructOpenAIChatCompletionChunk(delta, ""), nil
+
+		case string(constant.ValueOf[constant.ThinkingDelta]()):
+			// this should already include the case for redacted thinking: https://platform.claude.com/docs/en/build-with-claude/streaming#content-block-delta-types
+
+			reasoningDelta := &openai.StreamReasoningContent{}
+
+			// Map all relevant fields from the Bedrock delta to our flattened OpenAI delta struct.
+			if event.Delta.Thinking != "" {
+				reasoningDelta.Text = event.Delta.Thinking
+			}
+			if event.Delta.Signature != "" {
+				reasoningDelta.Signature = event.Delta.Signature
+			}
+
+			delta := openai.ChatCompletionResponseChunkChoiceDelta{
+				ReasoningContent: reasoningDelta,
+			}
+			return p.constructOpenAIChatCompletionChunk(delta, ""), nil
+
 		case string(constant.ValueOf[constant.InputJSONDelta]()):
 			tool, ok := p.activeToolCalls[p.toolIndex]
 			if !ok {
@@ -326,6 +331,7 @@ func (p *anthropicStreamParser) handleAnthropicStreamEvent(eventType []byte, dat
 			tool.inputJSON += event.Delta.PartialJSON
 			return p.constructOpenAIChatCompletionChunk(delta, ""), nil
 		}
+		// Do not process redacted thinking stream? Did not find the source
 
 	case string(constant.ValueOf[constant.ContentBlockStop]()):
 		// This event is for state cleanup, no chunk is sent.

internal/translator/openai_gcpanthropic_stream_test.go

@@ -539,7 +539,7 @@ event: content_block_start
 data: {"type": "content_block_start", "index": 0, "content_block": {"type": "thinking", "name": "web_searcher"}}
 
 event: content_block_delta
-data: {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "text": "Searching for information..."}}
+data: {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "Searching for information..."}}
 
 event: content_block_stop
 data: {"type": "content_block_stop", "index": 0}
@@ -564,6 +564,7 @@ data: {"type": "message_stop"}
 		bodyStr := string(bm)
 
 		var contentDeltas []string
+		var reasoningTexts []string
 		var foundToolCallWithArgs bool
 		var finalFinishReason openai.ChatCompletionChoicesFinishReason
 
@@ -586,6 +587,11 @@ data: {"type": "message_stop"}
 				if choice.Delta.Content != nil {
 					contentDeltas = append(contentDeltas, *choice.Delta.Content)
 				}
+				if choice.Delta.ReasoningContent != nil {
+					if choice.Delta.ReasoningContent.Text != "" {
+						reasoningTexts = append(reasoningTexts, choice.Delta.ReasoningContent.Text)
+					}
+				}
 				if len(choice.Delta.ToolCalls) > 0 {
 					toolCall := choice.Delta.ToolCalls[0]
 					// Check if this is the tool chunk that contains the arguments.
@@ -607,11 +613,155 @@ data: {"type": "message_stop"}
 			}
 		}
 
-		fullContent := strings.Join(contentDeltas, "")
-		assert.Contains(t, fullContent, "Searching for information...")
+		fullReasoning := strings.Join(reasoningTexts, "")
+
+		assert.Contains(t, fullReasoning, "Searching for information...")
 		require.True(t, foundToolCallWithArgs, "Did not find a tool call chunk with arguments to assert against")
 		assert.Equal(t, openai.ChatCompletionChoicesFinishReasonToolCalls, finalFinishReason, "Final finish reason should be 'tool_calls'")
 	})
+
+	t.Run("handles thinking delta stream with text only", func(t *testing.T) {
+		sseStream := `
+event: message_start
+data: {"type": "message_start", "message": {"id": "msg_thinking_1", "type": "message", "role": "assistant", "usage": {"input_tokens": 20, "output_tokens": 1}}}
+
+event: content_block_start
+data: {"type": "content_block_start", "index": 0, "content_block": {"type": "thinking"}}
+
+event: content_block_delta
+data: {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "Let me think about this problem step by step."}}
+
+event: content_block_delta
+data: {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": " First, I need to understand the requirements."}}
+
+event: content_block_stop
+data: {"type": "content_block_stop", "index": 0}
+
+event: message_delta
+data: {"type": "message_delta", "delta": {"stop_reason": "end_turn"}, "usage": {"output_tokens": 15}}
+
+event: message_stop
+data: {"type": "message_stop"}
+`
+		openAIReq := &openai.ChatCompletionRequest{Stream: true, Model: "test-model", MaxTokens: new(int64)}
+		translator := NewChatCompletionOpenAIToGCPAnthropicTranslator("", "").(*openAIToGCPAnthropicTranslatorV1ChatCompletion)
+		_, _, err := translator.RequestBody(nil, openAIReq, false)
+		require.NoError(t, err)
+
+		_, bm, _, _, err := translator.ResponseBody(map[string]string{}, strings.NewReader(sseStream), true, nil)
+		require.NoError(t, err)
+		require.NotNil(t, bm)
+		bodyStr := string(bm)
+
+		var reasoningTexts []string
+		var foundFinishReason bool
+
+		lines := strings.SplitSeq(strings.TrimSpace(bodyStr), "\n\n")
+		for line := range lines {
+			if !strings.HasPrefix(line, "data: ") || strings.Contains(line, "[DONE]") {
+				continue
+			}
+			jsonBody := strings.TrimPrefix(line, "data: ")
+
+			var chunk openai.ChatCompletionResponseChunk
+			err = json.Unmarshal([]byte(jsonBody), &chunk)
+			require.NoError(t, err, "Failed to unmarshal chunk: %s", jsonBody)
+
+			if len(chunk.Choices) == 0 {
+				continue
+			}
+			choice := chunk.Choices[0]
+			if choice.Delta != nil && choice.Delta.ReasoningContent != nil {
+				if choice.Delta.ReasoningContent.Text != "" {
+					reasoningTexts = append(reasoningTexts, choice.Delta.ReasoningContent.Text)
+				}
+			}
+			if choice.FinishReason == openai.ChatCompletionChoicesFinishReasonStop {
+				foundFinishReason = true
+			}
+		}
+
+		fullReasoning := strings.Join(reasoningTexts, "")
+		assert.Contains(t, fullReasoning, "Let me think about this problem step by step.")
+		assert.Contains(t, fullReasoning, " First, I need to understand the requirements.")
+		require.True(t, foundFinishReason, "Should find stop finish reason")
+	})
+
+	t.Run("handles thinking delta stream with text and signature", func(t *testing.T) {
+		sseStream := `
+event: message_start
+data: {"type": "message_start", "message": {"id": "msg_thinking_2", "type": "message", "role": "assistant", "usage": {"input_tokens": 25, "output_tokens": 1}}}
+
+event: content_block_start
+data: {"type": "content_block_start", "index": 0, "content_block": {"type": "thinking"}}
+
+event: content_block_delta
+data: {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "Processing request...", "signature": "sig_abc123"}}
+
+event: content_block_delta
+data: {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": " Analyzing data...", "signature": "sig_def456"}}
+
+event: content_block_stop
+data: {"type": "content_block_stop", "index": 0}
+
+event: message_delta
+data: {"type": "message_delta", "delta": {"stop_reason": "end_turn"}, "usage": {"output_tokens": 20}}
+
+event: message_stop
+data: {"type": "message_stop"}
+`
+		openAIReq := &openai.ChatCompletionRequest{Stream: true, Model: "test-model", MaxTokens: new(int64)}
+		translator := NewChatCompletionOpenAIToGCPAnthropicTranslator("", "").(*openAIToGCPAnthropicTranslatorV1ChatCompletion)
+		_, _, err := translator.RequestBody(nil, openAIReq, false)
+		require.NoError(t, err)
+
+		_, bm, _, _, err := translator.ResponseBody(map[string]string{}, strings.NewReader(sseStream), true, nil)
+		require.NoError(t, err)
+		require.NotNil(t, bm)
+		bodyStr := string(bm)
+
+		var reasoningTexts []string
+		var signatures []string
+		var foundFinishReason bool
+
+		lines := strings.SplitSeq(strings.TrimSpace(bodyStr), "\n\n")
+		for line := range lines {
+			if !strings.HasPrefix(line, "data: ") || strings.Contains(line, "[DONE]") {
+				continue
+			}
+			jsonBody := strings.TrimPrefix(line, "data: ")
+
+			var chunk openai.ChatCompletionResponseChunk
+			err = json.Unmarshal([]byte(jsonBody), &chunk)
+			require.NoError(t, err, "Failed to unmarshal chunk: %s", jsonBody)
+
+			if len(chunk.Choices) == 0 {
+				continue
+			}
+			choice := chunk.Choices[0]
+			if choice.Delta != nil && choice.Delta.ReasoningContent != nil {
+				if choice.Delta.ReasoningContent.Text != "" {
+					reasoningTexts = append(reasoningTexts, choice.Delta.ReasoningContent.Text)
+				}
+				if choice.Delta.ReasoningContent.Signature != "" {
+					signatures = append(signatures, choice.Delta.ReasoningContent.Signature)
+				}
+			}
+			if choice.FinishReason == openai.ChatCompletionChoicesFinishReasonStop {
+				foundFinishReason = true
+			}
+		}
+
+		fullReasoning := strings.Join(reasoningTexts, "")
+		assert.Contains(t, fullReasoning, "Processing request...")
+		assert.Contains(t, fullReasoning, " Analyzing data...")
+
+		allSignatures := strings.Join(signatures, ",")
+		assert.Contains(t, allSignatures, "sig_abc123")
+		assert.Contains(t, allSignatures, "sig_def456")
+
+		require.True(t, foundFinishReason, "Should find stop finish reason")
+	})
 }
 
 func TestAnthropicStreamParser_EventTypes(t *testing.T) {