Allow skipped requests to go through scheduling profiles (#1387)

* make sure skipped request is can also go through profiles

Signed-off-by: bobzetian <bobzetian@google.com>

* test(server): add assertions for skipped request routing

Signed-off-by: bobzetian <bobzetian@google.com>

* epp: unify skip routing and enforce RawPayload globally

EPP previously had inconsistent behavior when request parsing was skipped
(e.g., for unsupported gRPC paths in the vllmgrpc parser). The vllmgrpc
parser returned a nil body on skip, which caused nil-pointer panics in
the Scheduling Director when it attempted model rewrite or repackaging.

This change unifies the skip story by enforcing a non-nil body with
RawPayload globally at the EPP framework level (in server.go) whenever
a parser signals Skip. This ensures the Director can safely execute
smart routing, admission control, and subsetting for skipped requests
without panics, while still bypassing the expensive response-phase
interception.

The RequestSkipped state comments and logging have also been improved
to accurately reflect that skipped requests are smartly routed by the
director rather than falling back to random endpoints.

TEST=make test-unit

Signed-off-by: bobzetian <bobzetian@google.com>

* epp: simplify vertexai parser to return nil body on skip

Following the global Skip payload-force implementation in server.go,
individual parsers no longer need to create and return a dummy body
with RawPayload on skip.

This change simplifies the vertexai parser to return nil body on skip,
aligning it with the vllmgrpc parser and keeping parser implementations
boilerplate-free. The vertexai unit test has also been updated.

TEST=make test-unit

Signed-off-by: bobzetian <bobzetian@google.com>

* refactor: rename Skip to SkipResponseProcessing and RequestSkipped to RequestResponseProcessingSkipped

To improve readability, code clarity, and explicitly communicate the
architectural intent of EPP skip behavior, this change renames the
Skip field in the Parser ParseResult struct to SkipResponseProcessing.

Additionally, the corresponding internal state constant in server.go
is renamed from RequestSkipped to RequestResponseProcessingSkipped.

All parser implementations (openai, passthrough, vertexai, vllmgrpc)
and their corresponding unit tests, along with the main server unit
tests, have been updated to reflect this complete renaming.

TEST=make test-unit

Signed-off-by: bobzetian <bobzetian@google.com>

* enforce populate Body.Payload when SkipResponseProcessing is true

Signed-off-by: bobzetian <bobzetian@google.com>

---------

Signed-off-by: bobzetian <bobzetian@google.com>

Author: zetxqx

pkg/epp/framework/interface/requesthandling/plugins.go

@@ -30,10 +30,10 @@ type Parser interface {
 	// ParseRequest parses the request body and headers and returns the parsed result.
 	// There are three outcomes based on the return values:
 	// 1. err != nil: The request is invalid or cannot be parsed. The framework will fail the request early.
-	// 2. err == nil and result.Skip == true: The request is valid but should not be processed by the scheduling
-	//    framework (e.g., non-inference requests). The framework will fall back to routing the request to a random
-	//    endpoint and skip subsequent interception phases.
-	// 3. err == nil and result.Skip == false: The request is valid and will be processed by the scheduling framework.
+	// 2. err == nil and result.SkipResponseProcessing == true: The request is valid but EPP should stop intercepting the stream
+	//    after the request phase. The scheduling director will still route the request, but subsequent
+	//    response interception phases will be skipped.
+	// 3. err == nil and result.SkipResponseProcessing == false: The request is valid and will be processed by the scheduling framework.
 	ParseRequest(ctx context.Context, body []byte, headers map[string]string) (*ParseResult, error)
 
 	// ParseResponse parses the response payload.
@@ -52,16 +52,15 @@ type Parser interface {
 type ParseResult struct {
 	// Body contains the parsed inference request body.
 	Body *InferenceRequestBody
-	// Skip indicates whether to skip the remaining processing phases in the EPP.
-	// This should only be used for non-inference related requests.
-	// When set to true, the request will bypass the scheduling director and will be
-	// routed to a random endpoint. The EPP will also stop intercepting the stream
-	// for this request (e.g., response headers and body will not be processed).
+	// SkipResponseProcessing indicates whether to skip EPP stream interception for this request.
+	// When set to true, the request will still go through the scheduling director
+	// (allowing routing decisions, profiles, and admission control to run),
+	// but the EPP will stop intercepting the stream after the request phase completes
+	// (e.g., response headers and body will not be processed).
 	//
-	// Example: In a gRPC parser, if the request path does not match any known
-	// methods (e.g., neither Generate nor Embed for vLLM), setting Skip to true
-	// allows the request to proceed to a fallback endpoint without failing.
-	Skip bool
+	// This allows fallback or non-standard requests to be routed using the configured
+	// policies without paying the overhead of response-phase interception.
+	SkipResponseProcessing bool
 }
 
 type ParsedResponse struct {

pkg/epp/framework/plugins/requesthandling/parsers/anthropic/anthropic.go

@@ -101,7 +101,7 @@ func (p *AnthropicParser) ParseRequest(_ context.Context, body []byte, headers m
 		result.Stream = true
 	}
 
-	return &fwkrh.ParseResult{Body: result, Skip: false}, nil
+	return &fwkrh.ParseResult{Body: result, SkipResponseProcessing: false}, nil
 }
 
 func (p *AnthropicParser) ParseResponse(_ context.Context, body []byte, headers map[string]string, _ bool) (*fwkrh.ParsedResponse, error) {

pkg/epp/framework/plugins/requesthandling/parsers/anthropic/anthropic_test.go

@@ -432,8 +432,8 @@ func TestAnthropicParser_ParseRequest(t *testing.T) {
 				return
 			}
 
-			if got.Skip != false {
-				t.Errorf("ParseRequest() got.Skip = %v, want false", got.Skip)
+			if got.SkipResponseProcessing != false {
+				t.Errorf("ParseRequest() got.SkipResponseProcessing = %v, want false", got.SkipResponseProcessing)
 			}
 
 			if diff := cmp.Diff(tt.want, got.Body); diff != "" {

pkg/epp/framework/plugins/requesthandling/parsers/openai/openai.go

@@ -110,7 +110,7 @@ func (p *OpenAIParser) ParseRequest(ctx context.Context, body []byte, headers ma
 	if stream, ok := bodyMap["stream"].(bool); ok && stream {
 		extractedBody.Stream = true
 	}
-	return &fwkrh.ParseResult{Body: extractedBody, Skip: false}, nil
+	return &fwkrh.ParseResult{Body: extractedBody, SkipResponseProcessing: false}, nil
 }
 
 // ParseResponse extracts usage metadata from the provider's response.

pkg/epp/framework/plugins/requesthandling/parsers/openai/openai_test.go

@@ -809,8 +809,8 @@ func TestOpenAIParser_ParseRequest(t *testing.T) {
 				return
 			}
 
-			if got.Skip != false {
-				t.Errorf("ParseRequest() got.Skip = %v, want false", got.Skip)
+			if got.SkipResponseProcessing != false {
+				t.Errorf("ParseRequest() got.SkipResponseProcessing = %v, want false", got.SkipResponseProcessing)
 			}
 
 			if diff := cmp.Diff(tt.want, got.Body); diff != "" {

pkg/epp/framework/plugins/requesthandling/parsers/passthrough/passthrough.go

@@ -72,7 +72,7 @@ func (p *PassthroughParser) ParseRequest(ctx context.Context, body []byte, heade
 		Body: &fwkrh.InferenceRequestBody{
 			Payload: fwkrh.RawPayload(body),
 		},
-		Skip: false,
+		SkipResponseProcessing: false,
 	}, nil
 }
 

pkg/epp/framework/plugins/requesthandling/parsers/passthrough/passthrough_test.go

@@ -59,8 +59,8 @@ func TestPassthroughParser_ParseRequest(t *testing.T) {
 			if err != nil {
 				t.Fatalf("unexpected error: %v", err)
 			}
-			if got.Skip != false {
-				t.Errorf("got.Skip = %v, want false", got.Skip)
+			if got.SkipResponseProcessing != false {
+				t.Errorf("got.SkipResponseProcessing = %v, want false", got.SkipResponseProcessing)
 			}
 			if diff := cmp.Diff(tt.wantBody, got.Body); diff != "" {
 				t.Errorf("Unexpected body (-want +got):\n%s", diff)

pkg/epp/framework/plugins/requesthandling/parsers/vertexai/vertexai.go

@@ -114,7 +114,12 @@ func (p *VertexAIParser) ParseRequest(ctx context.Context, body []byte, headers
 		return p.parseVertexRequest(ctx, body, headers, &aiplatformpb.RawPredictRequest{}, "RawPredictRequest", openAIResponsesPath)
 
 	default:
-		return &fwkrh.ParseResult{Skip: true}, nil
+		return &fwkrh.ParseResult{
+			Body: &fwkrh.InferenceRequestBody{
+				Payload: fwkrh.RawPayload(body),
+			},
+			SkipResponseProcessing: true,
+		}, nil
 	}
 }
 
@@ -172,5 +177,5 @@ func (p *VertexAIParser) parseVertexRequest(ctx context.Context, body []byte, he
 
 	inferenceRequestBody := parseResult.Body
 	inferenceRequestBody.Payload = fwkrh.PayloadProto{Message: req}
-	return &fwkrh.ParseResult{Body: inferenceRequestBody, Skip: parseResult.Skip}, nil
+	return &fwkrh.ParseResult{Body: inferenceRequestBody, SkipResponseProcessing: parseResult.SkipResponseProcessing}, nil
 }

pkg/epp/framework/plugins/requesthandling/parsers/vertexai/vertexai_test.go

@@ -98,7 +98,7 @@ func TestParseRequest(t *testing.T) {
 					Stream:  true,
 					Payload: fwkrh.PayloadProto{Message: reqMsg},
 				},
-				Skip: false,
+				SkipResponseProcessing: false,
 			},
 		},
 		{
@@ -115,7 +115,7 @@ func TestParseRequest(t *testing.T) {
 					},
 					Payload: fwkrh.PayloadProto{Message: streamRawPredictReqMsg},
 				},
-				Skip: false,
+				SkipResponseProcessing: false,
 			},
 		},
 		{
@@ -132,15 +132,18 @@ func TestParseRequest(t *testing.T) {
 					},
 					Payload: fwkrh.PayloadProto{Message: rawPredictReqMsg},
 				},
-				Skip: false,
+				SkipResponseProcessing: false,
 			},
 		},
 		{
 			name:    "Unsupported Path",
 			body:    []byte{},
 			headers: map[string]string{":path": "/unsupported/path", "content-type": "application/grpc"},
 			wantResult: &fwkrh.ParseResult{
-				Skip: true,
+				Body: &fwkrh.InferenceRequestBody{
+					Payload: fwkrh.RawPayload([]byte{}),
+				},
+				SkipResponseProcessing: true,
 			},
 		},
 		{

pkg/epp/framework/plugins/requesthandling/parsers/vllmgrpc/vllmgrpc.go

@@ -98,7 +98,7 @@ func (p *VllmGRPCParser) ParseRequest(ctx context.Context, body []byte, headers
 			return nil, err
 		}
 		logger.V(logutil.TRACE).Info("parsed EmbedRequest")
-		return &fwkrh.ParseResult{Body: extractedBody, Skip: false}, nil
+		return &fwkrh.ParseResult{Body: extractedBody, SkipResponseProcessing: false}, nil
 
 	case vllmGeneratePath:
 		var req pb.GenerateRequest
@@ -110,11 +110,16 @@ func (p *VllmGRPCParser) ParseRequest(ctx context.Context, body []byte, headers
 			return nil, err
 		}
 		logger.V(logutil.TRACE).Info("parsed GenerateRequest")
-		return &fwkrh.ParseResult{Body: extractedBody, Skip: false}, nil
+		return &fwkrh.ParseResult{Body: extractedBody, SkipResponseProcessing: false}, nil
 
 	default:
 		logger.V(logutil.TRACE).Info("unsupported gRPC path, skipping", "path", headers[parsers.MethodPathKey])
-		return &fwkrh.ParseResult{Skip: true}, nil
+		return &fwkrh.ParseResult{
+			Body: &fwkrh.InferenceRequestBody{
+				Payload: fwkrh.RawPayload(body),
+			},
+			SkipResponseProcessing: true,
+		}, nil
 	}
 }