Repository: envoyproxy/ai-gateway
Pull request #497 (2025-03-14 19:11:11)

backport: completion stream + metrics and assistant content
**Commit Message**

PR to backport `mockChatCompletionMetrics`, chat completion stream fix, and openai content type.

Including:
- #459 (468 uses mock components introduced here)
- #468 
- #486

mathetake: nice
aabchoo: ```
Run amannn/action-semantic-pull-request@v5
Error: No release type found in pull request title " backport: completion stream + metrics and assistant content". Add a prefix to indicate what kind of release this pull request corresponds to. For reference, see https://www.conventionalcommits.org/

Available types:
 - docs
 - style
 - feat
 - test
 - build
 - ci
 - chore
 - revert
 - release
 - api
 - deps
 - e2e
 - extproc
 - controller
 - translator
 - examples
 - blog
 - site
 - backport
```

False negative??

Patch:
diff --git a/cmd/extproc/mainlib/main.go b/cmd/extproc/mainlib/main.go
index 6d6689e..b5f4fbd 100644
--- a/cmd/extproc/mainlib/main.go
+++ b/cmd/extproc/mainlib/main.go
@@ -13,6 +13,7 @@ import (
 	"log"
 	"log/slog"
 	"net"
+	"net/http"
 	"os"
 	"os/signal"
 	"strings"
@@ -20,10 +21,16 @@ import (
 	"time"
 
 	extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	otelprom "go.opentelemetry.io/otel/exporters/prometheus"
+	"go.opentelemetry.io/otel/metric"
+	metricsdk "go.opentelemetry.io/otel/sdk/metric"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/health/grpc_health_v1"
 
 	"github.com/envoyproxy/ai-gateway/internal/extproc"
+	"github.com/envoyproxy/ai-gateway/internal/metrics"
 	"github.com/envoyproxy/ai-gateway/internal/version"
 )
 
@@ -32,6 +39,7 @@ type extProcFlags struct {
 	configPath  string     // path to the configuration file.
 	extProcAddr string     // gRPC address for the external processor.
 	logLevel    slog.Level // log level for the external processor.
+	metricsAddr string     // HTTP address for the metrics server.
 }
 
 // parseAndValidateFlags parses and validates the flas passed to the external processor.
@@ -51,13 +59,14 @@ func parseAndValidateFlags(args []string) (extProcFlags, error) {
 	fs.StringVar(&flags.extProcAddr,
 		"extProcAddr",
 		":1063",
-		"gRPC address for the external processor. For example, :1063 or unix:///tmp/ext_proc.sock",
+		"gRPC address for the external processor. For example, :1063 or unix:///tmp/ext_proc.sock.",
 	)
 	logLevelPtr := fs.String(
 		"logLevel",
 		"info",
 		"log level for the external processor. One of 'debug', 'info', 'warn', or 'error'.",
 	)
+	fs.StringVar(&flags.metricsAddr, "metricsAddr", ":9190", "HTTP address for the metrics server.")
 
 	if err := fs.Parse(args); err != nil {
 		return extProcFlags{}, fmt.Errorf("failed to parse extProcFlags: %w", err)
@@ -102,11 +111,13 @@ func Main() {
 		log.Fatalf("failed to listen: %v", err)
 	}
 
+	metricsServer, meter := startMetricsServer(flags.metricsAddr, l)
+
 	server, err := extproc.NewServer(l)
 	if err != nil {
 		log.Fatalf("failed to create external processor server: %v", err)
 	}
-	server.Register("/v1/chat/completions", extproc.NewChatCompletionProcessor)
+	server.Register("/v1/chat/completions", extproc.ChatCompletionProcessorFactory(metrics.NewChatCompletion(meter)))
 	server.Register("/v1/models", extproc.NewModelsProcessor)
 
 	if err := extproc.StartConfigWatcher(ctx, flags.configPath, server, l, time.Second*5); err != nil {
@@ -116,10 +127,18 @@ func Main() {
 	s := grpc.NewServer()
 	extprocv3.RegisterExternalProcessorServer(s, server)
 	grpc_health_v1.RegisterHealthServer(s, server)
+
 	go func() {
 		<-ctx.Done()
 		s.GracefulStop()
+
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		if err := metricsServer.Shutdown(shutdownCtx); err != nil {
+			l.Error("Failed to shutdown metrics server gracefully", "error", err)
+		}
 	}()
+
 	_ = s.Serve(lis)
 }
 
@@ -130,3 +149,46 @@ func listenAddress(addrFlag string) (string, string) {
 	}
 	return "tcp", addrFlag
 }
+
+// startMetricsServer starts the HTTP server for Prometheus metrics.
+func startMetricsServer(addr string, logger *slog.Logger) (*http.Server, metric.Meter) {
+	registry := prometheus.NewRegistry()
+	exporter, err := otelprom.New(otelprom.WithRegisterer(registry))
+	if err != nil {
+		log.Fatal("failed to create metrics exporter")
+	}
+	provider := metricsdk.NewMeterProvider(metricsdk.WithReader(exporter))
+	meter := provider.Meter("envoyproxy/ai-gateway")
+
+	// Create a new HTTP server for metrics.
+	mux := http.NewServeMux()
+
+	// Register the metrics handler.
+	mux.Handle("/metrics", promhttp.HandlerFor(
+		registry,
+		promhttp.HandlerOpts{
+			EnableOpenMetrics: true,
+		},
+	))
+
+	// Add a simple health check endpoint.
+	mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte("OK"))
+	})
+
+	server := &http.Server{
+		Addr:              addr,
+		Handler:           mux,
+		ReadHeaderTimeout: 5 * time.Second,
+	}
+
+	go func() {
+		logger.Info("Starting metrics server", "address", addr)
+		if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			logger.Error("Metrics server failed", "error", err)
+		}
+	}()
+
+	return server, meter
+}
diff --git a/go.mod b/go.mod
index 30d0374..c1d24d4 100644
--- a/go.mod
+++ b/go.mod
@@ -13,8 +13,13 @@ require (
 	github.com/go-logr/logr v1.4.2
 	github.com/google/cel-go v0.23.2
 	github.com/google/go-cmp v0.7.0
-	github.com/openai/openai-go v0.1.0-alpha.59
+	github.com/openai/openai-go v0.1.0-alpha.61
+	github.com/prometheus/client_golang v1.20.5
 	github.com/stretchr/testify v1.10.0
+	go.opentelemetry.io/otel v1.34.0
+	go.opentelemetry.io/otel/exporters/prometheus v0.56.0
+	go.opentelemetry.io/otel/metric v1.34.0
+	go.opentelemetry.io/otel/sdk/metric v1.34.0
 	go.uber.org/goleak v1.3.0
 	go.uber.org/zap v1.27.0
 	golang.org/x/exp v0.0.0-20250128182459-e0ece0dbea4c
@@ -271,7 +276,6 @@ require (
 	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/polyfloyd/go-errorlint v1.7.1 // indirect
-	github.com/prometheus/client_golang v1.20.5 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.62.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
@@ -343,8 +347,7 @@ require (
 	go-simpler.org/sloglint v0.9.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.58.0 // indirect
-	go.opentelemetry.io/otel v1.34.0 // indirect
-	go.opentelemetry.io/otel/metric v1.34.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.34.0 // indirect
 	go.opentelemetry.io/otel/trace v1.34.0 // indirect
 	go.uber.org/automaxprocs v1.6.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
diff --git a/go.sum b/go.sum
index 4fce379..b4bb82d 100644
--- a/go.sum
+++ b/go.sum
@@ -632,8 +632,8 @@ github.com/onsi/ginkgo/v2 v2.22.2 h1:/3X8Panh8/WwhU/3Ssa6rCKqPLuAkVY2I0RoyDLySlU
 github.com/onsi/ginkgo/v2 v2.22.2/go.mod h1:oeMosUL+8LtarXBHu/c0bx2D/K9zyQ6uX3cTyztHwsk=
 github.com/onsi/gomega v1.36.2 h1:koNYke6TVk6ZmnyHrCXba/T/MoLBXFjeC1PtvYgw0A8=
 github.com/onsi/gomega v1.36.2/go.mod h1:DdwyADRjrc825LhMEkD76cHR5+pUnjhUN8GlHlRPHzY=
-github.com/openai/openai-go v0.1.0-alpha.59 h1:T3IYwKSCezfIlL9Oi+CGvU03fq0RoH33775S78Ti48Y=
-github.com/openai/openai-go v0.1.0-alpha.59/go.mod h1:3SdE6BffOX9HPEQv8IL/fi3LYZ5TUpRYaqGQZbyk11A=
+github.com/openai/openai-go v0.1.0-alpha.61 h1:dLJW1Dk15VAwm76xyPsiPt/Ky94NNGoMLETAI1ISoBY=
+github.com/openai/openai-go v0.1.0-alpha.61/go.mod h1:3SdE6BffOX9HPEQv8IL/fi3LYZ5TUpRYaqGQZbyk11A=
 github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQb2IpWsCzug=
diff --git a/internal/apischema/openai/openai.go b/internal/apischema/openai/openai.go
index c8600ed..3065952 100644
--- a/internal/apischema/openai/openai.go
+++ b/internal/apischema/openai/openai.go
@@ -10,6 +10,7 @@ package openai
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -149,6 +150,28 @@ func (c *ChatCompletionContentPartUserUnionParam) UnmarshalJSON(data []byte) err
 	return nil
 }
 
+type StringOrAssistantRoleContentUnion struct {
+	Value interface{}
+}
+
+func (s *StringOrAssistantRoleContentUnion) UnmarshalJSON(data []byte) error {
+	var str string
+	err := json.Unmarshal(data, &str)
+	if err == nil {
+		s.Value = str
+		return nil
+	}
+
+	var content ChatCompletionAssistantMessageParamContent
+	err = json.Unmarshal(data, &content)
+	if err == nil {
+		s.Value = content
+		return nil
+	}
+
+	return errors.New("cannot unmarshal JSON data as string or assistant content parts")
+}
+
 type StringOrArray struct {
 	Value interface{}
 }
@@ -335,7 +358,7 @@ type ChatCompletionAssistantMessageParam struct {
 	Audio ChatCompletionAssistantMessageParamAudio `json:"audio,omitempty"`
 	// The contents of the assistant message. Required unless `tool_calls` or
 	// `function_call` is specified.
-	Content ChatCompletionAssistantMessageParamContent `json:"content"`
+	Content StringOrAssistantRoleContentUnion `json:"content"`
 	// An optional name for the participant. Provides the model information to
 	// differentiate between participants of the same role.
 	Name string `json:"name,omitempty"`
diff --git a/internal/extproc/chatcompletion_processor.go b/internal/extproc/chatcompletion_processor.go
index 9b4e5c3..c24ce90 100644
--- a/internal/extproc/chatcompletion_processor.go
+++ b/internal/extproc/chatcompletion_processor.go
@@ -16,6 +16,7 @@ import (
 	"log/slog"
 
 	corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	extprocv3http "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/ext_proc/v3"
 	extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	typev3 "github.com/envoyproxy/go-control-plane/envoy/type/v3"
 	"google.golang.org/protobuf/types/known/structpb"
@@ -25,18 +26,22 @@ import (
 	"github.com/envoyproxy/ai-gateway/internal/apischema/openai"
 	"github.com/envoyproxy/ai-gateway/internal/extproc/translator"
 	"github.com/envoyproxy/ai-gateway/internal/llmcostcel"
+	"github.com/envoyproxy/ai-gateway/internal/metrics"
 )
 
-// NewChatCompletionProcessor implements [Processor] for the /chat/completions endpoint.
-func NewChatCompletionProcessor(config *processorConfig, requestHeaders map[string]string, logger *slog.Logger) (Processor, error) {
-	if config.schema.Name != filterapi.APISchemaOpenAI {
-		return nil, fmt.Errorf("unsupported API schema: %s", config.schema.Name)
+// ChatCompletionProcessorFactory returns a factory method to instantiate the chat completion processor.
+func ChatCompletionProcessorFactory(ccm metrics.ChatCompletion) ProcessorFactory {
+	return func(config *processorConfig, requestHeaders map[string]string, logger *slog.Logger) (Processor, error) {
+		if config.schema.Name != filterapi.APISchemaOpenAI {
+			return nil, fmt.Errorf("unsupported API schema: %s", config.schema.Name)
+		}
+		return &chatCompletionProcessor{
+			config:         config,
+			requestHeaders: requestHeaders,
+			logger:         logger,
+			metrics:        ccm,
+		}, nil
 	}
-	return &chatCompletionProcessor{
-		config:         config,
-		requestHeaders: requestHeaders,
-		logger:         logger,
-	}, nil
 }
 
 // chatCompletionProcessor handles the processing of the request and response messages for a single stream.
@@ -49,6 +54,10 @@ type chatCompletionProcessor struct {
 	translator       translator.Translator
 	// cost is the cost of the request that is accumulated during the processing of the response.
 	costs translator.LLMTokenUsage
+	// metrics tracking.
+	metrics metrics.ChatCompletion
+	// stream is set to true if the request is a streaming request.
+	stream bool
 }
 
 // selectTranslator selects the translator based on the output schema.
@@ -70,7 +79,10 @@ func (c *chatCompletionProcessor) selectTranslator(out filterapi.VersionedAPISch
 
 // ProcessRequestHeaders implements [Processor.ProcessRequestHeaders].
 func (c *chatCompletionProcessor) ProcessRequestHeaders(_ context.Context, _ *corev3.HeaderMap) (res *extprocv3.ProcessingResponse, err error) {
-	// The request headers have already been at the time the processor was created
+	// Start tracking metrics for this request.
+	c.metrics.StartRequest()
+
+	// The request headers have already been at the time the processor was created.
 	return &extprocv3.ProcessingResponse{Response: &extprocv3.ProcessingResponse_RequestHeaders{
 		RequestHeaders: &extprocv3.HeadersResponse{},
 	}}, nil
@@ -78,16 +90,23 @@ func (c *chatCompletionProcessor) ProcessRequestHeaders(_ context.Context, _ *co
 
 // ProcessRequestBody implements [Processor.ProcessRequestBody].
 func (c *chatCompletionProcessor) ProcessRequestBody(ctx context.Context, rawBody *extprocv3.HttpBody) (res *extprocv3.ProcessingResponse, err error) {
+	defer func() {
+		if err != nil {
+			c.metrics.RecordRequestCompletion(ctx, false)
+		}
+	}()
 	model, body, err := parseOpenAIChatCompletionBody(rawBody)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse request body: %w", err)
 	}
 	c.logger.Info("Processing request", "path", c.requestHeaders[":path"], "model", model)
 
+	c.metrics.SetModel(model)
 	c.requestHeaders[c.config.modelNameHeaderKey] = model
 	b, err := c.config.router.Calculate(c.requestHeaders)
 	if err != nil {
 		if errors.Is(err, x.ErrNoMatchingRule) {
+			c.metrics.RecordRequestCompletion(ctx, false)
 			return &extprocv3.ProcessingResponse{
 				Response: &extprocv3.ProcessingResponse_ImmediateResponse{
 					ImmediateResponse: &extprocv3.ImmediateResponse{
@@ -101,12 +120,13 @@ func (c *chatCompletionProcessor) ProcessRequestBody(ctx context.Context, rawBod
 		return nil, fmt.Errorf("failed to calculate route: %w", err)
 	}
 	c.logger.Info("Selected backend", "backend", b.Name)
+	c.metrics.SetBackend(*b)
 
 	if err = c.selectTranslator(b.Schema); err != nil {
 		return nil, fmt.Errorf("failed to select translator: %w", err)
 	}
 
-	headerMutation, bodyMutation, override, err := c.translator.RequestBody(body)
+	headerMutation, bodyMutation, err := c.translator.RequestBody(body)
 	if err != nil {
 		return nil, fmt.Errorf("failed to transform request: %w", err)
 	}
@@ -137,13 +157,18 @@ func (c *chatCompletionProcessor) ProcessRequestBody(ctx context.Context, rawBod
 				},
 			},
 		},
-		ModeOverride: override,
 	}
+	c.stream = body.Stream
 	return resp, nil
 }
 
 // ProcessResponseHeaders implements [Processor.ProcessResponseHeaders].
-func (c *chatCompletionProcessor) ProcessResponseHeaders(_ context.Context, headers *corev3.HeaderMap) (res *extprocv3.ProcessingResponse, err error) {
+func (c *chatCompletionProcessor) ProcessResponseHeaders(ctx context.Context, headers *corev3.HeaderMap) (res *extprocv3.ProcessingResponse, err error) {
+	defer func() {
+		if err != nil {
+			c.metrics.RecordRequestCompletion(ctx, false)
+		}
+	}()
 	c.responseHeaders = headersToMap(headers)
 	if enc := c.responseHeaders["content-encoding"]; enc != "" {
 		c.responseEncoding = enc
@@ -159,15 +184,23 @@ func (c *chatCompletionProcessor) ProcessResponseHeaders(_ context.Context, head
 	if err != nil {
 		return nil, fmt.Errorf("failed to transform response headers: %w", err)
 	}
+	var mode *extprocv3http.ProcessingMode
+	if c.stream && c.responseHeaders[":status"] == "200" {
+		// We only stream the response if the status code is 200 and the response is a stream.
+		mode = &extprocv3http.ProcessingMode{ResponseBodyMode: extprocv3http.ProcessingMode_STREAMED}
+	}
 	return &extprocv3.ProcessingResponse{Response: &extprocv3.ProcessingResponse_ResponseHeaders{
 		ResponseHeaders: &extprocv3.HeadersResponse{
 			Response: &extprocv3.CommonResponse{HeaderMutation: headerMutation},
 		},
-	}}, nil
+	}, ModeOverride: mode}, nil
 }
 
 // ProcessResponseBody implements [Processor.ProcessResponseBody].
-func (c *chatCompletionProcessor) ProcessResponseBody(_ context.Context, body *extprocv3.HttpBody) (res *extprocv3.ProcessingResponse, err error) {
+func (c *chatCompletionProcessor) ProcessResponseBody(ctx context.Context, body *extprocv3.HttpBody) (res *extprocv3.ProcessingResponse, err error) {
+	defer func() {
+		c.metrics.RecordRequestCompletion(ctx, err == nil)
+	}()
 	var br io.Reader
 	switch c.responseEncoding {
 	case "gzip":
@@ -200,20 +233,26 @@ func (c *chatCompletionProcessor) ProcessResponseBody(_ context.Context, body *e
 		},
 	}
 
-	// TODO: this is coupled with "LLM" specific logic. Once we have another use case, we need to refactor this.
+	// TODO: we need to investigate if we need to accumulate the token usage for streaming responses.
 	c.costs.InputTokens += tokenUsage.InputTokens
 	c.costs.OutputTokens += tokenUsage.OutputTokens
 	c.costs.TotalTokens += tokenUsage.TotalTokens
+
+	// Update metrics with token usage.
+	c.metrics.RecordTokenUsage(ctx, tokenUsage.InputTokens, tokenUsage.OutputTokens, tokenUsage.TotalTokens)
+	c.metrics.RecordTokenLatency(ctx, tokenUsage.OutputTokens)
+
 	if body.EndOfStream && len(c.config.requestCosts) > 0 {
 		resp.DynamicMetadata, err = c.maybeBuildDynamicMetadata()
 		if err != nil {
 			return nil, fmt.Errorf("failed to build dynamic metadata: %w", err)
 		}
 	}
+
 	return resp, nil
 }
 
-func parseOpenAIChatCompletionBody(body *extprocv3.HttpBody) (modelName string, rb translator.RequestBody, err error) {
+func parseOpenAIChatCompletionBody(body *extprocv3.HttpBody) (modelName string, rb *openai.ChatCompletionRequest, err error) {
 	var openAIReq openai.ChatCompletionRequest
 	if err := json.Unmarshal(body.Body, &openAIReq); err != nil {
 		return "", nil, fmt.Errorf("failed to unmarshal body: %w", err)
diff --git a/internal/extproc/translator/openai_awsbedrock.go b/internal/extproc/translator/openai_awsbedrock.go
index d3f3910..6f94605 100644
--- a/internal/extproc/translator/openai_awsbedrock.go
+++ b/internal/extproc/translator/openai_awsbedrock.go
@@ -17,7 +17,6 @@ import (
 
 	"github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream"
 	corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
-	extprocv3http "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/ext_proc/v3"
 	extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	"k8s.io/utils/ptr"
 
@@ -41,24 +40,12 @@ type openAIToAWSBedrockTranslatorV1ChatCompletion struct {
 }
 
 // RequestBody implements [Translator.RequestBody].
-func (o *openAIToAWSBedrockTranslatorV1ChatCompletion) RequestBody(body RequestBody) (
-	headerMutation *extprocv3.HeaderMutation, bodyMutation *extprocv3.BodyMutation, override *extprocv3http.ProcessingMode, err error,
+func (o *openAIToAWSBedrockTranslatorV1ChatCompletion) RequestBody(openAIReq *openai.ChatCompletionRequest) (
+	headerMutation *extprocv3.HeaderMutation, bodyMutation *extprocv3.BodyMutation, err error,
 ) {
-	openAIReq, ok := body.(*openai.ChatCompletionRequest)
-	if !ok {
-		return nil, nil, nil, fmt.Errorf("unexpected body type: %T", body)
-	}
-
 	var pathTemplate string
 	if openAIReq.Stream {
 		o.stream = true
-		// We need to change the processing mode for streaming requests.
-		// TODO: We can delete this explicit setting of ResponseHeaderMode below as it is the default value we use
-		// 	after https://github.com/envoyproxy/envoy/pull/38254 this is released.
-		override = &extprocv3http.ProcessingMode{
-			ResponseHeaderMode: extprocv3http.ProcessingMode_SEND,
-			ResponseBodyMode:   extprocv3http.ProcessingMode_STREAMED,
-		}
 		pathTemplate = "/model/%s/converse-stream"
 	} else {
 		pathTemplate = "/model/%s/converse"
@@ -83,22 +70,22 @@ func (o *openAIToAWSBedrockTranslatorV1ChatCompletion) RequestBody(body RequestB
 	// Convert Chat Completion messages.
 	err = o.openAIMessageToBedrockMessage(openAIReq, &bedrockReq)
 	if err != nil {
-		return nil, nil, nil, err
+		return nil, nil, err
 	}
 	// Convert ToolConfiguration.
 	if len(openAIReq.Tools) > 0 {
 		err = o.openAIToolsToBedrockToolConfiguration(openAIReq, &bedrockReq)
 		if err != nil {
-			return nil, nil, nil, err
+			return nil, nil, err
 		}
 	}
 
 	mut := &extprocv3.BodyMutation_Body{}
 	if mut.Body, err = json.Marshal(bedrockReq); err != nil {
-		return nil, nil, nil, fmt.Errorf("failed to marshal body: %w", err)
+		return nil, nil, fmt.Errorf("failed to marshal body: %w", err)
 	}
 	setContentLength(headerMutation, mut.Body)
-	return headerMutation, &extprocv3.BodyMutation{Mutation: mut}, override, nil
+	return headerMutation, &extprocv3.BodyMutation{Mutation: mut}, nil
 }
 
 // openAIToolsToBedrockToolConfiguration converts openai ChatCompletion tools to aws bedrock tool configurations.
@@ -258,11 +245,14 @@ func (o *openAIToAWSBedrockTranslatorV1ChatCompletion) openAIMessageToBedrockMes
 ) (*awsbedrock.Message, error) {
 	var bedrockMessage *awsbedrock.Message
 	contentBlocks := make([]*awsbedrock.ContentBlock, 0)
-	if openAiMessage.Content.Type == openai.ChatCompletionAssistantMessageParamContentTypeRefusal {
-		contentBlocks = append(contentBlocks, &awsbedrock.ContentBlock{Text: openAiMessage.Content.Refusal})
-	} else if openAiMessage.Content.Text != nil {
-		// TODO: we are sometimes missing the content (should fix)
-		contentBlocks = append(contentBlocks, &awsbedrock.ContentBlock{Text: openAiMessage.Content.Text})
+	if v, ok := openAiMessage.Content.Value.(string); ok {
+		contentBlocks = append(contentBlocks, &awsbedrock.ContentBlock{Text: &v})
+	} else if content, ok := openAiMessage.Content.Value.(openai.ChatCompletionAssistantMessageParamContent); ok {
+		if content.Type == openai.ChatCompletionAssistantMessageParamContentTypeRefusal {
+			contentBlocks = append(contentBlocks, &awsbedrock.ContentBlock{Text: content.Refusal})
+		} else if content.Text != nil {
+			contentBlocks = append(contentBlocks, &awsbedrock.ContentBlock{Text: content.Text})
+		}
 	}
 	bedrockMessage = &awsbedrock.Message{
 		Role:    role,
diff --git a/internal/extproc/translator/openai_openai.go b/internal/extproc/translator/openai_openai.go
index 62c110b..8baf484 100644
--- a/internal/extproc/translator/openai_openai.go
+++ b/internal/extproc/translator/openai_openai.go
@@ -12,7 +12,6 @@ import (
 	"io"
 	"strconv"
 
-	extprocv3http "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/ext_proc/v3"
 	extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 
 	"github.com/envoyproxy/ai-gateway/internal/apischema/openai"
@@ -31,23 +30,13 @@ type openAIToOpenAITranslatorV1ChatCompletion struct {
 }
 
 // RequestBody implements [Translator.RequestBody].
-func (o *openAIToOpenAITranslatorV1ChatCompletion) RequestBody(body RequestBody) (
-	headerMutation *extprocv3.HeaderMutation, bodyMutation *extprocv3.BodyMutation, override *extprocv3http.ProcessingMode, err error,
+func (o *openAIToOpenAITranslatorV1ChatCompletion) RequestBody(req *openai.ChatCompletionRequest) (
+	headerMutation *extprocv3.HeaderMutation, bodyMutation *extprocv3.BodyMutation, err error,
 ) {
-	req, ok := body.(*openai.ChatCompletionRequest)
-	if !ok {
-		return nil, nil, nil, fmt.Errorf("unexpected body type: %T", body)
-	}
 	if req.Stream {
 		o.stream = true
-		override = &extprocv3http.ProcessingMode{
-			// TODO: We can delete this explicit setting of ResponseHeaderMode below as it is the default value we use
-			// 	after https://github.com/envoyproxy/envoy/pull/38254 this is released.
-			ResponseHeaderMode: extprocv3http.ProcessingMode_SEND,
-			ResponseBodyMode:   extprocv3http.ProcessingMode_STREAMED,
-		}
 	}
-	return nil, nil, override, nil
+	return nil, nil, nil
 }
 
 // ResponseError implements [Translator.ResponseError]
diff --git a/internal/extproc/translator/translator.go b/internal/extproc/translator/translator.go
index 09f872e..ff07c7d 100644
--- a/internal/extproc/translator/translator.go
+++ b/internal/extproc/translator/translator.go
@@ -10,11 +10,12 @@ import (
 	"io"
 
 	corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
-	extprocv3http "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/ext_proc/v3"
 	extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
+
+	"github.com/envoyproxy/ai-gateway/internal/apischema/openai"
 )
 
-var (
+const (
 	statusHeaderName       = ":status"
 	contentTypeHeaderName  = "content-type"
 	awsErrorTypeHeaderName = "x-amzn-errortype"
@@ -43,11 +44,9 @@ type Translator interface {
 	// RequestBody translates the request body.
 	// 	- `body` is the request body already parsed by [router.RequestBodyParser]. The concrete type is specific to the schema and the path.
 	//	- This returns `headerMutation` and `bodyMutation` that can be nil to indicate no mutation.
-	//  - This returns `override` that to change the processing mode. This is used to process streaming requests properly.
-	RequestBody(body RequestBody) (
+	RequestBody(body *openai.ChatCompletionRequest) (
 		headerMutation *extprocv3.HeaderMutation,
 		bodyMutation *extprocv3.BodyMutation,
-		override *extprocv3http.ProcessingMode,
 		err error,
 	)
 
@@ -59,28 +58,16 @@ type Translator interface {
 		err error,
 	)
 
-	// ResponseBody translates the response body.
+	// ResponseBody translates the response body. When stream=true, this is called for each chunk of the response body.
 	// 	- `body` is the response body either chunk or the entire body, depending on the context.
 	//	- This returns `headerMutation` and `bodyMutation` that can be nil to indicate no mutation.
 	//  - This returns `tokenUsage` that is extracted from the body and will be used to do token rate limiting.
-	//
-	// TODO: this is coupled with "LLM" specific. Once we have another use case, we need to refactor this.
 	ResponseBody(respHeaders map[string]string, body io.Reader, endOfStream bool) (
 		headerMutation *extprocv3.HeaderMutation,
 		bodyMutation *extprocv3.BodyMutation,
 		tokenUsage LLMTokenUsage,
 		err error,
 	)
-
-	// ResponseError translates the response error.
-	//  - `headers` is the response headers.
-	//  - `body` is the response error body.
-	//  - This returns `headerMutation` and `bodyMutation` that can be nil to indicate no mutation.
-	ResponseError(respHeaders map[string]string, body io.Reader) (
-		headerMutation *extprocv3.HeaderMutation,
-		bodyMutation *extprocv3.BodyMutation,
-		err error,
-	)
 }
 
 func setContentLength(headers *extprocv3.HeaderMutation, body []byte) {
diff --git a/internal/metrics/chat_completion_metrics.go b/internal/metrics/chat_completion_metrics.go
new file mode 100644
index 0000000..1fcbd3f
--- /dev/null
+++ b/internal/metrics/chat_completion_metrics.go
@@ -0,0 +1,139 @@
+// Copyright Envoy AI Gateway Authors
+// SPDX-License-Identifier: Apache-2.0
+// The full text of the Apache license is available in the LICENSE file at
+// the root of the repo.
+
+package metrics
+
+import (
+	"context"
+	"time"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+
+	"github.com/envoyproxy/ai-gateway/filterapi"
+)
+
+// ChatCompletion is the interface for the chat completion AI Gateway metrics.
+type ChatCompletion interface {
+	// StartRequest initializes timing for a new request.
+	StartRequest()
+	// SetModel sets the model the request. This is usually called after parsing the request body .
+	SetModel(model string)
+	// SetBackend sets the selected backend when the routing decision has been made. This is usually called
+	// after parsing the request body to determine the model and invoke the routing logic.
+	SetBackend(backend filterapi.Backend)
+
+	// RecordTokenUsage records token usage metrics.
+	RecordTokenUsage(ctx context.Context, inputTokens, outputTokens, totalTokens uint32)
+	// RecordRequestCompletion records latency metrics for the entire request
+	RecordRequestCompletion(ctx context.Context, success bool)
+	// RecordTokenLatency records latency metrics for token generation.
+	RecordTokenLatency(ctx context.Context, tokens uint32)
+}
+
+// chatCompletion is the implementation for the chat completion AI Gateway metrics.
+type chatCompletion struct {
+	metrics        *genAI
+	firstTokenSent bool
+	requestStart   time.Time
+	lastTokenTime  time.Time
+	model          string
+	backend        string
+}
+
+// NewChatCompletion creates a new ChatCompletion instance.
+func NewChatCompletion(meter metric.Meter) ChatCompletion {
+	return &chatCompletion{
+		metrics: newGenAI(meter),
+		model:   "unknown",
+		backend: "unknown",
+	}
+}
+
+// StartRequest initializes timing for a new request.
+func (c *chatCompletion) StartRequest() {
+	c.requestStart = time.Now()
+	c.firstTokenSent = false
+}
+
+// SetModel sets the model for the request.
+func (c *chatCompletion) SetModel(model string) {
+	c.model = model
+}
+
+// SetBackend sets the name of the backend to be reported in the metrics according to:
+// https://opentelemetry.io/docs/specs/semconv/attributes-registry/gen-ai/#gen-ai-system
+func (c *chatCompletion) SetBackend(backend filterapi.Backend) {
+	switch backend.Schema.Name {
+	case filterapi.APISchemaOpenAI:
+		c.backend = genaiSystemOpenAI
+	case filterapi.APISchemaAWSBedrock:
+		c.backend = genAISystemAWSBedrock
+	default:
+		c.backend = backend.Name
+	}
+}
+
+// RecordTokenUsage implements [ChatCompletion.RecordTokenUsage].
+func (c *chatCompletion) RecordTokenUsage(ctx context.Context, inputTokens, outputTokens, totalTokens uint32) {
+	attrs := []attribute.KeyValue{
+		attribute.Key(genaiAttributeOperationName).String(genaiOperationChat),
+		attribute.Key(genaiAttributeSystemName).String(c.backend),
+		attribute.Key(genaiAttributeRequestModel).String(c.model),
+	}
+
+	c.metrics.tokenUsage.Record(ctx, float64(inputTokens),
+		metric.WithAttributes(attrs...),
+		metric.WithAttributes(attribute.Key(genaiAttributeTokenType).String(genaiTokenTypeInput)),
+	)
+	c.metrics.tokenUsage.Record(ctx, float64(outputTokens),
+		metric.WithAttributes(attrs...),
+		metric.WithAttributes(attribute.Key(genaiAttributeTokenType).String(genaiTokenTypeOutput)),
+	)
+	c.metrics.tokenUsage.Record(ctx, float64(totalTokens),
+		metric.WithAttributes(attrs...),
+		metric.WithAttributes(attribute.Key(genaiAttributeTokenType).String(genaiTokenTypeTotal)),
+	)
+}
+
+// RecordRequestCompletion implements [ChatCompletion.RecordRequestCompletion].
+func (c *chatCompletion) RecordRequestCompletion(ctx context.Context, success bool) {
+	attrs := []attribute.KeyValue{
+		attribute.Key(genaiAttributeOperationName).String(genaiOperationChat),
+		attribute.Key(genaiAttributeSystemName).String(c.backend),
+		attribute.Key(genaiAttributeRequestModel).String(c.model),
+	}
+
+	if success {
+		// According to the semantic conventions, the error attribute should not be added for successful operations
+		c.metrics.requestLatency.Record(ctx, time.Since(c.requestStart).Seconds(), metric.WithAttributes(attrs...))
+	} else {
+		// We don't have a set of typed errors yet, or a set of low-cardinality values, so we can just set the value to the
+		// placeholder one. See: https://opentelemetry.io/docs/specs/semconv/attributes-registry/error/#error-type
+		c.metrics.requestLatency.Record(ctx, time.Since(c.requestStart).Seconds(),
+			metric.WithAttributes(attrs...),
+			metric.WithAttributes(attribute.Key(genaiAttributeErrorType).String(genaiErrorTypeFallback)),
+		)
+	}
+}
+
+// RecordTokenLatency implements [ChatCompletion.RecordTokenLatency].
+func (c *chatCompletion) RecordTokenLatency(ctx context.Context, tokens uint32) {
+	attrs := []attribute.KeyValue{
+		attribute.Key(genaiAttributeOperationName).String(genaiOperationChat),
+		attribute.Key(genaiAttributeSystemName).String(c.backend),
+		attribute.Key(genaiAttributeRequestModel).String(c.model),
+	}
+
+	if !c.firstTokenSent {
+		c.firstTokenSent = true
+		c.metrics.firstTokenLatency.Record(ctx, time.Since(c.requestStart).Seconds(), metric.WithAttributes(attrs...))
+	} else if tokens > 0 {
+		// Calculate time between tokens.
+		itl := time.Since(c.lastTokenTime).Seconds() / float64(tokens)
+		c.metrics.outputTokenLatency.Record(ctx, itl, metric.WithAttributes(attrs...))
+	}
+	c.lastTokenTime = time.Now()
+}
diff --git a/internal/metrics/genai.go b/internal/metrics/genai.go
new file mode 100644
index 0000000..6f850e1
--- /dev/null
+++ b/internal/metrics/genai.go
@@ -0,0 +1,90 @@
+// Copyright Envoy AI Gateway Authors
+// SPDX-License-Identifier: Apache-2.0
+// The full text of the Apache license is available in the LICENSE file at
+// the root of the repo.
+
+package metrics
+
+import "go.opentelemetry.io/otel/metric"
+
+const (
+	// Metric names, attributes and values according to the Semantic Conventions for Generative AI Metrics.
+	// See: https://opentelemetry.io/docs/specs/semconv/gen-ai/gen-ai-metrics/
+
+	genaiMetricClientTokenUsage         = "gen_ai.client.token.usage" // #nosec G101: Potential hardcoded credentials
+	genaiMetricServerRequestDuration    = "gen_ai.server.request.duration"
+	genaiMetricServerTimeToFirstToken   = "gen_ai.server.time_to_first_token"   // #nosec G101: Potential hardcoded credentials
+	genaiMetricServerTimePerOutputToken = "gen_ai.server.time_per_output_token" // #nosec G101: Potential hardcoded credentials
+
+	genaiAttributeOperationName = "gen_ai.operation.name"
+	genaiAttributeSystemName    = "gen_ai.system.name"
+	genaiAttributeRequestModel  = "gen_ai.request.model"
+	genaiAttributeTokenType     = "gen_ai.token.type" // #nosec G101: Potential hardcoded credentials
+	genaiAttributeErrorType     = "error.type"
+
+	genaiOperationChat     = "chat"
+	genaiSystemOpenAI      = "openai"
+	genAISystemAWSBedrock  = "aws.bedrock"
+	genaiTokenTypeInput    = "input"
+	genaiTokenTypeOutput   = "output"
+	genaiTokenTypeTotal    = "total"
+	genaiErrorTypeFallback = "_OTHER"
+)
+
+// genAI holds metrics according to the Semantic Conventions for Generative AI Metrics.
+// See: https://opentelemetry.io/docs/specs/semconv/gen-ai/gen-ai-metrics/.
+type genAI struct {
+	// Number of tokens processed.
+	// See: https://opentelemetry.io/docs/specs/semconv/gen-ai/gen-ai-metrics/#metric-gen_aiclienttokenusage
+	tokenUsage metric.Float64Histogram
+	// requestLatency is the total latency of the request.
+	// Measured from the start of the received request headers in extproc to the end of the processed response body in extproc.
+	// See: https://opentelemetry.io/docs/specs/semconv/gen-ai/gen-ai-metrics/#metric-gen_aiserverrequestduration
+	requestLatency metric.Float64Histogram
+	// firstTokenLatency is the latency to receive the first token.
+	// Measured from the start of the received request headers in extproc to the receiving of the first token in the response body in extproc.
+	// See: https://opentelemetry.io/docs/specs/semconv/gen-ai/gen-ai-metrics/#metric-gen_aiservertime_to_first_token
+	firstTokenLatency metric.Float64Histogram
+	// outputTokenLatency is the latency between consecutive tokens, if supported, or by chunks/tokens otherwise, by backend, model.
+	// See: https://opentelemetry.io/docs/specs/semconv/gen-ai/gen-ai-metrics/#metric-gen_aiservertime_per_output_token
+	outputTokenLatency metric.Float64Histogram
+}
+
+// newGenAI creates a new genAI metrics instance.
+func newGenAI(meter metric.Meter) *genAI {
+	return &genAI{
+		tokenUsage: mustRegisterHistogram(meter,
+			genaiMetricClientTokenUsage,
+			metric.WithDescription("Number of tokens processed."),
+			metric.WithUnit("{token}"),
+			metric.WithExplicitBucketBoundaries(1, 4, 16, 64, 256, 1024, 4096, 16384, 65536, 262144, 1048576, 4194304, 16777216, 67108864),
+		),
+		requestLatency: mustRegisterHistogram(meter,
+			genaiMetricServerRequestDuration,
+			metric.WithDescription("Time spent processing request."),
+			metric.WithUnit("s"),
+			metric.WithExplicitBucketBoundaries(0.01, 0.02, 0.04, 0.08, 0.16, 0.32, 0.64, 1.28, 2.56, 5.12, 10.24, 20.48, 40.96, 81.92),
+		),
+		firstTokenLatency: mustRegisterHistogram(meter,
+			genaiMetricServerTimeToFirstToken,
+			metric.WithDescription("Time to receive first token in streaming responses."),
+			metric.WithUnit("s"),
+			metric.WithExplicitBucketBoundaries(0.001, 0.005, 0.01, 0.02, 0.04, 0.06, 0.08, 0.1, 0.25, 0.5, 0.75, 1.0, 2.5, 5.0, 7.5, 10.0),
+		),
+		outputTokenLatency: mustRegisterHistogram(meter,
+			genaiMetricServerTimePerOutputToken,
+			metric.WithDescription("Time between consecutive tokens in streaming responses."),
+			metric.WithUnit("s"),
+			metric.WithExplicitBucketBoundaries(0.01, 0.025, 0.05, 0.075, 0.1, 0.15, 0.2, 0.3, 0.4, 0.5, 0.75, 1.0, 2.5),
+		),
+	}
+}
+
+// mustRegisterHistogram registers a histogram with the meter and panics if it fails.
+func mustRegisterHistogram(meter metric.Meter, name string, options ...metric.Float64HistogramOption) metric.Float64Histogram {
+	h, err := meter.Float64Histogram(name, options...)
+	if err != nil {
+		panic(err)
+	}
+	return h
+}


Test patch:
diff --git a/cmd/extproc/mainlib/main_test.go b/cmd/extproc/mainlib/main_test.go
index 4f193af..8b2e41f 100644
--- a/cmd/extproc/mainlib/main_test.go
+++ b/cmd/extproc/mainlib/main_test.go
@@ -6,7 +6,9 @@
 package mainlib
 
 import (
+	"io"
 	"log/slog"
+	"net/http"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -104,3 +106,19 @@ func TestListenAddress(t *testing.T) {
 		})
 	}
 }
+
+func TestStartMetricsServer(t *testing.T) {
+	s, m := startMetricsServer("127.0.0.1:", slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{})))
+	t.Cleanup(func() { _ = s.Shutdown(t.Context()) })
+
+	require.NotNil(t, s)
+	require.NotNil(t, m)
+
+	require.HTTPStatusCode(t, s.Handler.ServeHTTP, http.MethodGet, "/", nil, http.StatusNotFound)
+
+	require.HTTPSuccess(t, s.Handler.ServeHTTP, http.MethodGet, "/health", nil)
+	require.HTTPBodyContains(t, s.Handler.ServeHTTP, http.MethodGet, "/health", nil, "OK")
+
+	require.HTTPSuccess(t, s.Handler.ServeHTTP, http.MethodGet, "/metrics", nil)
+	require.HTTPBodyContains(t, s.Handler.ServeHTTP, http.MethodGet, "/metrics", nil, "target_info{")
+}
diff --git a/internal/apischema/openai/openai_test.go b/internal/apischema/openai/openai_test.go
index ed0fd83..e0b621d 100644
--- a/internal/apischema/openai/openai_test.go
+++ b/internal/apischema/openai/openai_test.go
@@ -153,7 +153,35 @@ func TestOpenAIChatCompletionMessageUnmarshal(t *testing.T) {
 					{
 						Value: ChatCompletionAssistantMessageParam{
 							Role:    ChatMessageRoleAssistant,
-							Content: ChatCompletionAssistantMessageParamContent{Text: ptr.To("you are a helpful assistant")},
+							Content: StringOrAssistantRoleContentUnion{Value: ChatCompletionAssistantMessageParamContent{Text: ptr.To("you are a helpful assistant")}},
+						},
+						Type: ChatMessageRoleAssistant,
+					},
+				},
+			},
+		},
+		{
+			name: "assistant message string",
+			in: []byte(`{"model": "gpu-o4",
+                        "messages": [
+                         {"role": "assistant", "content": "you are a helpful assistant"},
+                         {"role": "assistant", "content": "{'text': 'you are a helpful assistant'}"}
+						 ]}
+`),
+			out: &ChatCompletionRequest{
+				Model: "gpu-o4",
+				Messages: []ChatCompletionMessageParamUnion{
+					{
+						Value: ChatCompletionAssistantMessageParam{
+							Role:    ChatMessageRoleAssistant,
+							Content: StringOrAssistantRoleContentUnion{Value: "you are a helpful assistant"},
+						},
+						Type: ChatMessageRoleAssistant,
+					},
+					{
+						Value: ChatCompletionAssistantMessageParam{
+							Role:    ChatMessageRoleAssistant,
+							Content: StringOrAssistantRoleContentUnion{Value: "{'text': 'you are a helpful assistant'}"},
 						},
 						Type: ChatMessageRoleAssistant,
 					},
diff --git a/internal/extproc/chatcompletion_processor_test.go b/internal/extproc/chatcompletion_processor_test.go
index a03d2c1..282b7b4 100644
--- a/internal/extproc/chatcompletion_processor_test.go
+++ b/internal/extproc/chatcompletion_processor_test.go
@@ -8,11 +8,13 @@ package extproc
 import (
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
 	"log/slog"
 	"testing"
 
 	corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	extprocv3http "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/ext_proc/v3"
 	extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	typev3 "github.com/envoyproxy/go-control-plane/envoy/type/v3"
 	"github.com/stretchr/testify/require"
@@ -24,15 +26,15 @@ import (
 	"github.com/envoyproxy/ai-gateway/internal/llmcostcel"
 )
 
-func TestChatCompletion_Scheema(t *testing.T) {
+func TestChatCompletion_Schema(t *testing.T) {
 	t.Run("unsupported", func(t *testing.T) {
 		cfg := &processorConfig{schema: filterapi.VersionedAPISchema{Name: "Foo", Version: "v123"}}
-		_, err := NewChatCompletionProcessor(cfg, nil, nil)
+		_, err := ChatCompletionProcessorFactory(nil)(cfg, nil, nil)
 		require.ErrorContains(t, err, "unsupported API schema: Foo")
 	})
 	t.Run("supported openai", func(t *testing.T) {
 		cfg := &processorConfig{schema: filterapi.VersionedAPISchema{Name: filterapi.APISchemaOpenAI, Version: "v123"}}
-		_, err := NewChatCompletionProcessor(cfg, nil, nil)
+		_, err := ChatCompletionProcessorFactory(nil)(cfg, nil, nil)
 		require.NoError(t, err)
 	})
 }
@@ -56,49 +58,100 @@ func TestChatCompletion_SelectTranslator(t *testing.T) {
 }
 
 func TestChatCompletion_ProcessRequestHeaders(t *testing.T) {
-	p := &chatCompletionProcessor{}
+	mm := &mockChatCompletionMetrics{}
+	p := &chatCompletionProcessor{
+		metrics: mm,
+	}
 	res, err := p.ProcessRequestHeaders(t.Context(), &corev3.HeaderMap{
 		Headers: []*corev3.HeaderValue{{Key: "foo", Value: "bar"}},
 	})
 	require.NoError(t, err)
 	_, ok := res.Response.(*extprocv3.ProcessingResponse_RequestHeaders)
 	require.True(t, ok)
+	require.NotZero(t, mm.requestStart)
+	mm.RequireRequestNotCompleted(t)
 }
 
 func TestChatCompletion_ProcessResponseHeaders(t *testing.T) {
 	t.Run("error translation", func(t *testing.T) {
+		mm := &mockChatCompletionMetrics{}
 		mt := &mockTranslator{t: t, expHeaders: make(map[string]string)}
-		p := &chatCompletionProcessor{translator: mt}
+		p := &chatCompletionProcessor{
+			translator: mt,
+			metrics:    mm,
+		}
 		mt.retErr = errors.New("test error")
 		_, err := p.ProcessResponseHeaders(t.Context(), nil)
 		require.ErrorContains(t, err, "test error")
+		mm.RequireRequestFailure(t)
 	})
-	t.Run("ok", func(t *testing.T) {
+	t.Run("ok/non-streaming", func(t *testing.T) {
 		inHeaders := &corev3.HeaderMap{
 			Headers: []*corev3.HeaderValue{{Key: "foo", Value: "bar"}, {Key: "dog", RawValue: []byte("cat")}},
 		}
 		expHeaders := map[string]string{"foo": "bar", "dog": "cat"}
+		mm := &mockChatCompletionMetrics{}
 		mt := &mockTranslator{t: t, expHeaders: expHeaders}
-		p := &chatCompletionProcessor{translator: mt}
+		p := &chatCompletionProcessor{
+			translator: mt,
+			metrics:    mm,
+		}
 		res, err := p.ProcessResponseHeaders(t.Context(), inHeaders)
 		require.NoError(t, err)
 		commonRes := res.Response.(*extprocv3.ProcessingResponse_ResponseHeaders).ResponseHeaders.Response
 		require.Equal(t, mt.retHeaderMutation, commonRes.HeaderMutation)
+		mm.RequireRequestNotCompleted(t)
+		require.Nil(t, res.ModeOverride)
+	})
+	t.Run("ok/streaming", func(t *testing.T) {
+		inHeaders := &corev3.HeaderMap{
+			Headers: []*corev3.HeaderValue{{Key: ":status", Value: "200"}, {Key: "dog", RawValue: []byte("cat")}},
+		}
+		expHeaders := map[string]string{":status": "200", "dog": "cat"}
+		mm := &mockChatCompletionMetrics{}
+		mt := &mockTranslator{t: t, expHeaders: expHeaders}
+		p := &chatCompletionProcessor{translator: mt, metrics: mm, stream: true}
+		res, err := p.ProcessResponseHeaders(t.Context(), inHeaders)
+		require.NoError(t, err)
+		commonRes := res.Response.(*extprocv3.ProcessingResponse_ResponseHeaders).ResponseHeaders.Response
+		require.Equal(t, mt.retHeaderMutation, commonRes.HeaderMutation)
+		require.Equal(t, &extprocv3http.ProcessingMode{ResponseBodyMode: extprocv3http.ProcessingMode_STREAMED}, res.ModeOverride)
+	})
+	t.Run("error/streaming", func(t *testing.T) {
+		inHeaders := &corev3.HeaderMap{
+			Headers: []*corev3.HeaderValue{{Key: ":status", Value: "500"}, {Key: "dog", RawValue: []byte("cat")}},
+		}
+		expHeaders := map[string]string{":status": "500", "dog": "cat"}
+		mm := &mockChatCompletionMetrics{}
+		mt := &mockTranslator{t: t, expHeaders: expHeaders}
+		p := &chatCompletionProcessor{translator: mt, metrics: mm, stream: true}
+		res, err := p.ProcessResponseHeaders(t.Context(), inHeaders)
+		require.NoError(t, err)
+		commonRes := res.Response.(*extprocv3.ProcessingResponse_ResponseHeaders).ResponseHeaders.Response
+		require.Equal(t, mt.retHeaderMutation, commonRes.HeaderMutation)
+		require.Nil(t, res.ModeOverride)
 	})
 }
 
 func TestChatCompletion_ProcessResponseBody(t *testing.T) {
 	t.Run("error translation", func(t *testing.T) {
+		mm := &mockChatCompletionMetrics{}
 		mt := &mockTranslator{t: t}
-		p := &chatCompletionProcessor{translator: mt}
+		p := &chatCompletionProcessor{
+			translator: mt,
+			metrics:    mm,
+		}
 		mt.retErr = errors.New("test error")
 		_, err := p.ProcessResponseBody(t.Context(), &extprocv3.HttpBody{})
 		require.ErrorContains(t, err, "test error")
+		mm.RequireRequestFailure(t)
+		mm.RequireTokensRecorded(t, 0)
 	})
 	t.Run("ok", func(t *testing.T) {
 		inBody := &extprocv3.HttpBody{Body: []byte("some-body"), EndOfStream: true}
 		expBodyMut := &extprocv3.BodyMutation{}
 		expHeadMut := &extprocv3.HeaderMutation{}
+		mm := &mockChatCompletionMetrics{}
 		mt := &mockTranslator{
 			t: t, expResponseBody: inBody,
 			retBodyMutation: expBodyMut, retHeaderMutation: expHeadMut,
@@ -109,26 +162,33 @@ func TestChatCompletion_ProcessResponseBody(t *testing.T) {
 		require.NoError(t, err)
 		celProgUint, err := llmcostcel.NewProgram("uint(9999)")
 		require.NoError(t, err)
-		p := &chatCompletionProcessor{translator: mt, logger: slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{})), config: &processorConfig{
-			metadataNamespace: "ai_gateway_llm_ns",
-			requestCosts: []processorConfigRequestCost{
-				{LLMRequestCost: &filterapi.LLMRequestCost{Type: filterapi.LLMRequestCostTypeOutputToken, MetadataKey: "output_token_usage"}},
-				{LLMRequestCost: &filterapi.LLMRequestCost{Type: filterapi.LLMRequestCostTypeInputToken, MetadataKey: "input_token_usage"}},
-				{
-					celProg:        celProgInt,
-					LLMRequestCost: &filterapi.LLMRequestCost{Type: filterapi.LLMRequestCostTypeCEL, MetadataKey: "cel_int"},
-				},
-				{
-					celProg:        celProgUint,
-					LLMRequestCost: &filterapi.LLMRequestCost{Type: filterapi.LLMRequestCostTypeCEL, MetadataKey: "cel_uint"},
+		p := &chatCompletionProcessor{
+			translator: mt,
+			logger:     slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{})),
+			metrics:    mm,
+			config: &processorConfig{
+				metadataNamespace: "ai_gateway_llm_ns",
+				requestCosts: []processorConfigRequestCost{
+					{LLMRequestCost: &filterapi.LLMRequestCost{Type: filterapi.LLMRequestCostTypeOutputToken, MetadataKey: "output_token_usage"}},
+					{LLMRequestCost: &filterapi.LLMRequestCost{Type: filterapi.LLMRequestCostTypeInputToken, MetadataKey: "input_token_usage"}},
+					{
+						celProg:        celProgInt,
+						LLMRequestCost: &filterapi.LLMRequestCost{Type: filterapi.LLMRequestCostTypeCEL, MetadataKey: "cel_int"},
+					},
+					{
+						celProg:        celProgUint,
+						LLMRequestCost: &filterapi.LLMRequestCost{Type: filterapi.LLMRequestCostTypeCEL, MetadataKey: "cel_uint"},
+					},
 				},
 			},
-		}}
+		}
 		res, err := p.ProcessResponseBody(t.Context(), inBody)
 		require.NoError(t, err)
 		commonRes := res.Response.(*extprocv3.ProcessingResponse_ResponseBody).ResponseBody.Response
 		require.Equal(t, expBodyMut, commonRes.BodyMutation)
 		require.Equal(t, expHeadMut, commonRes.HeaderMutation)
+		mm.RequireRequestSuccess(t)
+		mm.RequireTokensRecorded(t, 1)
 
 		md := res.DynamicMetadata
 		require.NotNil(t, md)
@@ -144,98 +204,159 @@ func TestChatCompletion_ProcessResponseBody(t *testing.T) {
 }
 
 func TestChatCompletion_ProcessRequestBody(t *testing.T) {
-	bodyFromModel := func(t *testing.T, model string) []byte {
-		var openAIReq openai.ChatCompletionRequest
-		openAIReq.Model = model
-		bytes, err := json.Marshal(openAIReq)
-		require.NoError(t, err)
-		return bytes
-	}
-	t.Run("body parser error", func(t *testing.T) {
-		p := &chatCompletionProcessor{}
-		_, err := p.ProcessRequestBody(t.Context(), &extprocv3.HttpBody{Body: []byte("nonjson")})
-		require.ErrorContains(t, err, "invalid character 'o' in literal null")
-	})
-	t.Run("router error", func(t *testing.T) {
-		headers := map[string]string{":path": "/foo"}
-		rt := mockRouter{t: t, expHeaders: headers, retErr: errors.New("test error")}
-		p := &chatCompletionProcessor{config: &processorConfig{router: rt}, requestHeaders: headers, logger: slog.Default()}
-		_, err := p.ProcessRequestBody(t.Context(), &extprocv3.HttpBody{Body: bodyFromModel(t, "some-model")})
-		require.ErrorContains(t, err, "failed to calculate route: test error")
-	})
-	t.Run("router error 404", func(t *testing.T) {
-		headers := map[string]string{":path": "/foo"}
-		rt := mockRouter{t: t, expHeaders: headers, retErr: x.ErrNoMatchingRule}
-		p := &chatCompletionProcessor{config: &processorConfig{router: rt}, requestHeaders: headers, logger: slog.Default()}
-		resp, err := p.ProcessRequestBody(t.Context(), &extprocv3.HttpBody{Body: bodyFromModel(t, "some-model")})
-		require.NoError(t, err)
-		require.NotNil(t, resp)
-		ir := resp.GetImmediateResponse()
-		require.NotNil(t, ir)
-		require.Equal(t, typev3.StatusCode_NotFound, ir.GetStatus().GetCode())
-		require.Equal(t, x.ErrNoMatchingRule.Error(), string(ir.GetBody()))
-	})
-	t.Run("translator not found", func(t *testing.T) {
-		headers := map[string]string{":path": "/foo"}
-		rt := mockRouter{
-			t: t, expHeaders: headers, retBackendName: "some-backend",
-			retVersionedAPISchema: filterapi.VersionedAPISchema{Name: "some-schema", Version: "v10.0"},
-		}
-		p := &chatCompletionProcessor{config: &processorConfig{router: rt}, requestHeaders: headers, logger: slog.Default()}
-		_, err := p.ProcessRequestBody(t.Context(), &extprocv3.HttpBody{Body: bodyFromModel(t, "some-model")})
-		require.ErrorContains(t, err, "unsupported API schema: backend={some-schema v10.0}")
-	})
-	t.Run("translator error", func(t *testing.T) {
-		headers := map[string]string{":path": "/foo"}
-		someBody := bodyFromModel(t, "some-model")
-		rt := mockRouter{
-			t: t, expHeaders: headers, retBackendName: "some-backend",
-			retVersionedAPISchema: filterapi.VersionedAPISchema{Name: "some-schema", Version: "v10.0"},
-		}
-		var body openai.ChatCompletionRequest
-		require.NoError(t, json.Unmarshal(someBody, &body))
-		tr := mockTranslator{t: t, retErr: errors.New("test error"), expRequestBody: &body}
-		p := &chatCompletionProcessor{
-			config:         &processorConfig{router: rt},
-			requestHeaders: headers, logger: slog.Default(), translator: tr,
-		}
-		_, err := p.ProcessRequestBody(t.Context(), &extprocv3.HttpBody{Body: someBody})
-		require.ErrorContains(t, err, "failed to transform request: test error")
-	})
-	t.Run("ok", func(t *testing.T) {
-		someBody := bodyFromModel(t, "some-model")
-		headers := map[string]string{":path": "/foo"}
-		rt := mockRouter{
-			t: t, expHeaders: headers, retBackendName: "some-backend",
-			retVersionedAPISchema: filterapi.VersionedAPISchema{Name: "some-schema", Version: "v10.0"},
-		}
-		headerMut := &extprocv3.HeaderMutation{}
-		bodyMut := &extprocv3.BodyMutation{}
+	for _, stream := range []bool{false, true} {
+		t.Run(fmt.Sprintf("stream%v", stream), func(t *testing.T) {
+			bodyFromModel := func(t *testing.T, model string) []byte {
+				var openAIReq openai.ChatCompletionRequest
+				openAIReq.Model = model
+				openAIReq.Stream = stream
+				bytes, err := json.Marshal(openAIReq)
+				require.NoError(t, err)
+				return bytes
+			}
+			t.Run("body parser error", func(t *testing.T) {
+				mm := &mockChatCompletionMetrics{}
+				p := &chatCompletionProcessor{
+					metrics: mm,
+				}
+				_, err := p.ProcessRequestBody(t.Context(), &extprocv3.HttpBody{Body: []byte("nonjson")})
+				require.ErrorContains(t, err, "invalid character 'o' in literal null")
+				mm.RequireRequestFailure(t)
+				mm.RequireTokensRecorded(t, 0)
+				mm.RequireSelected(t, "", "")
+				require.False(t, p.stream) // On error, stream should be false regardless of the input.
+			})
+			t.Run("router error", func(t *testing.T) {
+				headers := map[string]string{":path": "/foo"}
+				rt := mockRouter{t: t, expHeaders: headers, retErr: errors.New("test error")}
+				mm := &mockChatCompletionMetrics{}
+				p := &chatCompletionProcessor{
+					config:         &processorConfig{router: rt},
+					requestHeaders: headers,
+					logger:         slog.Default(),
+					metrics:        mm,
+				}
+				_, err := p.ProcessRequestBody(t.Context(), &extprocv3.HttpBody{Body: bodyFromModel(t, "some-model")})
+				require.ErrorContains(t, err, "failed to calculate route: test error")
+				mm.RequireRequestFailure(t)
+				mm.RequireTokensRecorded(t, 0)
+				mm.RequireSelected(t, "some-model", "")
+				require.False(t, p.stream) // On error, stream should be false regardless of the input.
+			})
+			t.Run("router error 404", func(t *testing.T) {
+				headers := map[string]string{":path": "/foo"}
+				rt := mockRouter{t: t, expHeaders: headers, retErr: x.ErrNoMatchingRule}
+				mm := &mockChatCompletionMetrics{}
+				p := &chatCompletionProcessor{
+					config:         &processorConfig{router: rt},
+					requestHeaders: headers,
+					logger:         slog.Default(),
+					metrics:        mm,
+				}
+				resp, err := p.ProcessRequestBody(t.Context(), &extprocv3.HttpBody{Body: bodyFromModel(t, "some-model")})
+				require.NoError(t, err)
+				require.NotNil(t, resp)
+				ir := resp.GetImmediateResponse()
+				require.NotNil(t, ir)
+				require.Equal(t, typev3.StatusCode_NotFound, ir.GetStatus().GetCode())
+				require.Equal(t, x.ErrNoMatchingRule.Error(), string(ir.GetBody()))
+				mm.RequireRequestFailure(t)
+				mm.RequireTokensRecorded(t, 0)
+				mm.RequireSelected(t, "some-model", "")
+				require.False(t, p.stream) // On error, stream should be false regardless of the input.
+			})
+			t.Run("translator not found", func(t *testing.T) {
+				headers := map[string]string{":path": "/foo"}
+				rt := mockRouter{
+					t: t, expHeaders: headers, retBackendName: "some-backend",
+					retVersionedAPISchema: filterapi.VersionedAPISchema{Name: "some-schema", Version: "v10.0"},
+				}
+				mm := &mockChatCompletionMetrics{}
+				p := &chatCompletionProcessor{
+					config:         &processorConfig{router: rt},
+					requestHeaders: headers,
+					logger:         slog.Default(),
+					metrics:        mm,
+				}
+				_, err := p.ProcessRequestBody(t.Context(), &extprocv3.HttpBody{Body: bodyFromModel(t, "some-model")})
+				require.ErrorContains(t, err, "unsupported API schema: backend={some-schema v10.0}")
+				mm.RequireRequestFailure(t)
+				mm.RequireTokensRecorded(t, 0)
+				mm.RequireSelected(t, "some-model", "some-backend")
+				require.False(t, p.stream) // On error, stream should be false regardless of the input.
+			})
+			t.Run("translator error", func(t *testing.T) {
+				headers := map[string]string{":path": "/foo"}
+				someBody := bodyFromModel(t, "some-model")
+				rt := mockRouter{
+					t: t, expHeaders: headers, retBackendName: "some-backend",
+					retVersionedAPISchema: filterapi.VersionedAPISchema{Name: "some-schema", Version: "v10.0"},
+				}
+				var body openai.ChatCompletionRequest
+				require.NoError(t, json.Unmarshal(someBody, &body))
+				tr := mockTranslator{t: t, retErr: errors.New("test error"), expRequestBody: &body}
+				mm := &mockChatCompletionMetrics{}
+				p := &chatCompletionProcessor{
+					config:         &processorConfig{router: rt},
+					requestHeaders: headers,
+					logger:         slog.Default(),
+					metrics:        mm,
+					translator:     tr,
+				}
+				_, err := p.ProcessRequestBody(t.Context(), &extprocv3.HttpBody{Body: someBody})
+				require.ErrorContains(t, err, "failed to transform request: test error")
+				mm.RequireRequestFailure(t)
+				mm.RequireTokensRecorded(t, 0)
+				mm.RequireSelected(t, "some-model", "some-backend")
+				require.False(t, p.stream) // On error, stream should be false regardless of the input.
+			})
+			t.Run("ok", func(t *testing.T) {
+				someBody := bodyFromModel(t, "some-model")
+				headers := map[string]string{":path": "/foo"}
+				rt := mockRouter{
+					t: t, expHeaders: headers, retBackendName: "some-backend",
+					retVersionedAPISchema: filterapi.VersionedAPISchema{Name: "some-schema", Version: "v10.0"},
+				}
+				headerMut := &extprocv3.HeaderMutation{}
+				bodyMut := &extprocv3.BodyMutation{}
 
-		var expBody openai.ChatCompletionRequest
-		require.NoError(t, json.Unmarshal(someBody, &expBody))
-		mt := mockTranslator{t: t, expRequestBody: &expBody, retHeaderMutation: headerMut, retBodyMutation: bodyMut}
-		p := &chatCompletionProcessor{config: &processorConfig{
-			router:                   rt,
-			selectedBackendHeaderKey: "x-ai-gateway-backend-key",
-			modelNameHeaderKey:       "x-ai-gateway-model-key",
-		}, requestHeaders: headers, logger: slog.Default(), translator: mt}
-		resp, err := p.ProcessRequestBody(t.Context(), &extprocv3.HttpBody{Body: someBody})
-		require.NoError(t, err)
-		require.Equal(t, mt, p.translator)
-		require.NotNil(t, resp)
-		commonRes := resp.Response.(*extprocv3.ProcessingResponse_RequestBody).RequestBody.Response
-		require.Equal(t, headerMut, commonRes.HeaderMutation)
-		require.Equal(t, bodyMut, commonRes.BodyMutation)
+				var expBody openai.ChatCompletionRequest
+				require.NoError(t, json.Unmarshal(someBody, &expBody))
+				mt := mockTranslator{t: t, expRequestBody: &expBody, retHeaderMutation: headerMut, retBodyMutation: bodyMut}
+				mm := &mockChatCompletionMetrics{}
+				p := &chatCompletionProcessor{
+					config: &processorConfig{
+						router:                   rt,
+						selectedBackendHeaderKey: "x-ai-gateway-backend-key",
+						modelNameHeaderKey:       "x-ai-gateway-model-key",
+					},
+					requestHeaders: headers,
+					logger:         slog.Default(),
+					metrics:        mm,
+					translator:     mt,
+				}
+				resp, err := p.ProcessRequestBody(t.Context(), &extprocv3.HttpBody{Body: someBody})
+				require.NoError(t, err)
+				require.Equal(t, mt, p.translator)
+				require.NotNil(t, resp)
+				commonRes := resp.Response.(*extprocv3.ProcessingResponse_RequestBody).RequestBody.Response
+				require.Equal(t, headerMut, commonRes.HeaderMutation)
+				require.Equal(t, bodyMut, commonRes.BodyMutation)
 
-		// Check the model and backend headers are set in headerMut.
-		hdrs := headerMut.SetHeaders
-		require.Len(t, hdrs, 2)
-		require.Equal(t, "x-ai-gateway-model-key", hdrs[0].Header.Key)
-		require.Equal(t, "some-model", string(hdrs[0].Header.RawValue))
-		require.Equal(t, "x-ai-gateway-backend-key", hdrs[1].Header.Key)
-		require.Equal(t, "some-backend", string(hdrs[1].Header.RawValue))
-	})
+				mm.RequireRequestNotCompleted(t)
+				mm.RequireSelected(t, "some-model", "some-backend")
+
+				// Check the model and backend headers are set in headerMut.
+				hdrs := headerMut.SetHeaders
+				require.Len(t, hdrs, 2)
+				require.Equal(t, "x-ai-gateway-model-key", hdrs[0].Header.Key)
+				require.Equal(t, "some-model", string(hdrs[0].Header.RawValue))
+				require.Equal(t, "x-ai-gateway-backend-key", hdrs[1].Header.Key)
+				require.Equal(t, "some-backend", string(hdrs[1].Header.RawValue))
+				require.Equal(t, stream, p.stream)
+			})
+		})
+	}
 }
 
 func TestChatCompletion_ParseBody(t *testing.T) {
diff --git a/internal/extproc/mocks_test.go b/internal/extproc/mocks_test.go
index 482371c..b709ba8 100644
--- a/internal/extproc/mocks_test.go
+++ b/internal/extproc/mocks_test.go
@@ -10,16 +10,18 @@ import (
 	"io"
 	"log/slog"
 	"testing"
+	"time"
 
 	corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
-	extprocv3http "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/ext_proc/v3"
 	extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc/metadata"
 
 	"github.com/envoyproxy/ai-gateway/filterapi"
 	"github.com/envoyproxy/ai-gateway/filterapi/x"
+	"github.com/envoyproxy/ai-gateway/internal/apischema/openai"
 	"github.com/envoyproxy/ai-gateway/internal/extproc/translator"
+	"github.com/envoyproxy/ai-gateway/internal/metrics"
 )
 
 var (
@@ -73,15 +75,14 @@ type mockTranslator struct {
 	expResponseBody   *extprocv3.HttpBody
 	retHeaderMutation *extprocv3.HeaderMutation
 	retBodyMutation   *extprocv3.BodyMutation
-	retOverride       *extprocv3http.ProcessingMode
 	retUsedToken      translator.LLMTokenUsage
 	retErr            error
 }
 
-// RequestBody implements [translator.Translator.RequestBody].
-func (m mockTranslator) RequestBody(body translator.RequestBody) (headerMutation *extprocv3.HeaderMutation, bodyMutation *extprocv3.BodyMutation, override *extprocv3http.ProcessingMode, err error) {
+// RequestBody implements [translator.OpenAIChatCompletionTranslator].
+func (m mockTranslator) RequestBody(body *openai.ChatCompletionRequest) (headerMutation *extprocv3.HeaderMutation, bodyMutation *extprocv3.BodyMutation, err error) {
 	require.Equal(m.t, m.expRequestBody, body)
-	return m.retHeaderMutation, m.retBodyMutation, m.retOverride, m.retErr
+	return m.retHeaderMutation, m.retBodyMutation, m.retErr
 }
 
 // ResponseHeaders implements [translator.Translator.ResponseHeaders].
@@ -167,3 +168,74 @@ func (m mockExternalProcessingStream) SendMsg(any) error { panic("TODO") }
 func (m mockExternalProcessingStream) RecvMsg(any) error { panic("TODO") }
 
 var _ extprocv3.ExternalProcessor_ProcessServer = &mockExternalProcessingStream{}
+
+// mockChatCompletionMetrics implements [metrics.ChatCompletion] for testing.
+type mockChatCompletionMetrics struct {
+	requestStart        time.Time
+	model               string
+	backend             string
+	requestSuccessCount int
+	requestErrorCount   int
+	tokenUsageCount     int
+	tokenLatencyCount   int
+}
+
+// StartRequest implements [metrics.ChatCompletion].
+func (m *mockChatCompletionMetrics) StartRequest() { m.requestStart = time.Now() }
+
+// SetModel implements [metrics.ChatCompletion].
+func (m *mockChatCompletionMetrics) SetModel(model string) { m.model = model }
+
+// SetBackend implements [metrics.ChatCompletion].
+func (m *mockChatCompletionMetrics) SetBackend(backend filterapi.Backend) { m.backend = backend.Name }
+
+// RecordTokenUsage implements [metrics.ChatCompletion].
+func (m *mockChatCompletionMetrics) RecordTokenUsage(_ context.Context, _, _, _ uint32) {
+	m.tokenUsageCount++
+}
+
+// RecordTokenLatency implements [metrics.ChatCompletion].
+func (m *mockChatCompletionMetrics) RecordTokenLatency(_ context.Context, _ uint32) {
+	m.tokenLatencyCount++
+}
+
+// RecordRequestCompletion implements [metrics.ChatCompletion].
+func (m *mockChatCompletionMetrics) RecordRequestCompletion(_ context.Context, success bool) {
+	if success {
+		m.requestSuccessCount++
+	} else {
+		m.requestErrorCount++
+	}
+}
+
+// RequireModelAndBackendSet asserts the model and backend set on the metrics.
+func (m *mockChatCompletionMetrics) RequireSelected(t *testing.T, model, backend string) {
+	require.Equal(t, model, m.model)
+	require.Equal(t, backend, m.backend)
+}
+
+// RequireRequestFailure asserts the request was marked as a failure.
+func (m *mockChatCompletionMetrics) RequireRequestFailure(t *testing.T) {
+	require.Equal(t, 0, m.requestSuccessCount)
+	require.Equal(t, 1, m.requestErrorCount)
+}
+
+// RequireRequestNotCompleted asserts the request was not completed.
+func (m *mockChatCompletionMetrics) RequireRequestNotCompleted(t *testing.T) {
+	require.Equal(t, 0, m.requestSuccessCount)
+	require.Equal(t, 0, m.requestErrorCount)
+}
+
+// RequireRequestSuccess asserts the request was marked as a success.
+func (m *mockChatCompletionMetrics) RequireRequestSuccess(t *testing.T) {
+	require.Equal(t, 1, m.requestSuccessCount)
+	require.Equal(t, 0, m.requestErrorCount)
+}
+
+// RequireTokensRecorded asserts the number of tokens recorded.
+func (m *mockChatCompletionMetrics) RequireTokensRecorded(t *testing.T, count int) {
+	require.Equal(t, count, m.tokenUsageCount)
+	require.Equal(t, count, m.tokenLatencyCount)
+}
+
+var _ metrics.ChatCompletion = &mockChatCompletionMetrics{}
diff --git a/internal/extproc/translator/openai_awsbedrock_test.go b/internal/extproc/translator/openai_awsbedrock_test.go
index 2cbe386..313c3eb 100644
--- a/internal/extproc/translator/openai_awsbedrock_test.go
+++ b/internal/extproc/translator/openai_awsbedrock_test.go
@@ -16,7 +16,6 @@ import (
 	"testing"
 
 	"github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream"
-	extprocv3http "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/ext_proc/v3"
 	extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	"github.com/google/go-cmp/cmp"
 	"github.com/stretchr/testify/require"
@@ -27,11 +26,6 @@ import (
 )
 
 func TestOpenAIToAWSBedrockTranslatorV1ChatCompletion_RequestBody(t *testing.T) {
-	t.Run("invalid body", func(t *testing.T) {
-		o := &openAIToAWSBedrockTranslatorV1ChatCompletion{}
-		_, _, _, err := o.RequestBody(&extprocv3.HttpBody{Body: []byte("invalid")})
-		require.Error(t, err)
-	})
 	tests := []struct {
 		name   string
 		output awsbedrock.ConverseInput
@@ -88,8 +82,11 @@ func TestOpenAIToAWSBedrockTranslatorV1ChatCompletion_RequestBody(t *testing.T)
 					},
 					{
 						Value: openai.ChatCompletionAssistantMessageParam{
-							Content: openai.ChatCompletionAssistantMessageParamContent{
-								Text: ptr.To("I dunno"),
+							Content: openai.StringOrAssistantRoleContentUnion{
+								Value: openai.ChatCompletionAssistantMessageParamContent{
+									Type: openai.ChatCompletionAssistantMessageParamContentTypeText,
+									Text: ptr.To("I dunno"),
+								},
 							},
 							ToolCalls: []openai.ChatCompletionMessageToolCallParam{
 								{
@@ -103,6 +100,13 @@ func TestOpenAIToAWSBedrockTranslatorV1ChatCompletion_RequestBody(t *testing.T)
 							},
 						}, Type: openai.ChatMessageRoleAssistant,
 					},
+					{
+						Value: openai.ChatCompletionAssistantMessageParam{
+							Content: openai.StringOrAssistantRoleContentUnion{
+								Value: "I also dunno",
+							},
+						}, Type: openai.ChatMessageRoleAssistant,
+					},
 				},
 			},
 			output: awsbedrock.ConverseInput{
@@ -170,6 +174,14 @@ func TestOpenAIToAWSBedrockTranslatorV1ChatCompletion_RequestBody(t *testing.T)
 							},
 						},
 					},
+					{
+						Role: openai.ChatMessageRoleAssistant,
+						Content: []*awsbedrock.ContentBlock{
+							{
+								Text: ptr.To("I also dunno"),
+							},
+						},
+					},
 				},
 			},
 		},
@@ -688,18 +700,13 @@ func TestOpenAIToAWSBedrockTranslatorV1ChatCompletion_RequestBody(t *testing.T)
 		t.Run(tt.name, func(t *testing.T) {
 			o := &openAIToAWSBedrockTranslatorV1ChatCompletion{}
 			originalReq := tt.input
-			hm, bm, mode, err := o.RequestBody(RequestBody(&originalReq))
+			hm, bm, err := o.RequestBody(&originalReq)
 			var expPath string
+			require.Equal(t, tt.input.Stream, o.stream)
 			if tt.input.Stream {
 				expPath = fmt.Sprintf("/model/%s/converse-stream", tt.input.Model)
-				require.True(t, o.stream)
-				require.NotNil(t, mode)
-				require.Equal(t, extprocv3http.ProcessingMode_STREAMED, mode.ResponseBodyMode)
-				require.Equal(t, extprocv3http.ProcessingMode_SEND, mode.ResponseHeaderMode)
 			} else {
 				expPath = fmt.Sprintf("/model/%s/converse", tt.input.Model)
-				require.False(t, o.stream)
-				require.Nil(t, mode)
 			}
 			require.NoError(t, err)
 			require.NotNil(t, hm)
diff --git a/internal/extproc/translator/openai_openai_test.go b/internal/extproc/translator/openai_openai_test.go
index 05d0595..4364210 100644
--- a/internal/extproc/translator/openai_openai_test.go
+++ b/internal/extproc/translator/openai_openai_test.go
@@ -13,7 +13,6 @@ import (
 	"strconv"
 	"testing"
 
-	extprocv3http "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/ext_proc/v3"
 	extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	"github.com/google/go-cmp/cmp"
 	"github.com/stretchr/testify/require"
@@ -23,28 +22,16 @@ import (
 )
 
 func TestOpenAIToOpenAITranslatorV1ChatCompletionRequestBody(t *testing.T) {
-	t.Run("invalid body", func(t *testing.T) {
-		o := &openAIToOpenAITranslatorV1ChatCompletion{}
-		_, _, _, err := o.RequestBody(&extprocv3.HttpBody{Body: []byte("invalid")})
-		require.Error(t, err)
-	})
 	t.Run("valid body", func(t *testing.T) {
 		for _, stream := range []bool{true, false} {
 			t.Run(fmt.Sprintf("stream=%t", stream), func(t *testing.T) {
 				originalReq := &openai.ChatCompletionRequest{Model: "foo-bar-ai", Stream: stream}
 
 				o := &openAIToOpenAITranslatorV1ChatCompletion{}
-				hm, bm, mode, err := o.RequestBody(RequestBody(originalReq))
+				hm, bm, err := o.RequestBody(originalReq)
 				require.Nil(t, bm)
 				require.NoError(t, err)
 				require.Equal(t, stream, o.stream)
-				if stream {
-					require.NotNil(t, mode)
-					require.Equal(t, extprocv3http.ProcessingMode_SEND, mode.ResponseHeaderMode)
-					require.Equal(t, extprocv3http.ProcessingMode_STREAMED, mode.ResponseBodyMode)
-				} else {
-					require.Nil(t, mode)
-				}
 
 				require.Nil(t, hm)
 			})
diff --git a/internal/metrics/chat_completion_metrics_test.go b/internal/metrics/chat_completion_metrics_test.go
new file mode 100644
index 0000000..8c24948
--- /dev/null
+++ b/internal/metrics/chat_completion_metrics_test.go
@@ -0,0 +1,177 @@
+// Copyright Envoy AI Gateway Authors
+// SPDX-License-Identifier: Apache-2.0
+// The full text of the Apache license is available in the LICENSE file at
+// the root of the repo.
+
+package metrics
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+
+	"github.com/envoyproxy/ai-gateway/filterapi"
+)
+
+func TestNewProcessorMetrics(t *testing.T) {
+	var (
+		mr    = metric.NewManualReader()
+		meter = metric.NewMeterProvider(metric.WithReader(mr)).Meter("test")
+		pm    = NewChatCompletion(meter).(*chatCompletion)
+	)
+
+	assert.NotNil(t, pm)
+	assert.False(t, pm.firstTokenSent)
+}
+
+func TestStartRequest(t *testing.T) {
+	var (
+		mr    = metric.NewManualReader()
+		meter = metric.NewMeterProvider(metric.WithReader(mr)).Meter("test")
+		pm    = NewChatCompletion(meter).(*chatCompletion)
+	)
+
+	before := time.Now()
+	pm.StartRequest()
+	after := time.Now()
+
+	assert.False(t, pm.firstTokenSent)
+	assert.GreaterOrEqual(t, pm.requestStart, before)
+	assert.LessOrEqual(t, pm.requestStart, after)
+}
+
+func TestRecordTokenUsage(t *testing.T) {
+	var (
+		mr    = metric.NewManualReader()
+		meter = metric.NewMeterProvider(metric.WithReader(mr)).Meter("test")
+		pm    = NewChatCompletion(meter).(*chatCompletion)
+
+		attrs = []attribute.KeyValue{
+			attribute.Key(genaiAttributeOperationName).String(genaiOperationChat),
+			attribute.Key(genaiAttributeSystemName).String(genaiSystemOpenAI),
+			attribute.Key(genaiAttributeRequestModel).String("test-model"),
+		}
+		inputAttrs  = attribute.NewSet(append(attrs, attribute.Key(genaiAttributeTokenType).String(genaiTokenTypeInput))...)
+		outputAttrs = attribute.NewSet(append(attrs, attribute.Key(genaiAttributeTokenType).String(genaiTokenTypeOutput))...)
+		totalAttrs  = attribute.NewSet(append(attrs, attribute.Key(genaiAttributeTokenType).String(genaiTokenTypeTotal))...)
+	)
+
+	pm.SetModel("test-model")
+	pm.SetBackend(filterapi.Backend{Schema: filterapi.VersionedAPISchema{Name: filterapi.APISchemaOpenAI}})
+	pm.RecordTokenUsage(t.Context(), 10, 5, 15)
+
+	count, sum := getHistogramValues(t, mr, genaiMetricClientTokenUsage, inputAttrs)
+	assert.Equal(t, uint64(1), count)
+	assert.Equal(t, 10.0, sum)
+
+	count, sum = getHistogramValues(t, mr, genaiMetricClientTokenUsage, outputAttrs)
+	assert.Equal(t, uint64(1), count)
+	assert.Equal(t, 5.0, sum)
+
+	count, sum = getHistogramValues(t, mr, genaiMetricClientTokenUsage, totalAttrs)
+	assert.Equal(t, uint64(1), count)
+	assert.Equal(t, 15.0, sum)
+}
+
+func TestRecordTokenLatency(t *testing.T) {
+	var (
+		mr    = metric.NewManualReader()
+		meter = metric.NewMeterProvider(metric.WithReader(mr)).Meter("test")
+		pm    = NewChatCompletion(meter).(*chatCompletion)
+
+		attrs = attribute.NewSet(
+			attribute.Key(genaiAttributeOperationName).String(genaiOperationChat),
+			attribute.Key(genaiAttributeSystemName).String(genAISystemAWSBedrock),
+			attribute.Key(genaiAttributeRequestModel).String("test-model"),
+		)
+	)
+
+	pm.StartRequest()
+	pm.SetModel("test-model")
+	pm.SetBackend(filterapi.Backend{Schema: filterapi.VersionedAPISchema{Name: filterapi.APISchemaAWSBedrock}})
+
+	// Test first token.
+	time.Sleep(10 * time.Millisecond)
+	pm.RecordTokenLatency(t.Context(), 1)
+	assert.True(t, pm.firstTokenSent)
+	count, sum := getHistogramValues(t, mr, genaiMetricServerTimeToFirstToken, attrs)
+	assert.Equal(t, uint64(1), count)
+	assert.Greater(t, sum, 0.0)
+
+	// Test subsequent tokens.
+	time.Sleep(10 * time.Millisecond)
+	pm.RecordTokenLatency(t.Context(), 5)
+	count, sum = getHistogramValues(t, mr, genaiMetricServerTimePerOutputToken, attrs)
+	assert.Equal(t, uint64(1), count)
+	assert.Greater(t, sum, 0.0)
+
+	// Test zero tokens case.
+	time.Sleep(10 * time.Millisecond)
+	pm.RecordTokenLatency(t.Context(), 0)
+	count, sum = getHistogramValues(t, mr, genaiMetricServerTimePerOutputToken, attrs)
+	assert.Equal(t, uint64(1), count)
+	assert.Greater(t, sum, 0.0)
+}
+
+func TestRecordRequestCompletion(t *testing.T) {
+	var (
+		mr    = metric.NewManualReader()
+		meter = metric.NewMeterProvider(metric.WithReader(mr)).Meter("test")
+		pm    = NewChatCompletion(meter).(*chatCompletion)
+
+		attrs = []attribute.KeyValue{
+			attribute.Key(genaiAttributeOperationName).String(genaiOperationChat),
+			attribute.Key(genaiAttributeSystemName).String("custom"),
+			attribute.Key(genaiAttributeRequestModel).String("test-model"),
+		}
+		attrsSuccess = attribute.NewSet(attrs...)
+		attrsFailure = attribute.NewSet(append(attrs, attribute.Key(genaiAttributeErrorType).String(genaiErrorTypeFallback))...)
+	)
+
+	pm.StartRequest()
+	pm.SetModel("test-model")
+	pm.SetBackend(filterapi.Backend{Name: "custom"})
+
+	time.Sleep(10 * time.Millisecond)
+	pm.RecordRequestCompletion(t.Context(), true)
+	count, sum := getHistogramValues(t, mr, genaiMetricServerRequestDuration, attrsSuccess)
+	assert.Equal(t, uint64(1), count)
+	assert.Greater(t, sum, 0.0)
+
+	// Test some failed requests.
+	pm.RecordRequestCompletion(t.Context(), false)
+	pm.RecordRequestCompletion(t.Context(), false)
+	count, sum = getHistogramValues(t, mr, genaiMetricServerRequestDuration, attrsFailure)
+	assert.Equal(t, uint64(2), count)
+	assert.Greater(t, sum, 0.0)
+}
+
+// getHistogramValues returns the count and sum of a histogram metric with the given attributes.
+func getHistogramValues(t *testing.T, reader metric.Reader, metric string, attrs attribute.Set) (uint64, float64) {
+	var data metricdata.ResourceMetrics
+	require.NoError(t, reader.Collect(t.Context(), &data))
+
+	var datapoints []metricdata.HistogramDataPoint[float64]
+	for _, sm := range data.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			if m.Name != metric {
+				continue
+			}
+			data := m.Data.(metricdata.Histogram[float64])
+			for _, dp := range data.DataPoints {
+				if dp.Attributes.Equals(&attrs) {
+					datapoints = append(datapoints, dp)
+				}
+			}
+		}
+	}
+
+	require.Len(t, datapoints, 1, "found %d datapoints for attributes: %v", len(datapoints), attrs)
+
+	return datapoints[0].Count, datapoints[0].Sum
+}
diff --git a/tests/extproc/extproc_test.go b/tests/extproc/extproc_test.go
index 5f47984..8b1d7b2 100644
--- a/tests/extproc/extproc_test.go
+++ b/tests/extproc/extproc_test.go
@@ -42,7 +42,7 @@ func requireExtProc(t *testing.T, stdout io.Writer, executable, configPath strin
 	cmd := exec.CommandContext(t.Context(), executable)
 	cmd.Stdout = stdout
 	cmd.Stderr = os.Stderr
-	cmd.Args = append(cmd.Args, "-configPath", configPath)
+	cmd.Args = append(cmd.Args, "-configPath", configPath, "-logLevel", "debug")
 	cmd.Env = append(os.Environ(), envs...)
 	require.NoError(t, cmd.Start())
 }
@@ -69,6 +69,7 @@ func requireRunEnvoy(t *testing.T, accessLogPath string) {
 	cmd := exec.CommandContext(t.Context(), "envoy",
 		"-c", envoyYamlPath,
 		"--log-level", "warn",
+		"--component-log-level", "ext_proc:debug",
 		"--concurrency", strconv.Itoa(max(runtime.NumCPU(), 2)),
 	)
 	cmd.Stdout = os.Stdout
diff --git a/tests/extproc/testupstream_test.go b/tests/extproc/testupstream_test.go
index fb72871..9e7890c 100644
--- a/tests/extproc/testupstream_test.go
+++ b/tests/extproc/testupstream_test.go
@@ -19,6 +19,8 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	openaigo "github.com/openai/openai-go"
+	"github.com/openai/openai-go/option"
 	"github.com/stretchr/testify/require"
 
 	"github.com/envoyproxy/ai-gateway/filterapi"
@@ -272,6 +274,70 @@ data: [DONE]
 			}, 10*time.Second, 500*time.Millisecond)
 		})
 	}
+
+	t.Run("stream non blocking", func(t *testing.T) {
+		// This receives a stream of 20 event messages. The testuptream server sleeps 200 ms between each message.
+		// Therefore, if envoy fails to process the response in a streaming manner, the test will fail taking more than 4 seconds.
+		client := openaigo.NewClient(
+			option.WithBaseURL(listenerAddress+"/v1/"),
+			option.WithHeader("x-test-backend", "openai"),
+			option.WithHeader(testupstreamlib.ResponseTypeKey, "sse"),
+			option.WithHeader(testupstreamlib.ResponseBodyHeaderKey,
+				base64.StdEncoding.EncodeToString([]byte(
+					`
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" This"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" is"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" a"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" test"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":"."},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" This"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" is"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" a"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" test"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":"."},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" This"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" is"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" a"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" test"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":"."},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" This"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" is"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" a"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":" test"},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[{"index":0,"delta":{"content":"."},"logprobs":null,"finish_reason":null}],"usage":null}
+{"id":"chatcmpl-B8ZKlXBoEXZVTtv3YBmewxuCpNW7b","object":"chat.completion.chunk","created":1741382147,"model":"gpt-4o-mini-2024-07-18","service_tier":"default","system_fingerprint":"fp_06737a9306","choices":[],"usage":{"prompt_tokens":25,"completion_tokens":61,"total_tokens":86,"prompt_tokens_details":{"cached_tokens":0,"audio_tokens":0},"completion_tokens_details":{"reasoning_tokens":0,"audio_tokens":0,"accepted_prediction_tokens":0,"rejected_prediction_tokens":0}}}
+[DONE]
+`,
+				))),
+		)
+
+		// NewStreaming below will block until the first event is received, so take the time before calling it.
+		start := time.Now()
+		stream := client.Chat.Completions.NewStreaming(t.Context(), openaigo.ChatCompletionNewParams{
+			Messages: openaigo.F([]openaigo.ChatCompletionMessageParamUnion{
+				openaigo.UserMessage("Say this is a test"),
+			}),
+			Model: openaigo.F("something"),
+		})
+		defer func() {
+			_ = stream.Close()
+		}()
+
+		asserted := false
+		for stream.Next() {
+			chunk := stream.Current()
+			if len(chunk.Choices) == 0 || chunk.Choices[0].Delta.Content == "" {
+				continue
+			}
+			t.Logf("%v: %v", time.Now(), chunk.Choices[0].Delta.Content)
+			// Check each event is received less than a second after the previous one.
+			require.Less(t, time.Since(start), time.Second)
+			start = time.Now()
+			asserted = true
+		}
+		require.True(t, asserted)
+		require.NoError(t, stream.Err())
+	})
 }
 
 func checkModelsIgnoringTimestamps(want openai.ModelList) func(t require.TestingT, body []byte) {