feat(processor): tenant-scoped model_gateways lookups (route_key_by_tenant)

charts/batch-gateway/templates/processor-configmap.yaml

@@ -130,6 +130,9 @@ data:
     {{- if .Values.processor.config.sendFairnessHeader }}
     send_fairness_header: true
     {{- end }}
+    {{- if .Values.processor.config.routeKeyByTenant }}
+    route_key_by_tenant: true
+    {{- end }}
     default_output_expiration_seconds: {{ .Values.processor.config.defaultOutputExpirationSeconds }}
     progress_ttl_seconds: {{ .Values.processor.config.progressTTLSeconds }}
 

charts/batch-gateway/tests/processor-configmap_test.yaml

@@ -388,6 +388,20 @@ tests:
           path: data["config.yaml"]
           pattern: 'send_fairness_header: true'
 
+  - it: should not render route_key_by_tenant when false (default)
+    asserts:
+      - notMatchRegex:
+          path: data["config.yaml"]
+          pattern: 'route_key_by_tenant:'
+
+  - it: should render route_key_by_tenant when enabled
+    set:
+      processor.config.routeKeyByTenant: true
+    asserts:
+      - matchRegex:
+          path: data["config.yaml"]
+          pattern: 'route_key_by_tenant: true'
+
   - it: should not render ConfigMap when processor is disabled
     set:
       processor.enabled: false

charts/batch-gateway/values.yaml

@@ -393,6 +393,10 @@ processor:
     #     inferenceObjective: "batch-sheddable-b"   # references gie-b pool
     #     requestTimeout: "2m"
     #     maxRetries: 1
+    # Scope modelGateways lookups by the persisted job tenant ID. When enabled,
+    # configure keys as "<tenantID>/<modelID>". The forwarded request body keeps
+    # its original model ID so the runtime still receives its served model name.
+    routeKeyByTenant: false
 
     # Async dispatch mode (alternative to sync).
     # Set dispatchMode: "async" and configure asyncDispatch.models instead
@@ -413,7 +417,6 @@ processor:
     #       inferenceObjective: "batch-sheddable-b"
     #       requestQueueName: "llm-d-async:requests:pool-b"
     #       resultQueueName: "llm-d-async:results:pool-b"
-
     defaultOutputExpirationSeconds: 7776000  # 90 days
     progressTTLSeconds: 86400  # 24 hours
     # Whether to send x-gateway-inference-fairness-id on inference requests.

internal/processor/config/config.go

@@ -174,6 +174,14 @@ type ProcessorConfig struct {
 	// Ignored when GlobalInferenceGateway is set.
 	ModelGateways map[string]ModelGatewayConfig `yaml:"model_gateways"`
 
+	// RouteKeyByTenant scopes model_gateways lookups by the job's tenant ID:
+	// when enabled, requests resolve gateways under "<tenantID>/<modelID>",
+	// so identically-named models of different tenants route to their own
+	// backends on a shared apiserver. The forwarded request body is left
+	// untouched (the runtime still sees the bare model name). Default false
+	// keeps bare-model lookups.
+	RouteKeyByTenant bool `yaml:"route_key_by_tenant"`
+
 	// DefaultOutputExpirationSeconds is the default TTL for batch output/error files in seconds.
 	// Used as fallback when the user does not provide output_expires_after in POST /v1/batches.
 	// 0 means no expiration (keep until explicitly deleted).

internal/processor/config/config_test.go

@@ -89,6 +89,9 @@ func TestNewConfig_Defaults(t *testing.T) {
 	if c.SendFairnessHeader {
 		t.Fatalf("SendFairnessHeader = true, want false by default")
 	}
+	if c.RouteKeyByTenant {
+		t.Fatalf("RouteKeyByTenant = true, want false by default")
+	}
 
 	want90Days := int64(90 * 24 * 60 * 60)
 	if c.DefaultOutputExpirationSeconds != want90Days {
@@ -562,6 +565,7 @@ model_gateways:
 default_output_expiration_seconds: 86400
 progress_ttl_seconds: 3600
 send_fairness_header: true
+route_key_by_tenant: true
 `)
 
 	if err := os.WriteFile(path, yamlData, 0o600); err != nil {
@@ -639,6 +643,9 @@ send_fairness_header: true
 	if !c.SendFairnessHeader {
 		t.Fatalf("SendFairnessHeader = false, want true")
 	}
+	if !c.RouteKeyByTenant {
+		t.Fatalf("RouteKeyByTenant = false, want true")
+	}
 }
 
 func TestProcessorConfig_Validate_AsyncDispatch(t *testing.T) {

internal/processor/worker/preprocessor.go

@@ -117,7 +117,7 @@ func (p *Processor) preProcessJob(ctx context.Context, jobInfo *batch_types.JobI
 	// Production paths hit Processor.validate() in prepare() before work runs.
 	// The guard also avoids panicking if a future caller wires a nil resolver.
 	isPerModelGateway := p.inference != nil && !p.inference.IsGlobal()
-	registeredModels := make(map[string]bool) // modelID -> registered (per-model only)
+	registeredModels := make(map[string]bool) // route key -> registered (per-model only)
 
 	// Always truncate error.jsonl at the start of ingestion so that re-enqueued
 	// jobs don't carry stale error entries from a previous attempt.
@@ -177,10 +177,14 @@ func (p *Processor) preProcessJob(ctx context.Context, jobInfo *batch_types.JobI
 		seenCustomIDs[requestMeta.CustomID] = struct{}{}
 
 		if isPerModelGateway {
-			registered, checked := registeredModels[requestMeta.ModelID]
+			// Look up the gateway by the route key (tenant-scoped when
+			// route_key_by_tenant is enabled). The raw model ID stays in the
+			// error message and plan grouping below.
+			lookupID := routeKey(p.cfg.RouteKeyByTenant, jobInfo.TenantID, requestMeta.ModelID)
+			registered, checked := registeredModels[lookupID]
 			if !checked {
-				registered = p.inference.ClientFor(requestMeta.ModelID) != nil
-				registeredModels[requestMeta.ModelID] = registered
+				registered = p.inference.ClientFor(lookupID) != nil
+				registeredModels[lookupID] = registered
 			}
 			if !registered {
 				// No plan entry exists yet, so generate a UUID for the batch request ID.

internal/processor/worker/route_key.go

@@ -0,0 +1,14 @@
+package worker
+
+// routeKey returns the model_gateways lookup key for a request model. When
+// byTenant is enabled and tenantID is non-empty, gateway entries are scoped
+// per tenant as "<tenantID>/<modelID>", letting identically-named models of
+// different tenants route to their own backends (e.g. per-InferSet gateways
+// sharing one batch-apiserver). Otherwise the bare model ID is used,
+// preserving the default behavior.
+func routeKey(byTenant bool, tenantID, modelID string) string {
+	if !byTenant || tenantID == "" {
+		return modelID
+	}
+	return tenantID + "/" + modelID
+}

internal/processor/worker/route_key_test.go

@@ -0,0 +1,43 @@
+package worker
+
+import "testing"
+
+func TestRouteKey(t *testing.T) {
+	tests := []struct {
+		name     string
+		byTenant bool
+		tenantID string
+		modelID  string
+		want     string
+	}{
+		{
+			name:     "disabled keeps bare model ID",
+			byTenant: false,
+			tenantID: "m-20260720103021-nvfbq",
+			modelID:  "test-model-v1",
+			want:     "test-model-v1",
+		},
+		{
+			name:     "enabled with empty tenant keeps bare model ID",
+			byTenant: true,
+			tenantID: "",
+			modelID:  "test-model-v1",
+			want:     "test-model-v1",
+		},
+		{
+			name:     "enabled scopes the lookup key by tenant",
+			byTenant: true,
+			tenantID: "m-20260720103021-nvfbq",
+			modelID:  "test-model-v1",
+			want:     "m-20260720103021-nvfbq/test-model-v1",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := routeKey(tt.byTenant, tt.tenantID, tt.modelID); got != tt.want {
+				t.Fatalf("routeKey(%v, %q, %q) = %q, want %q", tt.byTenant, tt.tenantID, tt.modelID, got, tt.want)
+			}
+		})
+	}
+}

internal/processor/worker/source_planfile.go

@@ -108,13 +108,19 @@ func (s *PlanFileSource) readEntry(entry planEntry, modelID string) (*pipeline.R
 		}, nil
 	}
 
+	// When route_key_by_tenant is enabled, scope the gateway lookup key by
+	// tenant so identically-named models of different tenants route to their
+	// own backends (model_gateways entries keyed "<tenantID>/<modelID>").
+	// The request body itself is forwarded verbatim to the inference backend.
+	lookupID := routeKey(s.cfg.RouteKeyByTenant, s.tenantID, modelID)
+
 	headers := maps.Clone(s.passThroughHeaders)
-	headers = s.mergeHeaders(headers, modelID)
+	headers = s.mergeHeaders(headers, lookupID)
 
 	return &pipeline.RequestItem{
 		RequestID: fmt.Sprintf("batch_req_%s", uuid.NewString()),
 		CustomID:  req.CustomID,
-		ModelID:   modelID,
+		ModelID:   lookupID,
 		Endpoint:  req.URL,
 		Body:      req.Body,
 		Headers:   headers,

internal/processor/worker/source_planfile_test.go

@@ -342,6 +342,117 @@ func TestPlanFileSource_Produce(t *testing.T) {
 	}
 }
 
+func TestPlanFileSource_Produce_TenantScopedLookup(t *testing.T) {
+	dir := t.TempDir()
+
+	requests := []batch_types.Request{
+		{CustomID: "c-1", Method: "POST", URL: "/v1/chat/completions", Body: map[string]any{"model": "m1", "prompt": "hello"}},
+	}
+
+	inputPath := filepath.Join(dir, "input.jsonl")
+	var entries []planEntry
+	f, err := os.Create(inputPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, req := range requests {
+		data, _ := json.Marshal(req)
+		data = append(data, '\n')
+		offset, _ := f.Seek(0, 1)
+		entries = append(entries, planEntry{
+			Offset: offset,
+			Length: uint32(len(data)),
+		})
+		if _, err := f.Write(data); err != nil {
+			t.Fatal(err)
+		}
+	}
+	f.Close()
+
+	plansDir := filepath.Join(dir, "plans")
+	writePlanFile(t, plansDir, "m1", entries)
+
+	inputFile, err := os.Open(inputPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer inputFile.Close()
+
+	client := &mockInferenceClient{}
+	resolver := inference.NewSingleClientResolver(client)
+	defer func() { _ = resolver.Close() }()
+
+	cfg := config.NewConfig()
+	cfg.RouteKeyByTenant = true
+	cfg.ModelGateways = map[string]config.ModelGatewayConfig{
+		"inferset-a/m1": {
+			URL:                "http://gw-a:8000",
+			InferenceObjective: "inferset-a-batch",
+		},
+	}
+
+	source := NewPlanFileSource(PlanFileSourceConfig{
+		InputFile: inputFile,
+		PlansDir:  plansDir,
+		ModelMap:  &modelMapFile{SafeToModel: map[string]string{"m1": "m1"}, LineCount: 1},
+		Resolver:  resolver,
+		Cfg:       cfg,
+		TenantID:  "inferset-a",
+		Logger:    logr.Discard(),
+	})
+
+	out := make(chan pipeline.RequestItem, 10)
+	if err := source.Produce(context.Background(), out); err != nil {
+		t.Fatalf("Produce error: %v", err)
+	}
+
+	var items []pipeline.RequestItem
+	for item := range out {
+		items = append(items, item)
+	}
+
+	if len(items) != 1 {
+		t.Fatalf("produced %d items, want 1", len(items))
+	}
+	if items[0].ModelID != "inferset-a/m1" {
+		t.Errorf("item 0 ModelID = %q, want tenant-scoped %q", items[0].ModelID, "inferset-a/m1")
+	}
+	// The body is forwarded verbatim: the runtime still sees the raw model name.
+	if items[0].Body["model"] != "m1" {
+		t.Errorf("item 0 body model = %v, want raw %q", items[0].Body["model"], "m1")
+	}
+	if items[0].Headers[inferenceObjectiveHeader] != "inferset-a-batch" {
+		t.Errorf("objective header = %q, want %q", items[0].Headers[inferenceObjectiveHeader], "inferset-a-batch")
+	}
+
+	// With the default config (route_key_by_tenant off), the same tenant must
+	// not affect the lookup key — guards the backward-compatible behavior.
+	cfgOff := config.NewConfig()
+	inputFile2, err := os.Open(inputPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer inputFile2.Close()
+	sourceOff := NewPlanFileSource(PlanFileSourceConfig{
+		InputFile: inputFile2,
+		PlansDir:  plansDir,
+		ModelMap:  &modelMapFile{SafeToModel: map[string]string{"m1": "m1"}, LineCount: 1},
+		Resolver:  resolver,
+		Cfg:       cfgOff,
+		TenantID:  "inferset-a",
+		Logger:    logr.Discard(),
+	})
+	outOff := make(chan pipeline.RequestItem, 10)
+	if err := sourceOff.Produce(context.Background(), outOff); err != nil {
+		t.Fatalf("Produce (route_key_by_tenant off) error: %v", err)
+	}
+	for item := range outOff {
+		if item.ModelID != "m1" {
+			t.Errorf("route_key_by_tenant off: ModelID = %q, want bare %q", item.ModelID, "m1")
+		}
+	}
+}
+
 func TestPlanFileSource_Produce_MultipleModels(t *testing.T) {
 	dir := t.TempDir()