ModelRouter/internal/scheduler/scheduler_test.go

package scheduler

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"llmsproxy/internal/types"
)

// fakeProvider is an in-memory Provider used to exercise chain scheduling
// deterministically without a Lua runtime.
type fakeProvider struct {
	name      string
	model     string
	pref      atomic.Int64
	available atomic.Bool
	busy      atomic.Bool
	fail      atomic.Bool
	chatHits  atomic.Int64
	// probeable makes an unavailable provider eligible as a half-cooldown
	// probe candidate. probePermit is the single-token permit; probeClaims
	// counts how many times it was handed out and probeDones how many times it
	// was returned, so tests can assert the permit is always released.
	probeable   atomic.Bool
	probePermit atomic.Bool
	probeClaims atomic.Int64
	probeDones  atomic.Int64
}

func fakeProv(name, model string) *fakeProvider {
	f := &fakeProvider{name: name, model: model}
	f.available.Store(true)
	return f
}

func (f *fakeProvider) Name() string { return f.name }

func (f *fakeProvider) ModelFor(reqModel string) string { return f.model }

func (f *fakeProvider) ModelAvailable(model string) bool { return f.available.Load() }

func (f *fakeProvider) ModelSchedulable(model string) (bool, bool) {
	if f.available.Load() {
		return true, false
	}
	if !f.probeable.Load() {
		return false, false
	}
	if f.probePermit.CompareAndSwap(false, true) {
		f.probeClaims.Add(1)
		return true, true
	}
	return false, false
}

func (f *fakeProvider) ProbeDone(model string) {
	f.probeDones.Add(1)
	f.probePermit.Store(false)
}

func (f *fakeProvider) Pref(model string) int64 { return f.pref.Load() }

func (f *fakeProvider) Chat(ctx context.Context, req *types.ChatRequest) (*types.UnifiedResponse, error) {
	f.chatHits.Add(1)
	if f.busy.Load() {
		return nil, types.ErrBusy
	}
	if f.fail.Load() {
		return nil, fmt.Errorf("upstream error")
	}
	return &types.UnifiedResponse{Content: f.name, FinishReason: "stop", TokenUsage: types.TokenUsage{Prompt: 1, Completion: 1, Total: 2}}, nil
}

func (f *fakeProvider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-chan types.UnifiedChunk, error) {
	if f.busy.Load() {
		return nil, types.ErrBusy
	}
	if f.fail.Load() {
		return nil, fmt.Errorf("upstream error")
	}
	ch := make(chan types.UnifiedChunk, 2)
	ch <- types.UnifiedChunk{Content: f.name}
	ch <- types.UnifiedChunk{Done: true}
	close(ch)
	return ch, nil
}

func (f *fakeProvider) Image(ctx context.Context, req *types.ImageGenRequest) (*types.UnifiedResponse, error) {
	return nil, errors.New("no image")
}

// TestDirectSkipsCooledCandidate: direct paths share the AUTO-chain rule that
// cooldown is the only hard skip — a cooling candidate must never be hit, and
// a fully cooling set must fail without touching upstream.
func TestDirectSkipsCooledCandidate(t *testing.T) {
	hot := fakeProv("hot", "m")
	cold := fakeProv("cold", "m")
	cold.available.Store(false)
	s := New(2)
	req := &types.ChatRequest{
		Model:    "m",
		Messages: []types.ChatMessage{{Role: "user", Content: types.StringContent("x")}},
	}
	resp, src, _, err := s.Chat(context.Background(), []Provider{cold, hot}, req)
	if err != nil || src != "hot" {
		t.Fatalf("want hot to serve, got src=%q err=%v", src, err)
	}
	if resp == nil || resp.Content != "hot" {
		t.Fatalf("bad response: %#v", resp)
	}
	if cold.chatHits.Load() != 0 {
		t.Fatalf("cooled candidate must not be hit, got %d hits", cold.chatHits.Load())
	}
	// all candidates cooled: direct path reports it instead of hitting upstream
	hot.available.Store(false)
	hits := hot.chatHits.Load()
	_, _, _, err = s.Chat(context.Background(), []Provider{cold, hot}, req)
	if err == nil || !strings.Contains(err.Error(), "cooling down") {
		t.Fatalf("want cooling-down error, got %v", err)
	}
	if hot.chatHits.Load() != hits || cold.chatHits.Load() != 0 {
		t.Fatalf("cooled candidates must not be hit, got hot=%d cold=%d", hot.chatHits.Load(), cold.chatHits.Load())
	}
}

// lookup resolves (model, source) -> provider for chain builders in tests.
type lookup func(m, s string) Provider

func bySource(ps ...*fakeProvider) lookup {
	return func(m, s string) Provider {
		for _, p := range ps {
			if p.name == s {
				return p
			}
		}
		return nil
	}
}

func chatReq() *types.ChatRequest {
	return &types.ChatRequest{Messages: []types.ChatMessage{{Role: "user", Content: types.StringContent("hi")}}}
}

func TestBuildChainTierOrdering(t *testing.T) {
	a, b, c, d := fakeProv("s1", "a"), fakeProv("s2", "b"), fakeProv("s3", "c"), fakeProv("s4", "d")
	ch := BuildChain([]Rule{
		{Tier: 1, Model: "a", Source: "s1"},
		{Tier: 3, Model: "c", Source: "s3"},
		{Tier: 2, Model: "b", Source: "s2"},
		{Tier: 1, Model: "d", Source: "s4"},
		{Tier: 9, Model: "gone", Source: "missing"},
	}, bySource(a, b, c, d))
	if len(ch.Tiers) != 3 {
		t.Fatalf("tiers = %d, want 3", len(ch.Tiers))
	}
	if ch.Tiers[0].Tier != 1 || ch.Tiers[1].Tier != 2 || ch.Tiers[2].Tier != 3 {
		t.Fatalf("tier order = %d,%d,%d, want 1,2,3", ch.Tiers[0].Tier, ch.Tiers[1].Tier, ch.Tiers[2].Tier)
	}
	// same-tier slots keep configured order, unresolvable rule is dropped
	if len(ch.Tiers[0].Slots) != 2 || ch.Tiers[0].Slots[0].Model != "a" || ch.Tiers[0].Slots[1].Model != "d" {
		t.Fatalf("tier 1 slots = %+v", ch.Tiers[0].Slots)
	}
}

func TestChainRoundRobin(t *testing.T) {
	a, b := fakeProv("s1", "a"), fakeProv("s2", "b")
	ch := BuildChain([]Rule{
		{Tier: 0, Model: "a", Source: "s1"},
		{Tier: 0, Model: "b", Source: "s2"},
	}, bySource(a, b))
	s := New(0)
	var got []string
	for i := 0; i < 4; i++ {
		_, src, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil, nil)
		if err != nil {
			t.Fatalf("iter %d: %v", i, err)
		}
		got = append(got, src)
	}
	want := []string{"s1", "s2", "s1", "s2"}
	for i := range want {
		if got[i] != want[i] {
			t.Fatalf("rr order = %v, want %v", got, want)
		}
	}
}

func TestChainPreferenceSinksButStaysReachable(t *testing.T) {
	neg := fakeProv("neg", "n")
	good := fakeProv("good", "g")
	neg.pref.Store(-5)
	good.fail.Store(true)
	ch := BuildChain([]Rule{
		{Tier: 0, Model: "n", Source: "neg"},
		{Tier: 0, Model: "g", Source: "good"},
	}, bySource(neg, good))
	s := New(0)
	resp, src, model, err := s.ChainChat(context.Background(), ch, chatReq(), nil, nil)
	if err != nil {
		t.Fatalf("chain: %v", err)
	}
	// Round-robin starts at index 0 (neg). neg is available (pref=-5 > -20)
	// and succeeds. good is never tried because neg already won.
	if src != "neg" || model != "n" || resp.Content != "neg" {
		t.Fatalf("served src=%q model=%q content=%q", src, model, resp.Content)
	}
	if good.chatHits.Load() != 0 {
		t.Fatal("neg was tried first and succeeded; good must not be attempted")
	}
	// Second request: cursor advances. neg wins again (good hard-fails).
	resp2, src2, _, err2 := s.ChainChat(context.Background(), ch, chatReq(), nil, nil)
	if err2 != nil {
		t.Fatalf("second chain: %v", err2)
	}
	if src2 != "neg" || resp2.Content != "neg" {
		t.Fatalf("second req src=%q content=%q", src2, resp2.Content)
	}
}

func TestChainBusySkipsWithoutPenalty(t *testing.T) {
	a, b := fakeProv("s1", "a"), fakeProv("s2", "b")
	a.busy.Store(true)
	ch := BuildChain([]Rule{
		{Tier: 0, Model: "a", Source: "s1"},
		{Tier: 0, Model: "b", Source: "s2"},
	}, bySource(a, b))
	s := New(0)
	_, src, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil, nil)
	if err != nil {
		t.Fatalf("chain: %v", err)
	}
	if src != "s2" {
		t.Fatalf("src = %q, want s2", src)
	}
	if a.chatHits.Load() == 0 {
		t.Fatal("busy slot must have been attempted")
	}
}

func TestChainAllBusyBoundedWaitThenNextTier(t *testing.T) {
	oldWait, oldPoll := busyWait, busyPoll
	busyWait, busyPoll = 60*time.Millisecond, 10*time.Millisecond
	t.Cleanup(func() { busyWait, busyPoll = oldWait, oldPoll })
	a, b, c := fakeProv("s1", "a"), fakeProv("s2", "b"), fakeProv("s3", "c")
	a.busy.Store(true)
	b.busy.Store(true)
	ch := BuildChain([]Rule{
		{Tier: 1, Model: "a", Source: "s1"},
		{Tier: 1, Model: "b", Source: "s2"},
		{Tier: 2, Model: "c", Source: "s3"},
	}, bySource(a, b, c))
	s := New(0)
	t0 := time.Now()
	_, src, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil, nil)
	el := time.Since(t0)
	if err != nil {
		t.Fatalf("chain: %v", err)
	}
	if src != "s3" {
		t.Fatalf("src = %q, want s3 (downgrade after bounded wait)", src)
	}
	if el > time.Second {
		t.Fatalf("busy wait not bounded: %v", el)
	}
}

func TestChainQuotaExhausted(t *testing.T) {
	a, b := fakeProv("s1", "a"), fakeProv("s2", "b")
	ch := BuildChain([]Rule{
		{Tier: 0, Model: "a", Source: "s1", Quota: 100, Period: "hour"},
		{Tier: 0, Model: "b", Source: "s2"},
	}, bySource(a, b))
	s := New(0)
	exhausted := func(sl *Slot) bool { return sl.Source == "s1" && sl.Quota > 0 }
	_, src, _, err := s.ChainChat(context.Background(), ch, chatReq(), exhausted, nil)
	if err != nil {
		t.Fatalf("chain: %v", err)
	}
	if src != "s2" {
		t.Fatalf("src = %q, want s2 (quota slot dropped)", src)
	}
	if a.chatHits.Load() != 0 {
		t.Fatal("quota-exhausted slot must not be called")
	}
}

func TestChainErrSummary(t *testing.T) {
	a, b, c := fakeProv("s1", "a"), fakeProv("s2", "b"), fakeProv("s3", "c")
	a.fail.Store(true)
	b.fail.Store(true)
	c.available.Store(false) // whole tier 1 cooling
	ch := BuildChain([]Rule{
		{Tier: 0, Model: "a", Source: "s1"},
		{Tier: 0, Model: "b", Source: "s2"},
		{Tier: 1, Model: "c", Source: "s3"},
	}, bySource(a, b, c))
	s := New(0)
	_, _, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil, nil)
	var ce *ChainErr
	if !errors.As(err, &ce) {
		t.Fatalf("err = %v, want *ChainErr", err)
	}
	if len(ce.Tiers) != 2 || ce.Tiers[0].Source != "s1" || ce.Tiers[1].Source != "s2" {
		t.Fatalf("tiers = %+v", ce.Tiers)
	}
	if len(ce.Skipped) != 1 {
		t.Fatalf("skipped = %+v", ce.Skipped)
	}
	msg := ce.Error()
	if !strings.Contains(msg, "s1") || !strings.Contains(msg, "s2") || !strings.Contains(msg, "tier 1: no schedulable slot") {
		t.Fatalf("summary = %q", msg)
	}
}

func TestChainStreamFallsBackBeforeFirstChunk(t *testing.T) {
	a, b := fakeProv("s1", "a"), fakeProv("s2", "b")
	a.fail.Store(true)
	ch := BuildChain([]Rule{
		{Tier: 0, Model: "a", Source: "s1"},
		{Tier: 0, Model: "b", Source: "s2"},
	}, bySource(a, b))
	s := New(0)
	chunks, src, model, err := s.ChainChatStream(context.Background(), ch, chatReq(), nil, nil)
	if err != nil {
		t.Fatalf("chain stream: %v", err)
	}
	if src != "s2" || model != "b" {
		t.Fatalf("src=%q model=%q", src, model)
	}
	var text string
	for ck := range chunks {
		text += ck.Content
	}
	if text != "s2" {
		t.Fatalf("text = %q", text)
	}
}

func TestChainResetCooldownAfterSwap(t *testing.T) {
	a := fakeProv("s1", "a")
	ch := BuildChain([]Rule{{Tier: 0, Model: "a", Source: "s1"}}, bySource(a))
	// a freshly built chain must schedule from index 0 (cursor starts at -1)
	if base := ch.Tiers[0].NextStart(); base != 0 {
		t.Fatalf("first start = %d, want 0", base)
	}
}

// ---- half-cooldown probe candidates (plan 阶段 1.3) ----

// TestChainProbeIsLastResort pins the ordering rule: a probe candidate must not
// steal traffic from a healthy slot in the same tier.
func TestChainProbeIsLastResort(t *testing.T) {
	healthy := fakeProv("healthy", "h")
	cooling := fakeProv("cooling", "c")
	cooling.available.Store(false)
	cooling.probeable.Store(true)
	ch := BuildChain([]Rule{
		{Tier: 0, Model: "c", Source: "cooling"}, // configured FIRST on purpose
		{Tier: 0, Model: "h", Source: "healthy"},
	}, bySource(healthy, cooling))
	s := New(0)
	for i := 0; i < 3; i++ {
		_, src, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil, nil)
		if err != nil {
			t.Fatalf("iter %d: %v", i, err)
		}
		if src != "healthy" {
			t.Fatalf("iter %d served by %q; a probe slot must never outrank a healthy one", i, src)
		}
	}
	if cooling.chatHits.Load() != 0 {
		t.Fatalf("probe slot was hit %d times while a healthy slot existed", cooling.chatHits.Load())
	}
	if got := cooling.probeDones.Load(); got != cooling.probeClaims.Load() {
		t.Fatalf("probe permits leaked: claims=%d dones=%d", cooling.probeClaims.Load(), got)
	}
}

// TestChainProbeServesWhenNothingElseCan is the recovery path: with every normal
// slot gone, the cooling slot's probe carries the request instead of the tier
// failing outright.
func TestChainProbeServesWhenNothingElseCan(t *testing.T) {
	cooling := fakeProv("cooling", "c")
	cooling.available.Store(false)
	cooling.probeable.Store(true)
	ch := BuildChain([]Rule{{Tier: 0, Model: "c", Source: "cooling"}}, bySource(cooling))
	s := New(0)
	resp, src, model, err := s.ChainChat(context.Background(), ch, chatReq(), nil, nil)
	if err != nil {
		t.Fatalf("probe must serve the request: %v", err)
	}
	if src != "cooling" || model != "c" {
		t.Fatalf("served by src=%q model=%q, want cooling/c", src, model)
	}
	if resp == nil || resp.Content != "cooling" {
		t.Fatalf("bad response: %#v", resp)
	}
	if cooling.chatHits.Load() != 1 {
		t.Fatalf("probe must issue exactly one request, got %d", cooling.chatHits.Load())
	}
	if cooling.probeClaims.Load() != 1 || cooling.probeDones.Load() != 1 {
		t.Fatalf("permit accounting: claims=%d dones=%d, want 1/1", cooling.probeClaims.Load(), cooling.probeDones.Load())
	}
}

// TestChainProbePermitReleasedOnFailure: a failed probe must still hand its
// permit back, otherwise the slot could never be probed again.
func TestChainProbePermitReleasedOnFailure(t *testing.T) {
	cooling := fakeProv("cooling", "c")
	cooling.available.Store(false)
	cooling.probeable.Store(true)
	cooling.fail.Store(true)
	ch := BuildChain([]Rule{{Tier: 0, Model: "c", Source: "cooling"}}, bySource(cooling))
	s := New(0)
	if _, _, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil, nil); err == nil {
		t.Fatal("expected the failing probe to surface an error")
	}
	if cooling.probeClaims.Load() != 1 || cooling.probeDones.Load() != 1 {
		t.Fatalf("permit accounting: claims=%d dones=%d, want 1/1", cooling.probeClaims.Load(), cooling.probeDones.Load())
	}
	// permit is free again for the next attempt
	if _, _, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil, nil); err == nil {
		t.Fatal("expected the second probe to fail too")
	}
	if cooling.probeClaims.Load() != 2 {
		t.Fatalf("permit must be reclaimable, claims=%d", cooling.probeClaims.Load())
	}
}

// TestChainProbeDoesNotBlockTierFallthrough: a cooling tier-1 slot that is not
// probeable must still let the request fall through to tier 2.
func TestChainProbeDoesNotBlockTierFallthrough(t *testing.T) {
	cold := fakeProv("cold", "c")
	cold.available.Store(false) // cooling and NOT probeable
	backup := fakeProv("backup", "b")
	ch := BuildChain([]Rule{
		{Tier: 1, Model: "c", Source: "cold"},
		{Tier: 2, Model: "b", Source: "backup"},
	}, bySource(cold, backup))
	s := New(0)
	_, src, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil, nil)
	if err != nil || src != "backup" {
		t.Fatalf("want fallthrough to backup, got src=%q err=%v", src, err)
	}
	if cold.chatHits.Load() != 0 {
		t.Fatal("non-probeable cooling slot must not be hit")
	}
}

// TestDirectProbeReleasesPermit covers the non-AUTO paths (Chat/ChatStream/Image).
func TestDirectProbeReleasesPermit(t *testing.T) {
	cooling := fakeProv("cooling", "m")
	cooling.available.Store(false)
	cooling.probeable.Store(true)
	s := New(0)
	_, src, _, err := s.Chat(context.Background(), []Provider{cooling}, chatReq())
	if err != nil || src != "cooling" {
		t.Fatalf("direct probe must serve: src=%q err=%v", src, err)
	}
	if cooling.probeClaims.Load() != 1 || cooling.probeDones.Load() != 1 {
		t.Fatalf("permit accounting: claims=%d dones=%d", cooling.probeClaims.Load(), cooling.probeDones.Load())
	}
}

// TestChainProbeRoundRobinUnaffected: adding a probe tail must not disturb the
// round-robin rotation over the healthy head.
func TestChainProbeRoundRobinUnaffected(t *testing.T) {
	a, b := fakeProv("s1", "a"), fakeProv("s2", "b")
	cooling := fakeProv("s3", "c")
	cooling.available.Store(false)
	cooling.probeable.Store(true)
	ch := BuildChain([]Rule{
		{Tier: 0, Model: "a", Source: "s1"},
		{Tier: 0, Model: "b", Source: "s2"},
		{Tier: 0, Model: "c", Source: "s3"},
	}, bySource(a, b, cooling))
	s := New(0)
	var got []string
	for i := 0; i < 4; i++ {
		_, src, _, err := s.ChainChat(context.Background(), ch, chatReq(), nil, nil)
		if err != nil {
			t.Fatalf("iter %d: %v", i, err)
		}
		got = append(got, src)
	}
	want := []string{"s1", "s2", "s1", "s2"}
	for i := range want {
		if got[i] != want[i] {
			t.Fatalf("rr order = %v, want %v (probe tail must not join the rotation)", got, want)
		}
	}
}