fix: address review — TTFT metric, chunk interface, nil checks

- Move TTFT metric recording after if/else so it applies to both
  buffered and streaming paths (comment 3+4)
- Change ResponseChunkProcessor interface: takes string (framework
  converts once) + InferenceResponse (for header mutation) (comment 5)
- Add nil profile check and empty processor check in HandleResponseChunk
  for bodiless requests like GET /v1/models (comment 6)
- Extract runResponseChunkProcessors and buildStreamedChunkResponse
  following the same patterns as the body processing path

Signed-off-by: Noy Itzikowitz <nitzikow@redhat.com>

Author: noyitz

pkg/framework/interface/requesthandling/plugins.go

@@ -52,11 +52,12 @@ type ResponseProcessor interface {
 }
 
 // ResponseChunkProcessor processes individual response body chunks as they
-// stream through without buffering. The framework calls ProcessResponseChunk
-// for each chunk; the returned bytes are forwarded to the client.
+// stream through without buffering. The framework converts the raw chunk bytes
+// to a string once and passes it to all chunk processors. Plugins receive the
+// InferenceResponse to allow header mutation.
 type ResponseChunkProcessor interface {
 	plugin.Plugin
-	ProcessResponseChunk(ctx context.Context, cycleState *plugin.CycleState, chunk []byte, isFinal bool) ([]byte, error)
+	ProcessResponseChunk(ctx context.Context, cycleState *plugin.CycleState, response *InferenceResponse, chunk string, isFinal bool) error
 }
 
 type PostProcessor interface {

pkg/handlers/response.go

@@ -133,24 +133,45 @@ func (s *Server) generateEmptyResponseBodyResponse(responseBodyBytes []byte) []*
 
 // HandleResponseChunk runs ResponseChunkProcessors on a single response body chunk
 // and wraps the result in the ext_proc streaming response format.
-func (s *Server) HandleResponseChunk(ctx context.Context, reqCtx *RequestContext, chunk []byte, endOfStream bool) ([]*eppb.ProcessingResponse, error) {
+func (s *Server) HandleResponseChunk(ctx context.Context, reqCtx *RequestContext, chunkBytes []byte, endOfStream bool) ([]*eppb.ProcessingResponse, error) {
+	// Bodiless requests (e.g., GET /v1/models) may not have a profile set.
+	if reqCtx.Profile == nil || len(reqCtx.Profile.ResponseChunkProcessors) == 0 {
+		return s.buildStreamedChunkResponse(chunkBytes, endOfStream), nil
+	}
+
+	logger := log.FromContext(ctx).V(logutil.DEFAULT)
+
+	chunk := string(chunkBytes)
+
+	if err := s.runResponseChunkProcessors(ctx, reqCtx.CycleState, reqCtx.Response, chunk, endOfStream, reqCtx.Profile.ResponseChunkProcessors); err != nil {
+		logger.Error(err, "Failed to run response chunk processors")
+		return nil, err
+	}
+
+	return s.buildStreamedChunkResponse(chunkBytes, endOfStream), nil
+}
+
+// runResponseChunkProcessors executes chunk processors in the order they were registered.
+func (s *Server) runResponseChunkProcessors(ctx context.Context, cycleState *plugin.CycleState, response *requesthandling.InferenceResponse, chunk string, isFinal bool, processors []requesthandling.ResponseChunkProcessor) error {
 	logger := log.FromContext(ctx).V(logutil.DEFAULT)
 	verboseLogger := logger.V(logutil.VERBOSE)
 
-	for _, cp := range reqCtx.Profile.ResponseChunkProcessors {
+	for _, cp := range processors {
 		if verboseLogger.Enabled() {
 			verboseLogger.Info("Executing response chunk plugin", "plugin", cp.TypedName())
 		}
-		var err error
 		before := time.Now()
-		chunk, err = cp.ProcessResponseChunk(ctx, reqCtx.CycleState, chunk, endOfStream)
+		err := cp.ProcessResponseChunk(ctx, cycleState, response, chunk, isFinal)
 		metrics.RecordPluginProcessingLatency(responsePluginExtensionPoint, cp.TypedName().Type, cp.TypedName().Name, time.Since(before))
 		if err != nil {
-			logger.Error(err, "Failed to execute response chunk plugin", "plugin", cp.TypedName())
-			return nil, err
+			return err
 		}
 	}
+	return nil
+}
 
+// buildStreamedChunkResponse wraps a chunk in the ext_proc streaming response format.
+func (s *Server) buildStreamedChunkResponse(chunk []byte, endOfStream bool) []*eppb.ProcessingResponse {
 	responses := []*eppb.ProcessingResponse{
 		{
 			Response: &eppb.ProcessingResponse_ResponseBody{
@@ -179,7 +200,7 @@ func (s *Server) HandleResponseChunk(ctx context.Context, reqCtx *RequestContext
 		responses = append([]*eppb.ProcessingResponse{headerResp}, responses...)
 	}
 
-	return responses, nil
+	return responses
 }
 
 // HandleResponseTrailers handles response trailers.

pkg/handlers/server.go

@@ -182,18 +182,18 @@ func (s *Server) Process(srv extProcPb.ExternalProcessor_ProcessServer) error {
 				if !v.ResponseBody.EndOfStream {
 					continue
 				}
-				reqCtx.ResponseCompleteTimestamp = time.Now()
-				model, _ := reqCtx.Request.Body["model"].(string)
-				metrics.RecordRequestTTFT(model, reqCtx.ResponseFirstChunkTimestamp.Sub(reqCtx.RequestReceivedTimestamp))
 				responses, err = s.HandleResponseBody(ctx, reqCtx, responseBody)
 				loggerVerbose.Info("processing response body complete")
 			} else {
-				if v.ResponseBody.EndOfStream {
-					reqCtx.ResponseCompleteTimestamp = time.Now()
-				}
 				responses, err = s.HandleResponseChunk(ctx, reqCtx, v.ResponseBody.Body, v.ResponseBody.EndOfStream)
 				loggerVerbose.Info("response chunk processing complete")
 			}
+
+			if v.ResponseBody.EndOfStream {
+				reqCtx.ResponseCompleteTimestamp = time.Now()
+				model, _ := reqCtx.Request.Body["model"].(string)
+				metrics.RecordRequestTTFT(model, reqCtx.ResponseFirstChunkTimestamp.Sub(reqCtx.RequestReceivedTimestamp))
+			}
 		case *extProcPb.ProcessingRequest_ResponseTrailers:
 			responses, err = s.HandleResponseTrailers(v.ResponseTrailers)
 		default: