package service import ( "context" "errors" "testing" "time" edgeevents "iop/apps/edge/internal/events" edgenode "iop/apps/edge/internal/node" "iop/packages/go/config" ) // TestProtocolProfileCandidateCopy verifies that the concrete profile is // deep-copied onto each provider candidate and that profileFacts extracts // the correct id and driver. func TestProtocolProfileCandidateCopy(t *testing.T) { profile := mustResolveProfileForService(t, "openai") rec := &edgenode.NodeRecord{ ID: "node-profile-copy", Adapters: config.AdaptersConf{ VllmInstances: []config.VllmInstanceConf{ {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, }, }, Providers: []config.NodeProviderConf{{ ID: "prov-copy", Type: "openai_api", Adapter: "vllm-gpu", Models: []string{"gpt-4"}, Health: "available", Capacity: 4, Profile: "openai", RuntimeProfile: &profile, }}, } store := edgenode.NewNodeStore() store.Add(rec) reg := edgenode.NewRegistry() reg.Register(&edgenode.NodeEntry{NodeID: "node-profile-copy"}) svc := New(reg, edgeevents.NewBus()) svc.SetRuntimeConfig(store, []config.ModelCatalogEntry{{ ID: "gpt-4", Providers: map[string]string{"prov-copy": "gpt-4"}, }}, groupPolicy{}) cands, _, err := svc.resolveProviderPoolCandidates(SubmitRunRequest{ ModelGroupKey: "gpt-4", ProviderPool: true, }, store, []config.ModelCatalogEntry{{ ID: "gpt-4", Providers: map[string]string{"prov-copy": "gpt-4"}, }}) if err != nil { t.Fatalf("resolveProviderPoolCandidates: %v", err) } if len(cands) != 1 { t.Fatalf("expected 1 candidate, got %d", len(cands)) } if cands[0].adapter != "prov-copy" { t.Fatalf("profile-backed dispatch adapter = %q, want provider id", cands[0].adapter) } c := cands[0] if c.profile == nil { t.Fatal("expected non-nil profile on candidate") } if c.profile.ID != "openai" { t.Errorf("profile.ID = %q, want %q", c.profile.ID, "openai") } if string(c.profile.Driver) != string(config.ProtocolDriverOpenAIChat) { t.Errorf("profile.Driver = %q, want %q", c.profile.Driver, config.ProtocolDriverOpenAIChat) } if c.profile.BaseURL != "https://api.openai.com/v1" { t.Errorf("profile.BaseURL = %q, want %q", c.profile.BaseURL, "https://api.openai.com/v1") } profile.Operations[string(config.OperationChatCompletions)] = "/mutated" profile.Capabilities[0] = "mutated" if c.profile.Operations[string(config.OperationChatCompletions)] == "/mutated" || c.profile.Capabilities[0] == "mutated" { t.Fatal("source profile mutation crossed candidate snapshot boundary") } // Verify profileFacts extracts the correct values. id, driver := profileFacts(c.profile) if id != "openai" { t.Errorf("profileFacts id = %q, want %q", id, "openai") } if driver != string(config.ProtocolDriverOpenAIChat) { t.Errorf("profileFacts driver = %q, want %q", driver, config.ProtocolDriverOpenAIChat) } // Verify profileFacts returns empty for nil profile (legacy candidate). id2, driver2 := profileFacts(nil) if id2 != "" || driver2 != "" { t.Errorf("profileFacts(nil) = (%q, %q), want empty", id2, driver2) } } func TestProtocolProfileAdapterDispatchUsesProviderID(t *testing.T) { profile := mustResolveProfileForService(t, "openai") provider := config.NodeProviderConf{ ID: "provider-dispatch", Type: "openai_compat", Adapter: "backing", Models: []string{"served"}, Health: "available", Capacity: 1, RuntimeProfile: &profile, } makeService := func(p config.NodeProviderConf) (*Service, *edgenode.NodeStore, []config.ModelCatalogEntry) { store := edgenode.NewNodeStore() store.Add(&edgenode.NodeRecord{ID: "node-dispatch", Adapters: config.AdaptersConf{ OpenAICompatInstances: []config.OpenAICompatInstanceConf{{Name: "backing", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}}, }, Providers: []config.NodeProviderConf{p}}) catalog := []config.ModelCatalogEntry{{ID: "model", Providers: map[string]string{p.ID: "served"}}} registry := edgenode.NewRegistry() registry.Register(&edgenode.NodeEntry{NodeID: "node-dispatch"}) service := New(registry, nil) return service, store, catalog } service, store, catalog := makeService(provider) candidates, _, err := service.resolveProviderPoolCandidates(SubmitRunRequest{ModelGroupKey: "model", ProviderPool: true}, store, catalog) if err != nil || len(candidates) != 1 || candidates[0].adapter != "provider-dispatch" { t.Fatalf("profile dispatch candidates=%+v err=%v", candidates, err) } provider.RuntimeProfile = nil service, store, catalog = makeService(provider) candidates, _, err = service.resolveProviderPoolCandidates(SubmitRunRequest{ModelGroupKey: "model", ProviderPool: true}, store, catalog) if err != nil || len(candidates) != 1 || candidates[0].adapter != "backing" { t.Fatalf("legacy dispatch candidates=%+v err=%v", candidates, err) } } // TestProtocolProfileCandidateCallback verifies that the // ProviderPoolCandidatePredicate receives the concrete profile facts and can // reject candidates based on them. func TestProtocolProfileCandidateCallback(t *testing.T) { profile := mustResolveProfileForService(t, "anthropic") rec := &edgenode.NodeRecord{ ID: "node-callback", Adapters: config.AdaptersConf{ VllmInstances: []config.VllmInstanceConf{ {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, }, }, Providers: []config.NodeProviderConf{{ ID: "prov-cb", Type: "openai_api", Adapter: "vllm-gpu", Models: []string{"claude-3"}, Health: "available", Capacity: 4, Profile: "anthropic", RuntimeProfile: &profile, }}, } store := edgenode.NewNodeStore() store.Add(rec) catalog := []config.ModelCatalogEntry{{ ID: "claude-3", Providers: map[string]string{"prov-cb": "claude-3"}, }} reg := edgenode.NewRegistry() reg.Register(&edgenode.NodeEntry{NodeID: "node-callback"}) svc := New(reg, nil) svc.SetRuntimeConfig(store, catalog, groupPolicy{}) cands, _, err := svc.resolveProviderPoolCandidates(SubmitRunRequest{ ModelGroupKey: "claude-3", ProviderPool: true, }, store, catalog) if err != nil { t.Fatalf("resolveProviderPoolCandidates: %v", err) } if len(cands) != 1 { t.Fatalf("expected 1 candidate, got %d", len(cands)) } // Accept predicate that only accepts anthropic_messages driver. accept := func(c ProviderPoolCandidate) bool { return c.ProfileDriver == string(config.ProtocolDriverAnthropicMessages) } accepted, rejected := filterProviderPoolCandidates(cands, accept) if rejected { t.Fatal("expected candidate to be accepted") } if len(accepted) != 1 { t.Fatalf("expected 1 accepted candidate, got %d", len(accepted)) } if accepted[0].profile.ID != "anthropic" { t.Errorf("accepted profile ID = %q, want %q", accepted[0].profile.ID, "anthropic") } // Reject predicate that only accepts openai_chat driver. reject := func(c ProviderPoolCandidate) bool { return c.ProfileDriver == string(config.ProtocolDriverOpenAIChat) } accepted2, rejected2 := filterProviderPoolCandidates(cands, reject) if !rejected2 { t.Fatal("expected candidate to be rejected") } if len(accepted2) != 0 { t.Fatalf("expected 0 accepted candidates, got %d", len(accepted2)) } } // TestProtocolProfileReResolution verifies that the concrete profile is // refreshed from NodeProviderConf when the store is updated via SetRuntimeConfig. func TestProtocolProfileReResolution(t *testing.T) { openaiProfile := mustResolveProfileForService(t, "openai") anthropicProfile := mustResolveProfileForService(t, "anthropic") newStore := func(profile *config.ConcreteProtocolProfile, profileID string) *edgenode.NodeStore { store := edgenode.NewNodeStore() store.Add(&edgenode.NodeRecord{ ID: "node-rez", Adapters: config.AdaptersConf{ VllmInstances: []config.VllmInstanceConf{ {Name: "vllm-gpu", Enabled: true, Endpoint: "http://127.0.0.1:8000/v1"}, }, }, Providers: []config.NodeProviderConf{{ ID: "prov-rez", Type: "openai_api", Adapter: "vllm-gpu", Models: []string{"model-a"}, Health: "available", Capacity: 4, Profile: profileID, RuntimeProfile: profile, }}, }) return store } catalog := []config.ModelCatalogEntry{{ ID: "model-a", Providers: map[string]string{"prov-rez": "model-a"}, }} reg := edgenode.NewRegistry() reg.Register(&edgenode.NodeEntry{NodeID: "node-rez"}) svc := New(reg, edgeevents.NewBus()) svc.SetRuntimeConfig(newStore(&openaiProfile, "openai"), catalog, groupPolicy{}) store, _, _ := svc.runtimeConfigSnapshot() cands, _, err := svc.resolveProviderPoolCandidates(SubmitRunRequest{ ModelGroupKey: "model-a", ProviderPool: true, }, store, catalog) if err != nil { t.Fatalf("resolve before refresh: %v", err) } if len(cands) != 1 || cands[0].profile.ID != "openai" { t.Fatalf("before refresh: expected profile ID openai, got %+v", cands) } queuedSnapshot := cands[0] // Refresh: swap to anthropic profile. svc.SetRuntimeConfig(newStore(&anthropicProfile, "anthropic"), catalog, groupPolicy{}) svc.queue.mu.Lock() liveCandidate, eligible := svc.queue.liveCandidateLocked(&queuedSnapshot) svc.queue.mu.Unlock() if !eligible || liveCandidate.profile == nil || liveCandidate.profile.ID != "anthropic" { t.Fatalf("queued refresh did not clone the live profile: eligible=%v candidate=%+v", eligible, liveCandidate) } if queuedSnapshot.profile == nil || queuedSnapshot.profile.ID != "openai" { t.Fatalf("queued snapshot changed across refresh: %+v", queuedSnapshot.profile) } liveCandidate.profile.Operations[string(config.OperationMessages)] = "/mutated-live" refreshedStore, _, _ := svc.runtimeConfigSnapshot() refreshedRecord, _ := refreshedStore.FindByID("node-rez") if got := refreshedRecord.Providers[0].RuntimeProfile.Operations[string(config.OperationMessages)]; got == "/mutated-live" { t.Fatal("live candidate mutation crossed refreshed store boundary") } store, _, _ = svc.runtimeConfigSnapshot() cands, _, err = svc.resolveProviderPoolCandidates(SubmitRunRequest{ ModelGroupKey: "model-a", ProviderPool: true, }, store, catalog) if err != nil { t.Fatalf("resolve after refresh: %v", err) } if len(cands) != 1 || cands[0].profile.ID != "anthropic" { t.Fatalf("after refresh: expected profile ID anthropic, got %+v", cands) } if string(cands[0].profile.Driver) != string(config.ProtocolDriverAnthropicMessages) { t.Errorf("after refresh: expected driver anthropic_messages, got %q", cands[0].profile.Driver) } } // TestProtocolProfileOperationWirePropagation verifies that the operation field // is propagated through the wire request builder. func TestProtocolProfileOperationWirePropagation(t *testing.T) { tunnelReq := SubmitProviderTunnelRequest{ RunID: "run-op-wire", Adapter: "openai_compat", Target: "gpt-4", Method: "POST", Operation: string(config.OperationChatCompletions), Path: "/v1/chat/completions", Body: []byte(`{"model":"gpt-4","messages":[]}`), Headers: map[string]string{"Authorization": "Bearer key"}, } wire, runID, err := buildProviderTunnelRequest(tunnelReq, "openai_compat", "gpt-4") if err != nil { t.Fatalf("buildProviderTunnelRequest: %v", err) } if wire.GetOperation() != string(config.OperationChatCompletions) { t.Errorf("wire Operation = %q, want %q", wire.GetOperation(), config.OperationChatCompletions) } if wire.GetPath() != "/v1/chat/completions" { t.Errorf("wire Path = %q, want %q", wire.GetPath(), "/v1/chat/completions") } if runID != "run-op-wire" { t.Errorf("runID = %q, want %q", runID, "run-op-wire") } } // TestProtocolProfileLegacyPathFallback verifies that when no profile is // present (legacy candidate), the Path field is still carried through. func TestProtocolProfileLegacyPathFallback(t *testing.T) { tunnelReq := SubmitProviderTunnelRequest{ RunID: "run-legacy-wire", Adapter: "openai_compat", Target: "gpt-4", Method: "POST", Path: "/v1/chat/completions", Body: []byte(`{"model":"gpt-4"}`), } wire, _, err := buildProviderTunnelRequest(tunnelReq, "openai_compat", "gpt-4") if err != nil { t.Fatalf("buildProviderTunnelRequest: %v", err) } if wire.GetPath() != "/v1/chat/completions" { t.Errorf("wire Path = %q, want %q", wire.GetPath(), "/v1/chat/completions") } if wire.GetOperation() != "" { t.Errorf("wire Operation = %q, want empty for legacy", wire.GetOperation()) } } // TestProtocolProfileUnsupportedProfileAdmission verifies that candidates // with nil profiles (legacy) are still eligible for dispatch and that the // filter does not panic when profile is nil. func TestProtocolProfileUnsupportedProfileAdmission(t *testing.T) { entry := &edgenode.NodeEntry{NodeID: "node-nil-profile"} cands := []candidateNode{{ entry: entry, capacity: 4, providerID: "prov-nil", servedTarget: "model-x", // profile is nil (legacy candidate) }} // An accept predicate that checks profile facts should get empty strings // for nil profiles and still be able to accept them. accept := func(c ProviderPoolCandidate) bool { // Legacy candidates have empty profile facts but are still valid. return c.ProfileID == "" && c.ProviderID == "prov-nil" } accepted, rejected := filterProviderPoolCandidates(cands, accept) if rejected { t.Fatal("expected legacy candidate with nil profile to be accepted") } if len(accepted) != 1 { t.Fatalf("expected 1 accepted candidate, got %d", len(accepted)) } } func TestProtocolProfileOperationAdmissionFiltersUnsupportedCandidate(t *testing.T) { env := newProviderTunnelTestEnv(t) store, catalog, policy := env.svc.runtimeConfigSnapshot() record, _ := store.FindByID("node-pool") gemini := mustResolveProfileForService(t, "gemini") openai := mustResolveProfileForService(t, "openai") record.Providers[0].RuntimeProfile = &gemini record.Providers[0].Profile = "gemini" record.Providers[0].Priority = 0 record.Providers = append(record.Providers, config.NodeProviderConf{ ID: "prov-responses", Type: "openai_api", Adapter: "vllm-gpu", Models: []string{"served-responses"}, Health: "available", Capacity: 1, Priority: 10, Profile: "openai", RuntimeProfile: &openai, }) catalog[0].Providers["prov-responses"] = "served-responses" env.svc.SetRuntimeConfig(store, catalog, policy) result, err := env.svc.SubmitProviderPool(context.Background(), ProviderPoolDispatchRequest{ Run: SubmitRunRequest{ModelGroupKey: "qwen3.6:35b", ProviderPool: true}, Tunnel: SubmitProviderTunnelRequest{Method: "POST", Operation: string(config.OperationResponses), Path: "/responses"}, }) if err != nil { t.Fatalf("SubmitProviderPool: %v", err) } defer result.Tunnel.Close() if result.DispatchInfo.ProviderID != "prov-responses" { t.Fatalf("selected provider = %q, want operation-compatible provider", result.DispatchInfo.ProviderID) } } func TestProtocolProfileOperationAdmissionRejectsUnsupportedCandidate(t *testing.T) { env := newProviderTunnelTestEnv(t) store, catalog, policy := env.svc.runtimeConfigSnapshot() record, _ := store.FindByID("node-pool") anthropic := mustResolveProfileForService(t, "anthropic") record.Providers[0].Profile = "anthropic" record.Providers[0].RuntimeProfile = &anthropic env.svc.SetRuntimeConfig(store, catalog, policy) _, submitErr := env.svc.SubmitProviderPool(context.Background(), ProviderPoolDispatchRequest{ Run: SubmitRunRequest{ModelGroupKey: "qwen3.6:35b", ProviderPool: true}, Tunnel: SubmitProviderTunnelRequest{Method: "POST", Operation: string(config.OperationResponses), Path: "/responses"}, }) if submitErr == nil { t.Fatal("expected SubmitProviderPool error for unsupported operation, got nil") } var opErr *ProviderPoolOperationUnsupportedError if !errors.As(submitErr, &opErr) { t.Fatalf("expected ProviderPoolOperationUnsupportedError, got %T (%v)", submitErr, submitErr) } if opErr.Operation != string(config.OperationResponses) { t.Fatalf("opErr.Operation = %q, want %q", opErr.Operation, config.OperationResponses) } if !errors.Is(submitErr, ErrProviderPoolCandidateRejected) { t.Fatalf("errors.Is(submitErr, ErrProviderPoolCandidateRejected) = false, want true") } if env.capturedRequest() != nil { t.Fatal("unsupported operation must not send any Node wire request") } if inflightRunCount(env.svc.queue) != 0 { t.Fatal("unsupported operation must leave zero stranded queue admission") } } func TestProtocolProfileOperationAdmissionNilProfileCompatibility(t *testing.T) { env := newProviderTunnelTestEnv(t) result, err := env.svc.SubmitProviderPool(context.Background(), ProviderPoolDispatchRequest{ Run: SubmitRunRequest{ModelGroupKey: "qwen3.6:35b", ProviderPool: true}, Tunnel: SubmitProviderTunnelRequest{Method: "POST", Operation: string(config.OperationMessages), Path: "/messages"}, }) if err != nil { t.Fatalf("legacy nil-profile candidate was rejected: %v", err) } defer result.Tunnel.Close() if result.DispatchInfo.ProviderID != "prov-vllm-01" { t.Fatalf("selected provider = %q", result.DispatchInfo.ProviderID) } } func TestProtocolProfileOperationAdmissionReResolution(t *testing.T) { env := newProviderTunnelTestEnv(t) store, catalog, policy := env.svc.runtimeConfigSnapshot() record, _ := store.FindByID("node-pool") openai := mustResolveProfileForService(t, "openai") record.Providers[0].Profile = "openai" record.Providers[0].RuntimeProfile = &openai record.Providers[0].Capacity = 1 env.svc.SetRuntimeConfig(store, catalog, policy) candidates, _, err := env.svc.resolveQueueCandidates(SubmitRunRequest{ModelGroupKey: "qwen3.6:35b", ProviderPool: true}) if err != nil { t.Fatalf("resolve holder: %v", err) } holder, err := env.svc.queue.admit(t.Context(), "qwen3.6:35b", "", "", candidates, groupPolicy{}, nil, false, true) if err != nil { t.Fatalf("reserve holder: %v", err) } resultCh := make(chan error, 1) ctx, cancel := context.WithTimeout(t.Context(), time.Second) defer cancel() go func() { _, submitErr := env.svc.SubmitProviderPool(ctx, ProviderPoolDispatchRequest{ Run: SubmitRunRequest{ModelGroupKey: "qwen3.6:35b", ProviderPool: true}, Tunnel: SubmitProviderTunnelRequest{Method: "POST", Operation: string(config.OperationResponses), Path: "/responses"}, }) resultCh <- submitErr }() requireProviderPoolPending(t, env.svc.queue, 1) refreshed := edgenode.NewNodeStore() copyRecord := *record copyRecord.Providers = append([]config.NodeProviderConf(nil), record.Providers...) anthropic := mustResolveProfileForService(t, "anthropic") copyRecord.Providers[0].Profile = "anthropic" copyRecord.Providers[0].RuntimeProfile = &anthropic refreshed.Add(©Record) env.svc.SetRuntimeConfig(refreshed, catalog, policy) env.svc.queue.releaseLease(holder.leaseID, "test-release") submitErr := <-resultCh if submitErr == nil { t.Fatal("expected queued re-resolution error, got nil") } var opErr *ProviderPoolOperationUnsupportedError if !errors.As(submitErr, &opErr) { t.Fatalf("queued re-resolution error = %T (%v), want ProviderPoolOperationUnsupportedError", submitErr, submitErr) } if opErr.Operation != string(config.OperationResponses) { t.Fatalf("queued re-resolution opErr.Operation = %q, want %q", opErr.Operation, config.OperationResponses) } if !errors.Is(submitErr, ErrProviderPoolCandidateRejected) { t.Fatalf("queued re-resolution error = %v, want candidate rejection sentinel compatibility", submitErr) } if env.capturedRequest() != nil { t.Fatal("re-resolution rejection must not send any Node wire request") } } // TestProtocolProfileCandidateFactsExtraction verifies that profileFacts // correctly extracts id and driver from a concrete profile and returns // empty strings for nil. func TestProtocolProfileCandidateFactsExtraction(t *testing.T) { profile := mustResolveProfileForService(t, "gemini") id, driver := profileFacts(&profile) if id != "gemini" { t.Errorf("id = %q, want %q", id, "gemini") } if driver != string(config.ProtocolDriverOpenAIChat) { t.Errorf("driver = %q, want %q", driver, config.ProtocolDriverOpenAIChat) } // Nil profile returns empty. id2, driver2 := profileFacts(nil) if id2 != "" || driver2 != "" { t.Errorf("profileFacts(nil) = (%q, %q), want empty", id2, driver2) } } // TestProtocolProfileCandidateRejectedByPredicate verifies that // ErrProviderPoolCandidateRejected is returned when all candidates are // rejected by the accept predicate, and that the error is correctly typed. func TestProtocolProfileCandidateRejectedByPredicate(t *testing.T) { profile := mustResolveProfileForService(t, "openai") entry := &edgenode.NodeEntry{NodeID: "node-reject"} cands := []candidateNode{{ entry: entry, capacity: 4, providerID: "prov-reject", servedTarget: "model-y", profile: &profile, }} // Reject everything. accept := func(c ProviderPoolCandidate) bool { return false } accepted, rejected := filterProviderPoolCandidates(cands, accept) if !rejected { t.Fatal("expected rejection when all candidates are rejected") } if len(accepted) != 0 { t.Fatalf("expected 0 accepted, got %d", len(accepted)) } } // TestProtocolProfileCandidatePredicateNil verifies that a nil predicate // passes all candidates through. func TestProtocolProfileCandidatePredicateNil(t *testing.T) { profile := mustResolveProfileForService(t, "openai") entry := &edgenode.NodeEntry{NodeID: "node-nil-pred"} cands := []candidateNode{{ entry: entry, capacity: 4, providerID: "prov-nil-pred", servedTarget: "model-z", profile: &profile, }} accepted, rejected := filterProviderPoolCandidates(cands, nil) if rejected { t.Fatal("expected no rejection with nil predicate") } if len(accepted) != 1 { t.Fatalf("expected 1 accepted candidate with nil predicate, got %d", len(accepted)) } } // mustResolveProfileForService resolves a built-in profile for use in service // tests. func mustResolveProfileForService(t *testing.T, id string) config.ConcreteProtocolProfile { t.Helper() profile, err := config.ResolveProtocolProfile(id, "", config.BuiltInProtocolProfiles) if err != nil { t.Fatalf("ResolveProtocolProfile(%q): %v", id, err) } return profile } // TestProtocolProfileSubmitProviderPoolTunnelOperation verifies that the // operation field survives the SubmitProviderTunnel path and reaches the // wire request. func TestProtocolProfileSubmitProviderPoolTunnelOperation(t *testing.T) { newProfileEnv := func(t *testing.T) *providerTunnelTestEnv { t.Helper() env := newProviderTunnelTestEnv(t) store, catalog, policy := env.svc.runtimeConfigSnapshot() record, ok := store.FindByID("node-pool") if !ok { t.Fatal("node-pool not found") } profile := mustResolveProfileForService(t, "openai") profile.Extensions = map[string]any{"nested": map[string]any{"revision": "one"}} record.Providers[0].Type = "ollama" // profile driver must override legacy type classification. record.Providers[0].Profile = "openai" record.Providers[0].RuntimeProfile = &profile env.svc.SetRuntimeConfig(store, catalog, policy) return env } t.Run("protocol_hook_selected_snapshot", func(t *testing.T) { env := newProfileEnv(t) protocolCalls := 0 legacyCalls := 0 result, err := env.svc.SubmitProviderPool(context.Background(), ProviderPoolDispatchRequest{ Run: SubmitRunRequest{ ModelGroupKey: "qwen3.6:35b", ProviderID: "not-the-selected-provider", ProviderPool: true, }, Tunnel: SubmitProviderTunnelRequest{Method: "POST", Path: "/v1/chat/completions"}, PrepareProtocolTunnel: func(req SubmitProviderTunnelRequest, selected ProviderPoolCandidate) (SubmitProviderTunnelRequest, error) { protocolCalls++ if req.ProviderID != "prov-vllm-01" || selected.ProviderID != "prov-vllm-01" { t.Fatalf("selected provider mismatch: req=%q candidate=%q", req.ProviderID, selected.ProviderID) } if selected.ProfileID != "openai" || selected.ProfileDriver != string(config.ProtocolDriverOpenAIChat) || selected.ProtocolProfile == nil { t.Fatalf("selected profile snapshot = %+v", selected) } selected.ProtocolProfile.Operations[string(config.OperationChatCompletions)] = "/callback-mutated" req.Operation = string(config.OperationChatCompletions) return req, nil }, PrepareTunnel: func(req SubmitProviderTunnelRequest) (SubmitProviderTunnelRequest, error) { legacyCalls++ return req, nil }, }) if err != nil { t.Fatalf("SubmitProviderPool: %v", err) } defer result.Tunnel.Close() if protocolCalls != 1 || legacyCalls != 0 { t.Fatalf("protocol calls=%d legacy calls=%d", protocolCalls, legacyCalls) } waitForCondition(t, func() bool { return env.capturedRequest() != nil }, "fake node did not receive protocol tunnel") wire := env.capturedRequest() if wire.GetOperation() != string(config.OperationChatCompletions) { t.Fatalf("wire operation = %q", wire.GetOperation()) } dispatch := result.DispatchInfo if dispatch.ProfileID != "openai" || dispatch.ProfileDriver != string(config.ProtocolDriverOpenAIChat) || dispatch.ExecutionPath != string(providerExecutionPathTunnel) { t.Fatalf("dispatch profile facts = %+v", dispatch) } if got := result.Tunnel.Dispatch().ProfileCapabilities; len(got) == 0 { t.Fatal("tunnel handle lost profile capabilities") } }) t.Run("legacy_hook_compatibility", func(t *testing.T) { env := newProfileEnv(t) legacyCalls := 0 result, err := env.svc.SubmitProviderPool(context.Background(), ProviderPoolDispatchRequest{ Run: SubmitRunRequest{ModelGroupKey: "qwen3.6:35b", ProviderPool: true}, Tunnel: SubmitProviderTunnelRequest{Method: "POST", Path: "/v1/chat/completions"}, PrepareTunnel: func(req SubmitProviderTunnelRequest) (SubmitProviderTunnelRequest, error) { legacyCalls++ req.Operation = string(config.OperationChatCompletions) return req, nil }, }) if err != nil { t.Fatalf("SubmitProviderPool: %v", err) } defer result.Tunnel.Close() waitForCondition(t, func() bool { return env.capturedRequest() != nil }, "fake node did not receive legacy tunnel") if legacyCalls != 1 || env.capturedRequest().GetOperation() != string(config.OperationChatCompletions) { t.Fatalf("legacy hook calls=%d operation=%q", legacyCalls, env.capturedRequest().GetOperation()) } waitForCondition(t, func() bool { return inflightRunCount(env.svc.queue) == 1 }, "tunnel admission was not tracked") result.Tunnel.Close() waitForCondition(t, func() bool { return inflightRunCount(env.svc.queue) == 0 }, "legacy tunnel admission did not release") }) }