Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions constant/context_key.go
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,10 @@ const (
// ContextKeyAdminRejectReason stores an admin-only reject/block reason extracted from upstream responses.
// It is not returned to end users, but can be persisted into consume/error logs for debugging.
ContextKeyAdminRejectReason ContextKey = "admin_reject_reason"
// ContextKeyUpstreamRefusal is a provider-neutral signal used by routing policies.
ContextKeyUpstreamRefusal ContextKey = "upstream_refusal"
// ContextKeyRoutingGroup overrides channel selection without changing the user's billing group.
ContextKeyRoutingGroup ContextKey = "routing_group"

// ContextKeyLanguage stores the user's language preference for i18n
ContextKeyLanguage ContextKey = "language"
Expand Down
9 changes: 9 additions & 0 deletions controller/option.go
Original file line number Diff line number Diff line change
Expand Up @@ -295,6 +295,15 @@ func UpdateOption(c *gin.Context) {
})
return
}
case "refusal_fallback_setting.rules":
err = operation_setting.ValidateRefusalFallbackRules(option.Value.(string))
if err != nil {
c.JSON(http.StatusOK, gin.H{
"success": false,
"message": err.Error(),
})
return
}
case "console_setting.api_info":
err = console_setting.ValidateConsoleSettings(option.Value.(string), "ApiInfo")
if err != nil {
Expand Down
4 changes: 2 additions & 2 deletions controller/relay.go
Original file line number Diff line number Diff line change
Expand Up @@ -180,7 +180,7 @@ func Relay(c *gin.Context, relayFormat types.RelayFormat) {

retryParam := &service.RetryParam{
Ctx: c,
TokenGroup: relayInfo.TokenGroup,
TokenGroup: service.ResolveRoutingGroup(c, relayInfo.TokenGroup),
ModelName: relayInfo.OriginModelName,
RequestPath: c.Request.URL.Path,
Retry: common.GetPointer(0),
Expand Down Expand Up @@ -509,7 +509,7 @@ func RelayTask(c *gin.Context) {

retryParam := &service.RetryParam{
Ctx: c,
TokenGroup: relayInfo.TokenGroup,
TokenGroup: service.ResolveRoutingGroup(c, relayInfo.TokenGroup),
ModelName: relayInfo.OriginModelName,
RequestPath: c.Request.URL.Path,
Retry: common.GetPointer(0),
Expand Down
79 changes: 53 additions & 26 deletions middleware/distributor.go
Original file line number Diff line number Diff line change
Expand Up @@ -101,32 +101,32 @@ func Distribute() func(c *gin.Context) {
}
}

if preferredChannelID, found := service.GetPreferredChannelByAffinity(c, modelRequest.Model, usingGroup); found {
affinityUsable := false
preferred, err := model.CacheGetChannel(preferredChannelID)
if err == nil && preferred != nil && preferred.Status == common.ChannelStatusEnabled &&
channelSupportsRequestPath(preferred, c.Request.URL.Path, modelRequest.Model) {
if usingGroup == "auto" {
userGroup := common.GetContextKeyString(c, constant.ContextKeyUserGroup)
autoGroups := service.GetUserAutoGroup(userGroup)
for _, g := range autoGroups {
if model.IsChannelEnabledForGroupModel(g, modelRequest.Model, preferred.Id) {
selectGroup = g
common.SetContextKey(c, constant.ContextKeyAutoGroup, g)
channel = preferred
affinityUsable = true
service.MarkChannelAffinityUsed(c, g, preferred.Id)
break
}
}
} else if model.IsChannelEnabledForGroupModel(usingGroup, modelRequest.Model, preferred.Id) {
channel = preferred
selectGroup = usingGroup
affinityUsable = true
service.MarkChannelAffinityUsed(c, usingGroup, preferred.Id)
}
fallbackGroup, fallbackFound := service.GetRefusalFallbackGroup(c, modelRequest.Model, usingGroup)
preferredChannelID, affinityFound := service.GetPreferredChannelByAffinity(c, modelRequest.Model, usingGroup)

if fallbackFound {
channel, selectGroup, err = service.CacheGetRandomSatisfiedChannel(&service.RetryParam{
Ctx: c,
ModelName: modelRequest.Model,
TokenGroup: fallbackGroup,
RequestPath: c.Request.URL.Path,
Retry: common.GetPointer(0),
})
if err == nil && channel != nil {
common.SetContextKey(c, constant.ContextKeyRoutingGroup, fallbackGroup)
service.MarkRefusalFallbackUsed(c, selectGroup, channel.Id)
} else {
service.ClearCurrentRefusalFallback(c)
channel = nil
err = nil
}
if !affinityUsable && !service.ShouldKeepChannelAffinityOnChannelDisabled() {
}

if channel == nil && affinityFound {
channel, selectGroup = selectUsableConfiguredChannel(c, preferredChannelID, modelRequest.Model, usingGroup)
if channel != nil {
service.MarkChannelAffinityUsed(c, selectGroup, channel.Id)
} else if !service.ShouldKeepChannelAffinityOnChannelDisabled() {
service.ClearCurrentChannelAffinityCache(c)
}
}
Expand Down Expand Up @@ -164,9 +164,36 @@ func Distribute() func(c *gin.Context) {
SetupContextForSelectedChannel(c, channel, modelRequest.Model)
c.Next()
if channel != nil && c.Writer != nil && c.Writer.Status() < http.StatusBadRequest {
service.RecordChannelAffinity(c, channel.Id)
if !service.WasRefusalFallbackUsed(c) {
service.RecordChannelAffinity(c, channel.Id)
}
}
service.ObserveRefusalFallback(c)
}
}

func selectUsableConfiguredChannel(c *gin.Context, channelID int, modelName, usingGroup string) (*model.Channel, string) {
selected, err := model.CacheGetChannel(channelID)
if err != nil || selected == nil || selected.Status != common.ChannelStatusEnabled ||
!channelSupportsRequestPath(selected, c.Request.URL.Path, modelName) {
return nil, ""
}

if usingGroup != "auto" {
if model.IsChannelEnabledForGroupModel(usingGroup, modelName, selected.Id) {
return selected, usingGroup
}
return nil, ""
}

userGroup := common.GetContextKeyString(c, constant.ContextKeyUserGroup)
for _, group := range service.GetUserAutoGroup(userGroup) {
if model.IsChannelEnabledForGroupModel(group, modelName, selected.Id) {
common.SetContextKey(c, constant.ContextKeyAutoGroup, group)
return selected, group
}
}
return nil, ""
}

// channelSupportsRequestPath reports whether a channel can serve the request path.
Expand Down
29 changes: 29 additions & 0 deletions pkg/cachex/hybrid_cache.go
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ type HybridCache[V any] struct {
redisEnabled func() bool

memOnce sync.Once
memMu sync.Mutex
memInit func() *hot.HotCache[string, V]
mem *hot.HotCache[string, V]
}
Expand Down Expand Up @@ -128,6 +129,34 @@ func (c *HybridCache[V]) SetWithTTL(key string, v V, ttl time.Duration) error {
return nil
}

// SetIfAbsentWithTTL stores a value only when the key does not already exist.
// The Redis path uses SET NX, while the in-memory path serializes the check and
// write so concurrent callers cannot refresh the original TTL.
func (c *HybridCache[V]) SetIfAbsentWithTTL(key string, v V, ttl time.Duration) (bool, error) {
full := c.ns.FullKey(key)
if full == "" {
return false, nil
}

if c.redisOn() {
raw, err := c.redisCodec.Encode(v)
if err != nil {
return false, err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultRedisOpTimeout)
defer cancel()
return c.redis.SetNX(ctx, full, raw, ttl).Result()
}

c.memMu.Lock()
defer c.memMu.Unlock()
if _, found, _ := c.memCache().Get(full); found {
return false, nil
}
c.memCache().SetWithTTL(full, v, ttl)
return true, nil
}

// Keys returns keys with valid values. In Redis, it returns all matching keys.
func (c *HybridCache[V]) Keys() ([]string, error) {
if c.redisOn() {
Expand Down
36 changes: 36 additions & 0 deletions pkg/cachex/hybrid_cache_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
package cachex

import (
"testing"
"time"

"github.com/samber/hot"
"github.com/stretchr/testify/require"
)

func TestSetIfAbsentWithTTLDoesNotOverwriteOrRefresh(t *testing.T) {
cache := NewHybridCache[string](HybridCacheConfig[string]{
Namespace: "set-if-absent-test",
Memory: func() *hot.HotCache[string, string] {
return hot.NewHotCache[string, string](hot.LRU, 10).Build()
},
})

created, err := cache.SetIfAbsentWithTTL("key", "first", 300*time.Millisecond)
require.NoError(t, err)
require.True(t, created)

time.Sleep(150 * time.Millisecond)
created, err = cache.SetIfAbsentWithTTL("key", "second", 300*time.Millisecond)
require.NoError(t, err)
require.False(t, created)
value, found, err := cache.Get("key")
require.NoError(t, err)
require.True(t, found)
require.Equal(t, "first", value)

time.Sleep(200 * time.Millisecond)
_, found, err = cache.Get("key")
require.NoError(t, err)
require.False(t, found)
}
1 change: 1 addition & 0 deletions relay/channel/claude/relay-claude.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ func maybeMarkClaudeRefusal(c *gin.Context, stopReason string) {
}
if strings.EqualFold(stopReason, "refusal") {
common.SetContextKey(c, constant.ContextKeyAdminRejectReason, "claude_stop_reason=refusal")
common.SetContextKey(c, constant.ContextKeyUpstreamRefusal, true)
}
}

Expand Down
20 changes: 20 additions & 0 deletions relay/channel/claude/relay_claude_test.go
Original file line number Diff line number Diff line change
@@ -1,11 +1,15 @@
package claude

import (
"net/http/httptest"
"strings"
"testing"

"github.com/QuantumNous/new-api/common"
"github.com/QuantumNous/new-api/constant"
"github.com/QuantumNous/new-api/dto"
"github.com/QuantumNous/new-api/service/relayconvert"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
Expand All @@ -14,6 +18,22 @@ func commonPointer[T any](value T) *T {
return &value
}

func TestMaybeMarkClaudeRefusalSetsProviderNeutralRoutingSignal(t *testing.T) {
ctx, _ := gin.CreateTestContext(httptest.NewRecorder())
maybeMarkClaudeRefusal(ctx, "refusal")

require.True(t, common.GetContextKeyBool(ctx, constant.ContextKeyUpstreamRefusal))
require.Equal(
t,
"claude_stop_reason=refusal",
common.GetContextKeyString(ctx, constant.ContextKeyAdminRejectReason),
)

normal, _ := gin.CreateTestContext(httptest.NewRecorder())
maybeMarkClaudeRefusal(normal, "end_turn")
require.False(t, common.GetContextKeyBool(normal, constant.ContextKeyUpstreamRefusal))
}

func TestResponseOpenAI2ClaudeToolUseInputIsObject(t *testing.T) {
tests := []struct {
name string
Expand Down
10 changes: 10 additions & 0 deletions service/channel_select.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,16 @@ type RetryParam struct {
resetNextTry bool
}

// ResolveRoutingGroup returns an internal routing override without changing
// ContextKeyUsingGroup, which remains the source of billing semantics.
func ResolveRoutingGroup(ctx *gin.Context, defaultGroup string) string {
override := common.GetContextKeyString(ctx, constant.ContextKeyRoutingGroup)
if override != "" {
return override
}
return defaultGroup
}

func (p *RetryParam) GetRetry() int {
if p.Retry == nil {
return 0
Expand Down
1 change: 1 addition & 0 deletions service/log_info_generate.go
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,7 @@ func GenerateTextOtherInfo(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, m
}

AppendChannelAffinityAdminInfo(ctx, adminInfo)
AppendRefusalFallbackAdminInfo(ctx, adminInfo)

other["admin_info"] = adminInfo
appendRequestPath(ctx, relayInfo, other)
Expand Down
Loading