iop/apps/edge/internal/service/protocol_profile_test.go

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(&copyRecord)
	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")
	})

}