Merge branch 'main' into fix/tolerate-truncated-gzip-eof

Author: ChrisJBurns

internal/extproc/processor_impl.go

@@ -267,12 +267,10 @@ func (r *routerProcessor[ReqT, RespT, RespChunkT, EndpointSpecT]) ProcessRequest
 		Header: &corev3.HeaderValue{Key: internalapi.ModelNameHeaderKeyDefault, RawValue: []byte(originalModel)},
 	})
 	originalPath := r.requestHeaders[":path"]
-	if r.requestHeaders[originalPathHeader] == "" {
-		r.requestHeaders[originalPathHeader] = originalPath
-		additionalHeaders = append(additionalHeaders, &corev3.HeaderValueOption{
-			Header: &corev3.HeaderValue{Key: originalPathHeader, RawValue: []byte(originalPath)},
-		})
-	}
+	r.requestHeaders[originalPathHeader] = originalPath
+	additionalHeaders = append(additionalHeaders, &corev3.HeaderValueOption{
+		Header: &corev3.HeaderValue{Key: originalPathHeader, RawValue: []byte(originalPath)},
+	})
 	if r.requestHeaders[internalapi.EnvoyOriginalPathHeader] == "" {
 		r.requestHeaders[internalapi.EnvoyOriginalPathHeader] = originalPath
 		additionalHeaders = append(additionalHeaders, &corev3.HeaderValueOption{

internal/mcpproxy/config.go

@@ -31,6 +31,7 @@ type (
 		tracer                     tracingapi.MCPTracer
 		client                     http.Client
 		logRequestHeaderAttributes map[string]string
+		maxRequestBodySize         int64 // maximum allowed POST body size in bytes
 	}
 
 	mcpProxyConfig struct {

internal/mcpproxy/handlers.go

@@ -276,8 +276,18 @@ func (m *mcpRequestContext) servePOST(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
-	body, err := io.ReadAll(r.Body)
+	limit := m.maxRequestBodySize
+	if limit <= 0 {
+		limit = defaultMaxRequestBodySize
+	}
+	body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, limit))
 	if err != nil {
+		var maxBytesErr *http.MaxBytesError
+		if errors.As(err, &maxBytesErr) {
+			errType = metrics.MCPErrorInternal
+			onErrorResponse(w, http.StatusRequestEntityTooLarge, "request body too large")
+			return
+		}
 		errType = metrics.MCPErrorInternal
 		onErrorResponse(w, http.StatusBadRequest, err.Error())
 		return

internal/mcpproxy/handlers_test.go

@@ -168,6 +168,20 @@ func TestServePOST_InvalidJSONRPC(t *testing.T) {
 	require.Contains(t, rr.Body.String(), "invalid JSON-RPC message")
 }
 
+func TestServePOST_OversizedBody(t *testing.T) {
+	proxy := newTestMCPProxy()
+	proxy.maxRequestBodySize = 16 // tiny limit to exercise the guard without allocating a large buffer.
+
+	body := strings.NewReader(`{"jsonrpc":"2.0","method":"initialize","id":1,"params":{}}`) // > 16 bytes
+	req := httptest.NewRequest(http.MethodPost, "/mcp", body)
+	rr := httptest.NewRecorder()
+
+	proxy.servePOST(rr, req)
+
+	require.Equal(t, http.StatusRequestEntityTooLarge, rr.Code)
+	require.Contains(t, rr.Body.String(), "request body too large")
+}
+
 func TestServePOST_InvalidSessionID(t *testing.T) {
 	proxy := newTestMCPProxy()
 	req := httptest.NewRequest(http.MethodPost, "/mcp", strings.NewReader(`{"jsonrpc":"2.0","method":"tools/call","params":{"name":"test-tool"},"id":"1"}`))

internal/mcpproxy/mcpproxy.go

@@ -14,6 +14,8 @@ import (
 	"log/slog"
 	"maps"
 	"net/http"
+	"os"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -39,6 +41,20 @@ type mcpRequestContext struct {
 	perBackendMetricsRecorded bool
 }
 
+// defaultMaxRequestBodySize is the default maximum allowed POST body size in bytes (4 MiB).
+const defaultMaxRequestBodySize = 4 * 1024 * 1024
+
+// getMaxRequestBodySize returns the configured POST body limit from the environment variable,
+// falling back to 4 MiB if the variable is unset or invalid.
+func getMaxRequestBodySize() int64 {
+	if v, ok := os.LookupEnv("MCP_PROXY_MAX_REQUEST_BODY_SIZE"); ok {
+		if n, err := strconv.ParseInt(v, 10, 64); err == nil && n > 0 {
+			return n
+		}
+	}
+	return defaultMaxRequestBodySize
+}
+
 // NewMCPProxy creates a new MCPProxy instance.
 func NewMCPProxy(l *slog.Logger, mcpMetrics metrics.MCPMetrics, tracer tracingapi.MCPTracer, sessionCrypto SessionCrypto, logRequestHeaderAttributes map[string]string) (*ProxyConfig, *http.ServeMux, error) {
 	toolChangeSignaler := newMultiWatcherSignaler() // used to signal changes to all active sessions.
@@ -49,6 +65,7 @@ func NewMCPProxy(l *slog.Logger, mcpMetrics metrics.MCPMetrics, tracer tracingap
 		l:                          l,
 		client:                     http.Client{}, // No timeout as it's enforced at Envoy level.
 		logRequestHeaderAttributes: maps.Clone(logRequestHeaderAttributes),
+		maxRequestBodySize:         getMaxRequestBodySize(),
 	}
 	mux := http.NewServeMux()
 	mux.HandleFunc(

internal/mcpproxy/mcpproxy_test.go

@@ -60,6 +60,25 @@ func (f *fakeTracer) StartSpanAndInjectMeta(context.Context, *jsonrpc.Request, m
 
 var noopTracer = tracingapi.NoopMCPTracer{}
 
+func TestGetMaxRequestBodySize(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		envValue string
+		want     int64
+	}{
+		{name: "default when env var not set", envValue: "", want: defaultMaxRequestBodySize},
+		{name: "custom value from env var", envValue: "1048576", want: 1048576},
+		{name: "default when env var is not a number", envValue: "not-a-number", want: defaultMaxRequestBodySize},
+		{name: "default when env var is zero", envValue: "0", want: defaultMaxRequestBodySize},
+		{name: "default when env var is negative", envValue: "-1", want: defaultMaxRequestBodySize},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Setenv("MCP_PROXY_MAX_REQUEST_BODY_SIZE", tc.envValue)
+			require.Equal(t, tc.want, getMaxRequestBodySize())
+		})
+	}
+}
+
 func TestNewMCPProxy(t *testing.T) {
 	l := slog.Default()
 	proxy, mux, err := NewMCPProxy(l, stubMetrics{}, noopTracer, NewPBKDF2AesGcmSessionCrypto("test", 100), nil)

internal/pprof/pprof.go

@@ -40,7 +40,7 @@ func run(ctx context.Context, l *log.Logger) {
 		mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
 		mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
 		mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
-		server := &http.Server{Addr: ":" + pprofPort, Handler: mux, ReadHeaderTimeout: 5 * time.Second}
+		server := &http.Server{Addr: "127.0.0.1:" + pprofPort, Handler: mux, ReadHeaderTimeout: 5 * time.Second}
 		go func() {
 			l.Printf("starting pprof server on port %s", pprofPort)
 			if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {

internal/translator/anthropic_helper.go

@@ -966,8 +966,8 @@ func (p *anthropicStreamParser) handleAnthropicStreamEvent(eventType []byte, dat
 			&u.CacheReadInputTokens,
 			&u.CacheCreationInputTokens,
 		)
-		// For message_start, we store the initial usage but don't add to the accumulated
-		// The message_delta event will contain the final totals
+		// Set all input token counts (input, cache read, cache creation) from message_start.
+		// message_delta may also contain these fields but only output_tokens is used from it.
 		if input, ok := usage.InputTokens(); ok {
 			p.tokenUsage.SetInputTokens(input)
 		}
@@ -1061,17 +1061,6 @@ func (p *anthropicStreamParser) handleAnthropicStreamEvent(eventType []byte, dat
 		if output, ok := usage.OutputTokens(); ok {
 			p.tokenUsage.AddOutputTokens(output)
 		}
-		// Update input tokens to include any cache tokens from delta
-		if cached, ok := usage.CachedInputTokens(); ok {
-			p.tokenUsage.AddInputTokens(cached)
-			// Accumulate any additional cache tokens from delta
-			p.tokenUsage.AddCachedInputTokens(cached)
-		}
-		if cacheCreation, ok := usage.CacheCreationInputTokens(); ok {
-			p.tokenUsage.AddInputTokens(cacheCreation)
-			// Accumulate cache creation tokens
-			p.tokenUsage.AddCacheCreationInputTokens(cacheCreation)
-		}
 		p.tokenUsage.SetReasoningTokens(uint32(u.OutputTokensDetails.ThinkingTokens)) //nolint:gosec
 		if event.Delta.StopReason != "" {
 			p.stopReason = event.Delta.StopReason

internal/translator/anthropic_helper_test.go

@@ -7,15 +7,18 @@ package translator
 
 import (
 	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/anthropics/anthropic-sdk-go"
 	"github.com/anthropics/anthropic-sdk-go/shared/constant"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"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"
 )
 
 // mockErrorReader is a helper for testing io.Reader failures.
@@ -1157,3 +1160,148 @@ func TestBuildAnthropicParamsWithReasoningEffort(t *testing.T) {
 		require.Equal(t, anthropic.OutputConfigEffort(""), params.OutputConfig.Effort)
 	})
 }
+
+func TestAnthropicStreamParser_StreamingTokenUsage(t *testing.T) {
+	tests := []struct {
+		name                        string
+		events                      string
+		expectedInputTokens         uint32
+		expectedOutputTokens        uint32
+		expectedTotalTokens         uint32
+		expectedCachedTokens        uint32
+		expectedCacheCreationTokens uint32
+	}{
+		{
+			name: "with cache tokens",
+			events: `event: message_start
+data: {"type": "message_start", "message": {"id": "msg_abc123", "type": "message", "role": "assistant", "content": [], "model": "claude-sonnet-4-6", "usage": {"input_tokens": 678, "cache_read_input_tokens": 13363, "cache_creation_input_tokens": 0, "output_tokens": 1}}}
+
+event: content_block_start
+data: {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}
+
+event: content_block_delta
+data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Hi"}}
+
+event: content_block_stop
+data: {"type": "content_block_stop", "index": 0}
+
+event: message_delta
+data: {"type": "message_delta", "delta": {"stop_reason": "end_turn"}, "usage": {"input_tokens": 678, "cache_read_input_tokens": 13363, "cache_creation_input_tokens": 0, "output_tokens": 5}}
+
+event: message_stop
+data: {"type": "message_stop"}
+
+`,
+			expectedInputTokens:         14041, // 678 + 13363 + 0
+			expectedOutputTokens:        5,
+			expectedTotalTokens:         14046, // 14041 + 5
+			expectedCachedTokens:        13363,
+			expectedCacheCreationTokens: 0,
+		},
+		{
+			name: "without cache tokens",
+			events: `event: message_start
+data: {"type": "message_start", "message": {"id": "msg_abc456", "type": "message", "role": "assistant", "content": [], "model": "claude-sonnet-4-6", "usage": {"input_tokens": 100, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 0, "output_tokens": 1}}}
+
+event: content_block_start
+data: {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}
+
+event: content_block_delta
+data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Hello"}}
+
+event: content_block_stop
+data: {"type": "content_block_stop", "index": 0}
+
+event: message_delta
+data: {"type": "message_delta", "delta": {"stop_reason": "end_turn"}, "usage": {"input_tokens": 100, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 0, "output_tokens": 10}}
+
+event: message_stop
+data: {"type": "message_stop"}
+
+`,
+			expectedInputTokens:         100,
+			expectedOutputTokens:        10,
+			expectedTotalTokens:         110,
+			expectedCachedTokens:        0,
+			expectedCacheCreationTokens: 0,
+		},
+		{
+			name: "with cache creation tokens",
+			events: `event: message_start
+data: {"type": "message_start", "message": {"id": "msg_abc789", "type": "message", "role": "assistant", "content": [], "model": "claude-sonnet-4-6", "usage": {"input_tokens": 200, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 5000, "output_tokens": 1}}}
+
+event: content_block_start
+data: {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}
+
+event: content_block_delta
+data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Response"}}
+
+event: content_block_stop
+data: {"type": "content_block_stop", "index": 0}
+
+event: message_delta
+data: {"type": "message_delta", "delta": {"stop_reason": "end_turn"}, "usage": {"input_tokens": 200, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 5000, "output_tokens": 8}}
+
+event: message_stop
+data: {"type": "message_stop"}
+
+`,
+			expectedInputTokens:         5200, // 200 + 5000 + 0
+			expectedOutputTokens:        8,
+			expectedTotalTokens:         5208, // 5200 + 8
+			expectedCachedTokens:        0,
+			expectedCacheCreationTokens: 5000,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			parser := newAnthropicStreamParser("claude-sonnet-4-6")
+
+			// Feed each event block separately (simulating chunked SSE delivery),
+			// with the last chunk marked as endOfStream.
+			chunks := splitSSEEvents(tt.events)
+			var tokenUsage metrics.TokenUsage
+			for i, chunk := range chunks {
+				endOfStream := i == len(chunks)-1
+				_, _, usage, _, err := parser.Process(strings.NewReader(chunk), endOfStream, nil)
+				require.NoError(t, err)
+				if endOfStream {
+					tokenUsage = usage
+				}
+			}
+
+			inputTokens, inputSet := tokenUsage.InputTokens()
+			assert.True(t, inputSet, "InputTokens should be set")
+			assert.Equal(t, tt.expectedInputTokens, inputTokens, "InputTokens mismatch")
+
+			outputTokens, outputSet := tokenUsage.OutputTokens()
+			assert.True(t, outputSet, "OutputTokens should be set")
+			assert.Equal(t, tt.expectedOutputTokens, outputTokens, "OutputTokens mismatch")
+
+			totalTokens, totalSet := tokenUsage.TotalTokens()
+			assert.True(t, totalSet, "TotalTokens should be set")
+			assert.Equal(t, tt.expectedTotalTokens, totalTokens, "TotalTokens mismatch")
+
+			cachedTokens, cachedSet := tokenUsage.CachedInputTokens()
+			assert.True(t, cachedSet, "CachedInputTokens should be set")
+			assert.Equal(t, tt.expectedCachedTokens, cachedTokens, "CachedInputTokens mismatch")
+
+			cacheCreation, cacheCreationSet := tokenUsage.CacheCreationInputTokens()
+			assert.True(t, cacheCreationSet, "CacheCreationInputTokens should be set")
+			assert.Equal(t, tt.expectedCacheCreationTokens, cacheCreation, "CacheCreationInputTokens mismatch")
+		})
+	}
+}
+
+func splitSSEEvents(data string) []string {
+	parts := strings.Split(data, "\n\n")
+	var events []string
+	for _, p := range parts {
+		trimmed := strings.TrimSpace(p)
+		if trimmed != "" {
+			events = append(events, p+"\n\n")
+		}
+	}
+	return events
+}

site/src/components/HomepageFeatures/index.tsx

@@ -19,11 +19,11 @@ const FeatureList: FeatureItem[] = [
     ),
   },
   {
-    title: 'v0.5 Release now available',
+    title: 'v0.7 Release now available',
     image: require('@site/static/img/3.png').default,
     description: (
       <>
-        The v0.5 Release of Envoy AI Gateway is now available. See the <a href="/release-notes/v0.5">release notes</a> for more information.
+        The v0.7 Release of Envoy AI Gateway is now available. See the <a href="/release-notes/v0.7">release notes</a> for more information.
       </>
     ),
   },