feat: add ResponseChunkProcessor interface for streaming response processing

Split response processing into two interfaces per review feedback:

- ResponseProcessor: processes the complete buffered response body (existing)
- ResponseChunkProcessor: processes individual chunks without buffering (new)

The Profile now carries both processor lists and a NeedsResponseBuffering
flag. The config loader sorts response plugins into the right list based
on which interface they implement. A plugin can implement both.

When ResponseProcessor plugins are present → buffer the full response.
When only ResponseChunkProcessors are present → stream chunks through.
When neither is present → buffer (backward compatible).

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

Author: noyitz

pkg/config/loader/configloader.go

@@ -214,9 +214,7 @@ func buildProfiles(rawProfiles []configapi.Profile, handle plugin.Handle) (map[s
 			return nil, fmt.Errorf("the profile %s must have one or both of the Request and Response sections", rawProfile.Name)
 		}
 
-		theProfile := requesthandling.Profile{
-			ResponsePlugins: make([]requesthandling.ResponseProcessor, len(rawProfile.Plugins.Response)),
-		}
+		theProfile := requesthandling.Profile{}
 
 		for _, pluginRef := range rawProfile.Plugins.Request {
 			rawPlugin := handle.Plugin(pluginRef.PluginRef)
@@ -245,17 +243,25 @@ func buildProfiles(rawProfiles []configapi.Profile, handle plugin.Handle) (map[s
 			}
 		}
 
-		for idx, pluginRef := range rawProfile.Plugins.Response {
+		for _, pluginRef := range rawProfile.Plugins.Response {
 			rawPlugin := handle.Plugin(pluginRef.PluginRef)
 			if rawPlugin == nil {
 				return nil, fmt.Errorf("there is no plugin named %s", pluginRef.PluginRef)
 			}
-			thePlugin, ok := rawPlugin.(requesthandling.ResponseProcessor)
-			if !ok {
-				return nil, fmt.Errorf("the plugin named %s is not a ResponseProcessor", pluginRef.PluginRef)
+			matched := false
+			if bodyPlugin, ok := rawPlugin.(requesthandling.ResponseProcessor); ok {
+				theProfile.ResponsePlugins = append(theProfile.ResponsePlugins, bodyPlugin)
+				matched = true
+			}
+			if chunkPlugin, ok := rawPlugin.(requesthandling.ResponseChunkProcessor); ok {
+				theProfile.ResponseChunkProcessors = append(theProfile.ResponseChunkProcessors, chunkPlugin)
+				matched = true
+			}
+			if !matched {
+				return nil, fmt.Errorf("the plugin named %s is not a ResponseProcessor or ResponseChunkProcessor", pluginRef.PluginRef)
 			}
-			theProfile.ResponsePlugins[idx] = thePlugin
 		}
+		theProfile.NeedsResponseBuffering = len(theProfile.ResponsePlugins) > 0
 
 		profiles[rawProfile.Name] = &theProfile
 	}

pkg/framework/interface/requesthandling/plugins.go

@@ -43,13 +43,22 @@ type RequestProcessor interface {
 	ProcessRequest(ctx context.Context, cycleState *plugin.CycleState, request *InferenceRequest) error
 }
 
+// ResponseProcessor processes the complete buffered response body.
+// If any plugin in a profile implements this interface, the framework buffers
+// the entire response before calling ProcessResponse on each such plugin.
 type ResponseProcessor interface {
 	plugin.Plugin
-	// ProcessResponse runs the ResponseProcessor plugin.
-	// ResponseProcessor can mutate the headers and/or the body of the response.
 	ProcessResponse(ctx context.Context, cycleState *plugin.CycleState, response *InferenceResponse) error
 }
 
+// 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.
+type ResponseChunkProcessor interface {
+	plugin.Plugin
+	ProcessResponseChunk(ctx context.Context, cycleState *plugin.CycleState, chunk []byte, isFinal bool) ([]byte, error)
+}
+
 type PostProcessor interface {
 	plugin.Plugin
 

pkg/framework/interface/requesthandling/types.go

@@ -114,9 +114,14 @@ type Profile struct {
 	// RequestPlugins are the request processing plugin instances executed by the request handler,
 	// in the same order provided in the configuration file.
 	RequestPlugins []RequestProcessor
-	// ResponsePlugins are the response processing plugin instances executed by the response handler,
-	// in the same order provided in the configuration file.
+	// ResponsePlugins process the complete buffered response body.
 	ResponsePlugins []ResponseProcessor
+	// ResponseChunkProcessors process individual response chunks without buffering.
+	ResponseChunkProcessors []ResponseChunkProcessor
+	// NeedsResponseBuffering is true when any ResponsePlugin is present.
+	// The framework uses this to decide whether to buffer the full response body
+	// or stream chunks through ResponseChunkProcessors.
+	NeedsResponseBuffering bool
 	// ModelSelectorPlugins are the Filter, Scorer (including WeightedScorer), and Picker plugin
 	// instances to be wired into any model-selector plugin present in RequestPlugins.
 	ModelSelectorPlugins []plugin.Plugin

pkg/handlers/response.go

@@ -131,6 +131,56 @@ func (s *Server) generateEmptyResponseBodyResponse(responseBodyBytes []byte) []*
 	return responses
 }
 
+// buildChunkPassthroughResponse wraps a single response body chunk in the
+// ext_proc streaming response format. Used when no ResponseProcessor (buffered)
+// plugins are present — chunks flow through ResponseChunkProcessors and are
+// forwarded immediately.
+func (s *Server) buildChunkPassthroughResponse(chunk []byte, isFinal bool) []*eppb.ProcessingResponse {
+	if isFinal {
+		return []*eppb.ProcessingResponse{
+			{
+				Response: &eppb.ProcessingResponse_ResponseHeaders{
+					ResponseHeaders: &eppb.HeadersResponse{},
+				},
+			},
+			{
+				Response: &eppb.ProcessingResponse_ResponseBody{
+					ResponseBody: &eppb.BodyResponse{
+						Response: &eppb.CommonResponse{
+							BodyMutation: &eppb.BodyMutation{
+								Mutation: &eppb.BodyMutation_StreamedResponse{
+									StreamedResponse: &eppb.StreamedBodyResponse{
+										Body:        chunk,
+										EndOfStream: true,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		}
+	}
+	return []*eppb.ProcessingResponse{
+		{
+			Response: &eppb.ProcessingResponse_ResponseBody{
+				ResponseBody: &eppb.BodyResponse{
+					Response: &eppb.CommonResponse{
+						BodyMutation: &eppb.BodyMutation{
+							Mutation: &eppb.BodyMutation_StreamedResponse{
+								StreamedResponse: &eppb.StreamedBodyResponse{
+									Body:        chunk,
+									EndOfStream: false,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
 // HandleResponseTrailers handles response trailers.
 func (s *Server) HandleResponseTrailers(trailers *eppb.HttpTrailers) ([]*eppb.ProcessingResponse, error) {
 	return []*eppb.ProcessingResponse{

pkg/handlers/server.go

@@ -176,15 +176,39 @@ func (s *Server) Process(srv extProcPb.ExternalProcessor_ProcessServer) error {
 			if reqCtx.ResponseFirstChunkTimestamp.IsZero() {
 				reqCtx.ResponseFirstChunkTimestamp = time.Now()
 			}
-			responseBody = append(responseBody, v.ResponseBody.Body...)
-			if !v.ResponseBody.EndOfStream {
-				continue
+
+			hasChunkProcessors := reqCtx.Profile != nil && len(reqCtx.Profile.ResponseChunkProcessors) > 0
+			needsBuffering := !hasChunkProcessors
+			if reqCtx.Profile != nil && reqCtx.Profile.NeedsResponseBuffering {
+				needsBuffering = true
+			}
+			if needsBuffering {
+				responseBody = append(responseBody, v.ResponseBody.Body...)
+				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 {
+				chunk := v.ResponseBody.Body
+				isFinal := v.ResponseBody.EndOfStream
+				if reqCtx.Profile != nil {
+					for _, cp := range reqCtx.Profile.ResponseChunkProcessors {
+						chunk, err = cp.ProcessResponseChunk(ctx, reqCtx.CycleState, chunk, isFinal)
+						if err != nil {
+							break
+						}
+					}
+				}
+				if isFinal {
+					reqCtx.ResponseCompleteTimestamp = time.Now()
+				}
+				responses = s.buildChunkPassthroughResponse(chunk, isFinal)
+				loggerVerbose.Info("response body streaming pass-through complete")
 			}
-			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")
 		case *extProcPb.ProcessingRequest_ResponseTrailers:
 			responses, err = s.HandleResponseTrailers(v.ResponseTrailers)
 		default: