fix(health): 改进凭据自动恢复检测

This commit is contained in:
tbphp committed 2026-08-11 11:33:50 +08:00
1 parent 2b0a104929
commit 1ef573ae34
31 files changed
+533 -147

No files matched your search

-20
View File
@@ -184,26 +184,6 @@ func utilityRequestShape(
case channel.ProviderGemini:
return protocol.Gemini, http.MethodGet, "/v1beta/models", nil, nil
}
case execution.OperationProbe:
model = strings.TrimSpace(model)
if model == "" {
return "", "", "", nil, app_errors.ErrValidation
}
switch providerKind {
case channel.ProviderOpenAI, channel.ProviderOpenAICompatible,
channel.ProviderDeepSeek, channel.ProviderOpenRouter, channel.ProviderGroq, channel.ProviderXAI:
return protocol.OpenAICompletions, http.MethodPost, "/v1/chat/completions",
[]byte(`{"model":"probe","messages":[{"role":"user","content":"ping"}],"max_tokens":1}`), nil
case channel.ProviderAzureOpenAI, channel.ProviderAWSBedrock, channel.ProviderGoogleVertex:
return protocol.OpenAICompletions, http.MethodPost, "/v1/chat/completions",
[]byte(`{"model":"probe","messages":[{"role":"user","content":"ping"}],"max_tokens":1}`), nil
case channel.ProviderAnthropic:
return protocol.Anthropic, http.MethodPost, "/v1/messages",
[]byte(`{"model":"probe","max_tokens":1,"messages":[{"role":"user","content":"ping"}]}`), nil
case channel.ProviderGemini:
return protocol.Gemini, http.MethodPost, "/v1beta/models/probe:generateContent",
[]byte(`{"contents":[{"parts":[{"text":"ping"}]}],"generationConfig":{"maxOutputTokens":1}}`), nil
}
}
return "", "", "", nil, app_errors.ErrValidation
}
+3 -5
View File
@@ -141,14 +141,12 @@ func TestNativeOpenAIStyleProvidersUseOpenAIUtilityRequestShape(t *testing.T) {
t.Fatalf("ListModels shape = %q %q %q %s, err=%v", clientProtocol, method, path, body, err)
}
clientProtocol, method, path, body, err = utilityRequestShape(
if _, _, _, _, err := utilityRequestShape(
providerKind,
execution.OperationProbe,
"model-one",
)
if err != nil || clientProtocol != protocol.OpenAICompletions ||
method != http.MethodPost || path != "/v1/chat/completions" || len(body) == 0 {
t.Fatalf("Probe shape = %q %q %q %s, err=%v", clientProtocol, method, path, body, err)
); err == nil {
t.Fatal("utilityRequestShape still owns provider Probe wire shape")
}
})
}
+55 -9
View File
@@ -14,10 +14,10 @@ import (
)
const (
autoWeightInterval = 30 * time.Second
validationInterval = 30 * time.Minute
maxValidationJitter = 3 * time.Minute
retentionInterval = time.Hour
autoWeightInterval = 30 * time.Second
defaultValidationInterval = 10 * time.Minute
maxValidationJitter = 3 * time.Minute
retentionInterval = time.Hour
)
type autoWeightRegistry interface {
@@ -68,6 +68,7 @@ type Runtime struct {
requestLogCleaner RequestLogCleaner
operationRecovery operationRecoveryRuntime
catalogSync catalogSyncRuntime
manager *state.Manager
autoWeightInterval time.Duration
validationInterval time.Duration
validationJitter func() time.Duration
@@ -94,8 +95,9 @@ func NewRuntime(
requestLogCleaner: requestLogCleaner,
operationRecovery: operationRecovery,
catalogSync: catalogSync,
manager: manager,
autoWeightInterval: autoWeightInterval,
validationInterval: validationInterval,
validationInterval: defaultValidationInterval,
validationJitter: func() time.Duration {
return time.Duration(rand.Int64N(int64(maxValidationJitter) + 1))
},
@@ -118,7 +120,11 @@ func NewRuntime(
func (runtime *Runtime) Run(ctx context.Context) {
autoTicker := runtime.newTicker(runtime.autoWeightInterval)
validationTicker := runtime.newTicker(runtime.validationInterval + runtime.validationJitter())
currentValidationInterval, validationUpdates := runtime.currentValidationSchedule()
validationTicker := runtime.newTicker(validationTickerInterval(
currentValidationInterval,
runtime.validationJitter(),
))
var wait sync.WaitGroup
wait.Add(2)
@@ -128,7 +134,7 @@ func (runtime *Runtime) Run(ctx context.Context) {
}()
go func() {
defer wait.Done()
runtime.runValidation(ctx, validationTicker)
runtime.runValidation(ctx, validationTicker, currentValidationInterval, validationUpdates)
}()
if runtime.requestLogCleaner != nil {
retentionTicker := runtime.newTicker(retentionInterval)
@@ -170,12 +176,26 @@ func (runtime *Runtime) runAutoWeight(ctx context.Context, ticker runtimeTicker)
}
}
func (runtime *Runtime) runValidation(ctx context.Context, ticker runtimeTicker) {
defer ticker.Stop()
func (runtime *Runtime) runValidation(
ctx context.Context,
ticker runtimeTicker,
interval time.Duration,
updates <-chan struct{},
) {
defer func() { ticker.Stop() }()
for {
select {
case <-ctx.Done():
return
case <-updates:
nextInterval, nextUpdates := runtime.currentValidationSchedule()
updates = nextUpdates
if nextInterval == interval {
continue
}
ticker.Stop()
interval = nextInterval
ticker = runtime.newTicker(validationTickerInterval(interval, runtime.validationJitter()))
case <-ticker.C():
if ctx.Err() != nil {
return
@@ -187,6 +207,32 @@ func (runtime *Runtime) runValidation(ctx context.Context, ticker runtimeTicker)
}
}
func validationTickerInterval(interval, jitter time.Duration) time.Duration {
const maxDuration = time.Duration(1<<63 - 1)
if jitter <= 0 {
return interval
}
if interval > maxDuration-jitter {
return maxDuration
}
return interval + jitter
}
func (runtime *Runtime) currentValidationSchedule() (time.Duration, <-chan struct{}) {
interval := runtime.validationInterval
if interval <= 0 {
interval = defaultValidationInterval
}
if runtime.manager == nil {
return interval, nil
}
snapshot, updates := runtime.manager.CurrentWithUpdates()
if snapshot != nil && snapshot.Settings.ValidationInterval > 0 {
interval = snapshot.Settings.ValidationInterval
}
return interval, updates
}
func (runtime *Runtime) runRetention(ctx context.Context, ticker runtimeTicker) {
defer ticker.Stop()
if ctx.Err() != nil {
+93
View File
@@ -2,6 +2,7 @@ package control
import (
"context"
"encoding/json"
"reflect"
"sync"
"sync/atomic"
@@ -10,6 +11,7 @@ import (
"gpt-load/internal/channel"
"gpt-load/internal/health"
"gpt-load/internal/platform/config"
"gpt-load/internal/protocol"
"gpt-load/internal/state"
)
@@ -344,6 +346,97 @@ func TestRuntimeCreatesAutoWeightAndJitteredValidationTickers(t *testing.T) {
awaitSignal(t, validationTicker.stopped)
}
func TestRuntimeValidationJitterDoesNotOverflow(t *testing.T) {
autoTicker := newFakeRuntimeTicker()
validationTicker := newFakeRuntimeTicker()
created := make(chan time.Duration, 2)
runtime := newTestRuntime(
newFakeAutoWeightRegistry(1),
health.NewStatsStore(),
newFakeValidationSweep(false),
autoTicker,
validationTicker,
created,
time.Now,
)
runtime.validationInterval = time.Duration(1<<63 - 1)
runtime.validationJitter = func() time.Duration { return maxValidationJitter }
runtime.newTicker = func(interval time.Duration) runtimeTicker {
created <- interval
if interval == runtime.autoWeightInterval {
return autoTicker
}
return validationTicker
}
cancel, done := startRuntime(t, runtime)
if interval := awaitValue(t, created); interval != 30*time.Second {
t.Fatalf("auto-weight ticker interval = %v, want 30s", interval)
}
if interval := awaitValue(t, created); interval != time.Duration(1<<63-1) {
t.Fatalf("validation ticker interval = %v, want capped maximum duration", interval)
}
stopRuntime(t, cancel, done)
awaitSignal(t, autoTicker.stopped)
awaitSignal(t, validationTicker.stopped)
}
func TestRuntimeReschedulesValidationWhenPublishedIntervalChanges(t *testing.T) {
manager := state.NewManager()
if _, err := manager.Publish(state.CompileInput{}); err != nil {
t.Fatal(err)
}
autoTicker := newFakeRuntimeTicker()
defaultTicker := newFakeRuntimeTicker()
overriddenTicker := newFakeRuntimeTicker()
created := make(chan time.Duration, 3)
runtime := newTestRuntime(
newFakeAutoWeightRegistry(1),
health.NewStatsStore(),
newFakeValidationSweep(false),
autoTicker,
defaultTicker,
created,
time.Now,
)
runtime.manager = manager
runtime.newTicker = func(interval time.Duration) runtimeTicker {
created <- interval
switch interval {
case 30 * time.Second:
return autoTicker
case 12 * time.Minute:
return defaultTicker
case 22 * time.Minute:
return overriddenTicker
default:
testingPanic("unexpected ticker interval", interval)
return nil
}
}
cancel, done := startRuntime(t, runtime)
if interval := awaitValue(t, created); interval != 30*time.Second {
t.Fatalf("auto-weight ticker interval = %v, want 30s", interval)
}
if interval := awaitValue(t, created); interval != 12*time.Minute {
t.Fatalf("default validation ticker interval = %v, want 12m", interval)
}
if _, err := manager.Publish(state.CompileInput{SystemSettings: config.Settings{
state.SettingValidationInterval: json.Number("1200"),
}}); err != nil {
t.Fatal(err)
}
if interval := awaitValue(t, created); interval != 22*time.Minute {
t.Fatalf("updated validation ticker interval = %v, want 22m", interval)
}
awaitSignal(t, defaultTicker.stopped)
stopRuntime(t, cancel, done)
awaitSignal(t, autoTicker.stopped)
awaitSignal(t, overriddenTicker.stopped)
}
func TestRuntimeWaitsForTickBeforeRecompute(t *testing.T) {
registry := newFakeAutoWeightRegistry(1)
runtime, autoTicker, _, created := newRuntimeHarness(registry, health.NewStatsStore(), newFakeValidationSweep(false), time.Now)
+2 -2
View File
@@ -2282,13 +2282,13 @@ func TestSettingsHTTPStableSuccessEnvelopeUpdateAndReset(t *testing.T) {
NewServer(&config.Config{AuthKey: "test-auth-key"}, fixture.service).RegisterRoutes(engine)
get := serveSettingsRequest(t, engine, http.MethodGet, "test-auth-key", "")
const wantDefault = `{"code":0,"data":{"overrides":[],"values":{"first_byte_timeout":120,"header_rules":{"remove":[],"set":{}},"inject_usage_options":true,"models_dev_auto_sync_enabled":true,"request_log_retention_days":7,"request_timeout":600,"stream_idle_timeout":300}},"message":"Success"}`
const wantDefault = `{"code":0,"data":{"overrides":[],"values":{"first_byte_timeout":120,"header_rules":{"remove":[],"set":{}},"inject_usage_options":true,"models_dev_auto_sync_enabled":true,"request_log_retention_days":7,"request_timeout":600,"stream_idle_timeout":300,"validation_interval":600}},"message":"Success"}`
if get.Code != http.StatusOK || strings.TrimSpace(get.Body.String()) != wantDefault {
t.Fatalf("default response = %d %s, want %s", get.Code, get.Body.String(), wantDefault)
}
update := serveSettingsRequest(t, engine, http.MethodPut, "test-auth-key", `{"settings":{"request_timeout":900,"header_rules":{"set":{},"remove":[]}}}`)
const wantUpdate = `{"code":0,"data":{"overrides":["header_rules","request_timeout"],"values":{"first_byte_timeout":120,"header_rules":{"remove":[],"set":{}},"inject_usage_options":true,"models_dev_auto_sync_enabled":true,"request_log_retention_days":7,"request_timeout":900,"stream_idle_timeout":300}},"message":"Success"}`
const wantUpdate = `{"code":0,"data":{"overrides":["header_rules","request_timeout"],"values":{"first_byte_timeout":120,"header_rules":{"remove":[],"set":{}},"inject_usage_options":true,"models_dev_auto_sync_enabled":true,"request_log_retention_days":7,"request_timeout":900,"stream_idle_timeout":300,"validation_interval":600}},"message":"Success"}`
if update.Code != http.StatusOK || strings.TrimSpace(update.Body.String()) != wantUpdate {
t.Fatalf("update response = %d %s, want %s", update.Code, update.Body.String(), wantUpdate)
}
+2
View File
@@ -31,6 +31,7 @@ type SettingsValuesResponse struct {
StreamIdleTimeout int64 `json:"stream_idle_timeout"`
HeaderRules HeaderRulesResponse `json:"header_rules"`
InjectUsageOptions bool `json:"inject_usage_options"`
ValidationInterval int64 `json:"validation_interval"`
RequestLogRetentionDays int `json:"request_log_retention_days"`
ModelsDevAutoSyncEnabled bool `json:"models_dev_auto_sync_enabled"`
}
@@ -364,6 +365,7 @@ func mapSettingsResponse(
Remove: remove,
},
InjectUsageOptions: settings.InjectUsageOptions,
ValidationInterval: durationSeconds(settings.ValidationInterval),
RequestLogRetentionDays: settings.RequestLogRetentionDays,
ModelsDevAutoSyncEnabled: modelsDevAutoSyncEnabled,
},
+1
View File
@@ -37,6 +37,7 @@ func TestSettingsWireETagMatchesCheckedInFixture(t *testing.T) {
Remove: []string{},
},
InjectUsageOptions: true,
ValidationInterval: 600,
RequestLogRetentionDays: 7,
},
Overrides: []string{},
+1 -1
View File
@@ -29,7 +29,7 @@ func TestSettingsHTTPUsesCanonicalLocalizedStrongRepresentation(t *testing.T) {
t.Fatalf("English GET = %d %s", english.Code, english.Body.String())
}
assertSettingsRepresentationHeaders(t, english, "en-US")
const wantEnglish = `{"code":0,"data":{"overrides":[],"values":{"first_byte_timeout":120,"header_rules":{"remove":[],"set":{}},"inject_usage_options":true,"models_dev_auto_sync_enabled":true,"request_log_retention_days":7,"request_timeout":600,"stream_idle_timeout":300}},"message":"Success"}`
const wantEnglish = `{"code":0,"data":{"overrides":[],"values":{"first_byte_timeout":120,"header_rules":{"remove":[],"set":{}},"inject_usage_options":true,"models_dev_auto_sync_enabled":true,"request_log_retention_days":7,"request_timeout":600,"stream_idle_timeout":300,"validation_interval":600}},"message":"Success"}`
if english.Body.String() != wantEnglish {
t.Fatalf("English body = %s, want %s", english.Body.String(), wantEnglish)
}
+30
View File
@@ -34,6 +34,7 @@ func TestGetSettingsReturnsSnapshotDefaultsAndNoOverrides(t *testing.T) {
}
if got.Values.FirstByteTimeout != 120 ||
got.Values.RequestTimeout != 600 || got.Values.StreamIdleTimeout != 300 ||
got.Values.ValidationInterval != 600 ||
got.Values.RequestLogRetentionDays != 7 || !got.Values.InjectUsageOptions {
t.Fatalf("values = %#v", got.Values)
}
@@ -51,6 +52,35 @@ func TestGetSettingsReturnsSnapshotDefaultsAndNoOverrides(t *testing.T) {
}
}
func TestUpdateSettingsChangesAndResetsValidationInterval(t *testing.T) {
fixture := newServiceFixture(t)
updated, err := fixture.service.UpdateSettings(t.Context(), SettingsUpdateRequest{
Settings: map[string]json.RawMessage{
state.SettingValidationInterval: json.RawMessage("900"),
},
})
if err != nil {
t.Fatal(err)
}
if updated.Values.ValidationInterval != 900 ||
fixture.manager.Current().Settings.ValidationInterval != 15*time.Minute ||
!reflect.DeepEqual(updated.Overrides, []string{state.SettingValidationInterval}) {
t.Fatalf("updated validation interval = %#v", updated)
}
reset, err := fixture.service.UpdateSettings(t.Context(), SettingsUpdateRequest{
Settings: map[string]json.RawMessage{
state.SettingValidationInterval: json.RawMessage("null"),
},
})
if err != nil {
t.Fatal(err)
}
if reset.Values.ValidationInterval != 600 || len(reset.Overrides) != 0 {
t.Fatalf("reset validation interval = %#v", reset)
}
}
func TestModelsDevEnvironmentOverrideWinsAndIsReadOnlyWithoutPersistence(t *testing.T) {
fixture := newServiceFixture(t)
disabled := false
+3 -3
View File
@@ -1,6 +1,6 @@
{
"domain_hex": "6770742d6c6f61642f73657474696e67732d776972652d7631",
"canonical_response_hex": "7b22636f6465223a302c2264617461223a7b226f7665727269646573223a5b5d2c2276616c756573223a7b2266697273745f627974655f74696d656f7574223a3132302c226865616465725f72756c6573223a7b2272656d6f7665223a5b5d2c22736574223a7b7d7d2c22696e6a6563745f75736167655f6f7074696f6e73223a747275652c226d6f64656c735f6465765f6175746f5f73796e635f656e61626c6564223a66616c73652c22726571756573745f6c6f675f726574656e74696f6e5f64617973223a372c22726571756573745f74696d656f7574223a3630302c2273747265616d5f69646c655f74696d656f7574223a3330307d7d2c226d657373616765223a2253756363657373227d",
"framed_input_hex": "6770742d6c6f61642f73657474696e67732d776972652d7631007b22636f6465223a302c2264617461223a7b226f7665727269646573223a5b5d2c2276616c756573223a7b2266697273745f627974655f74696d656f7574223a3132302c226865616465725f72756c6573223a7b2272656d6f7665223a5b5d2c22736574223a7b7d7d2c22696e6a6563745f75736167655f6f7074696f6e73223a747275652c226d6f64656c735f6465765f6175746f5f73796e635f656e61626c6564223a66616c73652c22726571756573745f6c6f675f726574656e74696f6e5f64617973223a372c22726571756573745f74696d656f7574223a3630302c2273747265616d5f69646c655f74696d656f7574223a3330307d7d2c226d657373616765223a2253756363657373227d",
"expected_digest": "a7123aeb34c6078d958fe458586c575a0af8d9a41bf9add80f9d0757fba07e13"
"canonical_response_hex": "7b22636f6465223a302c2264617461223a7b226f7665727269646573223a5b5d2c2276616c756573223a7b2266697273745f627974655f74696d656f7574223a3132302c226865616465725f72756c6573223a7b2272656d6f7665223a5b5d2c22736574223a7b7d7d2c22696e6a6563745f75736167655f6f7074696f6e73223a747275652c226d6f64656c735f6465765f6175746f5f73796e635f656e61626c6564223a66616c73652c22726571756573745f6c6f675f726574656e74696f6e5f64617973223a372c22726571756573745f74696d656f7574223a3630302c2273747265616d5f69646c655f74696d656f7574223a3330302c2276616c69646174696f6e5f696e74657276616c223a3630307d7d2c226d657373616765223a2253756363657373227d",
"framed_input_hex": "6770742d6c6f61642f73657474696e67732d776972652d7631007b22636f6465223a302c2264617461223a7b226f7665727269646573223a5b5d2c2276616c756573223a7b2266697273745f627974655f74696d656f7574223a3132302c226865616465725f72756c6573223a7b2272656d6f7665223a5b5d2c22736574223a7b7d7d2c22696e6a6563745f75736167655f6f7074696f6e73223a747275652c226d6f64656c735f6465765f6175746f5f73796e635f656e61626c6564223a66616c73652c22726571756573745f6c6f675f726574656e74696f6e5f64617973223a372c22726571756573745f74696d656f7574223a3630302c2273747265616d5f69646c655f74696d656f7574223a3330302c2276616c69646174696f6e5f696e74657276616c223a3630307d7d2c226d657373616765223a2253756363657373227d",
"expected_digest": "480e9e8c4ea26881e16051a9a884673689543a711e8a003cd668261dbcbda80a"
}
+15 -22
View File
@@ -186,11 +186,6 @@ func (worker *validationWorker) validateRef(ctx context.Context, snapshot *state
return
}
apiKey, _ := credential.Value("api_key")
method, path, body, err := validationRequestShape(group, target)
if err != nil {
logValidationFailure(ref, string(target.protocol), "request")
return
}
routeMode, supported := group.ResolvedTarget.Mode(target.protocol, execution.OperationProbe)
if !supported {
logValidationFailure(ref, string(target.protocol), "request")
@@ -219,7 +214,7 @@ func (worker *validationWorker) validateRef(ctx context.Context, snapshot *state
ChannelID: string(group.ChannelID), TargetKind: string(group.ResolvedTarget.ProviderKind),
RouteMode: execution.RouteMode(routeMode), ClientProtocol: target.protocol,
Operation: execution.OperationProbe, ClientModel: target.model, UpstreamModel: target.model,
Method: method, Path: path, Header: applyControlHeaderRules(group.HeaderRules, apiKey), Body: body,
Header: applyControlHeaderRules(group.HeaderRules, apiKey),
TargetConfig: group.ResolvedTarget.TargetConfig,
Timeouts: executionTimeouts(group.Timeouts),
Credential: execution.NewCredentialSnapshot(
@@ -292,12 +287,8 @@ func buildGroupValidationTarget(group state.GroupView) (groupValidationTarget, b
if probeModel == "" {
return groupValidationTarget{}, false
}
selectedProtocol, _, _, _, err := utilityRequestShape(
group.ResolvedTarget.ProviderKind,
execution.OperationProbe,
probeModel,
)
if err != nil {
selectedProtocol, ok := validationProtocol(group.ResolvedTarget)
if !ok {
return groupValidationTarget{}, false
}
return groupValidationTarget{
@@ -307,16 +298,18 @@ func buildGroupValidationTarget(group state.GroupView) (groupValidationTarget, b
}, true
}
func validationRequestShape(
group state.GroupView,
target groupValidationTarget,
) (string, string, []byte, error) {
_, method, path, body, err := utilityRequestShape(
group.ResolvedTarget.ProviderKind,
execution.OperationProbe,
target.model,
)
return method, path, body, err
func validationProtocol(target channel.ResolvedTarget) (protocol.Protocol, bool) {
for _, clientProtocol := range protocol.DataPlaneProtocols() {
if mode, ok := target.Mode(clientProtocol, execution.OperationProbe); ok && mode == channel.RouteNative {
return clientProtocol, true
}
}
for _, clientProtocol := range protocol.DataPlaneProtocols() {
if _, ok := target.Mode(clientProtocol, execution.OperationProbe); ok {
return clientProtocol, true
}
}
return "", false
}
func computeGroupValidationSignature(
+43
View File
@@ -51,6 +51,45 @@ func TestValidationWorkerUsesExplicitModelAndCanonicalRepresentativeProtocol(t *
}
}
func TestValidationProtocolUsesFirstNativePresetProtocolThenConfiguredFallback(t *testing.T) {
t.Parallel()
tests := []struct {
name string
channelID channel.ID
params json.RawMessage
want protocol.Protocol
wantMode channel.RouteMode
}{
{name: "OpenAI", channelID: channel.OpenAI, params: json.RawMessage(`{}`), want: protocol.OpenAICompletions, wantMode: channel.RouteNative},
{name: "Anthropic", channelID: channel.Anthropic, params: json.RawMessage(`{}`), want: protocol.Anthropic, wantMode: channel.RouteNative},
{name: "Gemini", channelID: channel.Gemini, params: json.RawMessage(`{}`), want: protocol.Gemini, wantMode: channel.RouteNative},
{name: "OpenRouter", channelID: channel.OpenRouter, params: json.RawMessage(`{}`), want: protocol.OpenAICompletions, wantMode: channel.RouteNative},
{name: "OpenAI compatible", channelID: channel.OpenAICompatible, params: json.RawMessage(`{"base_url":"https://upstream.example/v1"}`), want: protocol.OpenAICompletions, wantMode: channel.RouteNative},
{name: "Azure", channelID: channel.AzureOpenAI, params: json.RawMessage(`{"endpoint":"https://example.openai.azure.com"}`), want: protocol.OpenAICompletions, wantMode: channel.RouteConverted},
{name: "Bedrock", channelID: channel.AWSBedrock, params: json.RawMessage(`{"region":"us-east-1"}`), want: protocol.OpenAICompletions, wantMode: channel.RouteConverted},
{name: "Vertex", channelID: channel.GoogleVertex, params: json.RawMessage(`{"project_id":"project","location":"us-central1"}`), want: protocol.OpenAICompletions, wantMode: channel.RouteConverted},
}
registry := channel.NewRegistry()
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
resolved, err := registry.Resolve(test.channelID, test.params)
if err != nil {
t.Fatal(err)
}
got, ok := validationProtocol(resolved)
if !ok || got != test.want {
t.Fatalf("validationProtocol() = %q/%t, want %q/true", got, ok, test.want)
}
mode, exists := resolved.Mode(got, execution.OperationProbe)
if !exists || mode != test.wantMode {
t.Fatalf("probe mode = %q/%t, want %q/true", mode, exists, test.wantMode)
}
})
}
}
func TestValidationWorkerFallsBackToFirstRealModelID(t *testing.T) {
probes := &validationProbeRecorder{}
worker := newValidationWorkerForTest(
@@ -136,6 +175,10 @@ func TestValidationWorkerProbesStructuredCloudCredential(t *testing.T) {
observed.UpstreamModel != "anthropic.claude-test" {
t.Fatalf("attempt = %#v", observed)
}
if observed.Method != "" || observed.Path != "" || observed.RawQuery != "" ||
len(observed.Query) != 0 || len(observed.Body) != 0 {
t.Fatalf("probe attempt contains provider wire shape: %#v", observed)
}
if got, want := worker.recorder.events(), []string{
fmt.Sprintf("registry.weight:7:%d", state.DefaultWeight), "registry.recover:7", "stats.reset:7",
}; !reflect.DeepEqual(got, want) {
@@ -178,11 +178,11 @@ func TestCompatibleListModelsAndProbeUseConfiguredPrefix(t *testing.T) {
baseSuffix string
modelsPath string
probePath string
probeBody []byte
probeReply string
}{
{name: "OpenAI", channelID: channel.OpenAICompatible, protocol: protocol.OpenAICompletions, baseSuffix: "/custom/v1", modelsPath: "/custom/v1/models", probePath: "/custom/v1/chat/completions", probeBody: []byte(`{"model":"client","messages":[{"role":"user","content":"ping"}]}`)},
{name: "Anthropic", channelID: channel.Anthropic, protocol: protocol.Anthropic, baseSuffix: "/custom", modelsPath: "/custom/v1/models", probePath: "/custom/v1/messages", probeBody: []byte(`{"model":"client","max_tokens":1,"messages":[{"role":"user","content":"ping"}]}`)},
{name: "Gemini", channelID: channel.Gemini, protocol: protocol.Gemini, baseSuffix: "/custom/v1beta", modelsPath: "/custom/v1beta/models", probePath: "/custom/v1beta/models/probe-upstream:generateContent", probeBody: []byte(`{"contents":[{"parts":[{"text":"ping"}]}]}`)},
{name: "OpenAI", channelID: channel.OpenAICompatible, protocol: protocol.OpenAICompletions, baseSuffix: "/custom/v1", modelsPath: "/custom/v1/models", probePath: "/custom/v1/chat/completions", probeReply: `{"id":"chat_1","object":"chat.completion","created":1,"model":"probe-upstream","choices":[{"index":0,"message":{"role":"assistant","content":"pong"},"finish_reason":"stop"}]}`},
{name: "Anthropic", channelID: channel.Anthropic, protocol: protocol.Anthropic, baseSuffix: "/custom", modelsPath: "/custom/v1/models", probePath: "/custom/v1/messages", probeReply: `{"id":"msg_1","type":"message","role":"assistant","model":"probe-upstream","content":[{"type":"text","text":"pong"}],"stop_reason":"end_turn","usage":{"input_tokens":1,"output_tokens":1}}`},
{name: "Gemini", channelID: channel.Gemini, protocol: protocol.Gemini, baseSuffix: "/custom/v1beta", modelsPath: "/custom/v1beta/models", probePath: "/custom/v1beta/models/probe-upstream:generateContent", probeReply: `{"candidates":[{"content":{"role":"model","parts":[{"text":"pong"}]},"finishReason":"STOP"}],"modelVersion":"probe-upstream"}`},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
@@ -199,7 +199,7 @@ func TestCompatibleListModelsAndProbeUseConfiguredPrefix(t *testing.T) {
t.Errorf("probe method = %s", request.Method)
}
writer.Header().Set("Content-Type", "application/json")
_, _ = io.WriteString(writer, `{"ok":true}`)
_, _ = io.WriteString(writer, test.probeReply)
default:
t.Errorf("unexpected path = %s", request.URL.Path)
writer.WriteHeader(http.StatusNotFound)
@@ -222,14 +222,7 @@ func TestCompatibleListModelsAndProbeUseConfiguredPrefix(t *testing.T) {
t.Fatalf("models = %+v err=%v", modelsResult, err)
}
probeClientPath := "/v1/chat/completions"
if test.protocol == protocol.Anthropic {
probeClientPath = "/v1/messages"
}
if test.protocol == protocol.Gemini {
probeClientPath = "/v1beta/models/client:generateContent"
}
probe := utilitySpec(test.channelID, test.protocol, execution.OperationProbe, http.MethodPost, probeClientPath, test.probeBody)
probe := utilitySpec(test.channelID, test.protocol, execution.OperationProbe, "", "", nil)
probe.TargetConfig = target
probe = freezeTestAttempt(probe)
probe.ClientModel, probe.UpstreamModel = "probe-client", "probe-upstream"
@@ -250,6 +243,13 @@ func TestOpenAICompatibleNonV1PrefixKeepsListModelsAndProbeFunctional(t *testing
case "/vendor/api/v4/models":
_, _ = io.WriteString(writer, `{"object":"list","data":[{"id":"model-1","object":"model","created":1,"owned_by":"vendor"}]}`)
case "/vendor/api/v4/chat/completions":
body, err := io.ReadAll(request.Body)
if err != nil {
t.Errorf("read probe body: %v", err)
} else if !bytes.Contains(body, []byte(`"max_tokens": 1`)) ||
bytes.Contains(body, []byte(`"max_completion_tokens"`)) {
t.Errorf("compatible probe token limit = %s, want legacy max_tokens only", body)
}
_, _ = io.WriteString(writer, `{"id":"chat_1","object":"chat.completion","created":1,"model":"served","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}`)
default:
t.Errorf("path = %s", request.URL.Path)
@@ -268,7 +268,7 @@ func TestOpenAICompatibleNonV1PrefixKeepsListModelsAndProbeFunctional(t *testing
t.Fatalf("models = %+v err=%v body=%s", modelsResult, err, modelsResult.Body)
}
probe := utilitySpec(channel.OpenAICompatible, protocol.OpenAICompletions, execution.OperationProbe, http.MethodPost, "/v1/chat/completions", []byte(`{"model":"client","messages":[{"role":"user","content":"ping"}]}`))
probe := utilitySpec(channel.OpenAICompatible, protocol.OpenAICompletions, execution.OperationProbe, "", "", nil)
probe.TargetConfig = target
probe = freezeTestAttempt(probe)
probe.ClientModel, probe.UpstreamModel = "probe-client", "probe-upstream"
+2 -2
View File
@@ -45,7 +45,7 @@ func buildConvertedResponsesRequest(spec execution.AttemptSpec, provider schemas
}
request = wire.ToBifrostResponsesRequest(conversionContext)
case protocol.Anthropic:
if spec.Operation != execution.OperationChatCompletion && spec.Operation != execution.OperationProbe {
if spec.Operation != execution.OperationChatCompletion {
return nil, fmt.Errorf("converted Anthropic operation is not supported")
}
var wire anthropic.AnthropicMessageRequest
@@ -54,7 +54,7 @@ func buildConvertedResponsesRequest(spec execution.AttemptSpec, provider schemas
}
request = wire.ToBifrostResponsesRequest(conversionContext)
case protocol.Gemini:
if spec.Operation != execution.OperationChatCompletion && spec.Operation != execution.OperationProbe {
if spec.Operation != execution.OperationChatCompletion {
return nil, fmt.Errorf("converted Gemini operation is not supported")
}
var wire gemini.GeminiGenerationRequest
+60 -13
View File
@@ -416,8 +416,7 @@ func (r *Runtime) prepare(spec execution.AttemptSpec, stream bool) (preparedAtte
if r.fixedConfig.custom {
targetBaseURL = r.fixedConfig.targetBaseURL
}
if mode == channel.RouteNative &&
(spec.Operation == execution.OperationListModels || spec.Operation == execution.OperationProbe) &&
if mode == channel.RouteNative && spec.Operation == execution.OperationListModels &&
!providerKindNativeForClient(providerKind, spec.ClientProtocol) {
failure := notSentUnaryFailure(execution.ErrorKindInvalidRequest, "native route does not match the client protocol")
return preparedAttempt{}, &failure
@@ -440,6 +439,32 @@ func (r *Runtime) prepare(spec execution.AttemptSpec, stream bool) (preparedAtte
failure := notSentUnaryFailure(execution.ErrorKindInvalidRequest, "invalid credential")
return preparedAttempt{}, &failure
}
if spec.Operation == execution.OperationProbe {
request := newProbeRequest(provider, providerKind, spec.UpstreamModel)
responses := spec.ClientProtocol == protocol.OpenAIResponses
typedURL, targetErr := convertedTypedTarget(
providerKind,
targetBaseURL,
spec.UpstreamModel,
responses,
false,
"",
)
if targetErr != nil {
failure := notSentUnaryFailure(execution.ErrorKindInvalidRequest, "invalid channel probe target")
return preparedAttempt{}, &failure
}
prepared := preparedAttempt{
provider: provider, mode: mode, typedURL: typedURL,
clientProtocol: spec.ClientProtocol, directKey: directKey, secrets: secrets,
}
if responses {
prepared.responsesRequest = request.ToResponsesRequest()
} else {
prepared.request = request
}
return prepared, nil
}
if mode == channel.RouteNative && providerSupportsPassthrough(providerKind, targetBaseURL) {
body, err := sanitizeNativeRequestBody(spec, stream)
if err != nil {
@@ -483,8 +508,7 @@ func (r *Runtime) prepare(spec execution.AttemptSpec, stream bool) (preparedAtte
secrets: secrets,
}, nil
}
if spec.ClientProtocol != protocol.OpenAICompletions ||
(spec.Operation != execution.OperationChatCompletion && spec.Operation != execution.OperationProbe) {
if spec.ClientProtocol != protocol.OpenAICompletions || spec.Operation != execution.OperationChatCompletion {
request, conversionErr := buildConvertedResponsesRequest(spec, provider)
if conversionErr != nil {
failure := notSentUnaryFailure(execution.ErrorKindInvalidRequest, conversionErr.Error())
@@ -538,6 +562,30 @@ func (r *Runtime) prepare(spec execution.AttemptSpec, stream bool) (preparedAtte
}, nil
}
func newProbeRequest(
provider schemas.ModelProvider,
providerKind channel.ProviderKind,
model string,
) *schemas.BifrostChatRequest {
content := "ping"
params := &schemas.ChatParameters{MaxCompletionTokens: schemas.Ptr(1)}
if providerKind == channel.ProviderOpenAICompatible {
params.MaxCompletionTokens = nil
params.ExtraParams = map[string]any{"max_tokens": 1}
}
return &schemas.BifrostChatRequest{
Provider: provider,
Model: model,
Input: []schemas.ChatMessage{{
Role: schemas.ChatMessageRoleUser,
Content: &schemas.ChatMessageContent{
ContentStr: &content,
},
}},
Params: params,
}
}
func providerSupportsPassthrough(providerKind channel.ProviderKind, baseURL string) bool {
switch providerKind {
case channel.ProviderOpenAI, channel.ProviderAnthropic, channel.ProviderGemini:
@@ -582,18 +630,13 @@ func supportedRequestShape(spec execution.AttemptSpec, stream bool) bool {
}
}
if spec.Operation == execution.OperationProbe {
if stream || spec.Method != http.MethodPost || strings.TrimSpace(spec.UpstreamModel) == "" {
if stream || strings.TrimSpace(spec.UpstreamModel) == "" || spec.Method != "" ||
spec.Path != "" || len(spec.Query) != 0 || spec.RawQuery != "" || len(spec.Body) != 0 {
return false
}
switch spec.ClientProtocol {
case protocol.OpenAICompletions:
return spec.Path == openAIChatPath
case protocol.OpenAIResponses:
return spec.Path == "/v1/responses"
case protocol.Anthropic:
return spec.Path == "/v1/messages"
case protocol.Gemini:
return validGeminiGeneratePath(spec.Path, "generateContent")
case protocol.OpenAICompletions, protocol.OpenAIResponses, protocol.Anthropic, protocol.Gemini:
return true
default:
return false
}
@@ -862,6 +905,10 @@ func (r *Runtime) newSDKContext(parent context.Context, spec execution.AttemptSp
bifrostContext.SetValue(schemas.BifrostContextKeyRequestID, spec.RequestID)
bifrostContext.SetValue(schemas.BifrostContextKeyDirectKey, directKey)
bifrostContext.SetValue(schemas.BifrostContextKeyLargeResponseThreshold, r.maxUnaryResponseBodyBytes)
if spec.Operation == execution.OperationProbe &&
channel.ProviderKind(spec.TargetKind) == channel.ProviderOpenAICompatible {
bifrostContext.SetValue(schemas.BifrostContextKeyPassthroughExtraParams, true)
}
if spec.Timeouts.StreamIdle > 0 {
bifrostContext.SetValue(schemas.BifrostContextKeyStreamIdleTimeout, spec.Timeouts.StreamIdle)
}
+3 -15
View File
@@ -53,7 +53,7 @@ func sanitizeNativeRequestBody(spec execution.AttemptSpec, stream bool) ([]byte,
case spec.Operation == execution.OperationListModels:
return nil, nil
case (spec.ClientProtocol == protocol.OpenAICompletions || spec.ClientProtocol == protocol.Anthropic) &&
(spec.Operation == execution.OperationChatCompletion || spec.Operation == execution.OperationProbe):
spec.Operation == execution.OperationChatCompletion:
body, err := sanitizeNativeChatBody(spec.Body, spec.UpstreamModel, stream)
if err != nil {
return nil, err
@@ -64,20 +64,8 @@ func sanitizeNativeRequestBody(spec execution.AttemptSpec, stream bool) ([]byte,
return nil, err
}
}
if spec.Operation != execution.OperationProbe {
return body, nil
}
object, err := decodeNativeJSONObject(body)
if err != nil {
return nil, err
}
if _, exists := object["stream"]; !exists {
object["stream"] = json.RawMessage("false")
return encodeNativeJSONObject(object)
}
return body, nil
case spec.ClientProtocol == protocol.Gemini &&
(spec.Operation == execution.OperationChatCompletion || spec.Operation == execution.OperationProbe):
case spec.ClientProtocol == protocol.Gemini && spec.Operation == execution.OperationChatCompletion:
object, err := decodeNativeJSONObject(spec.Body)
if err != nil {
// Preserve malformed/opaque provider payloads for the upstream to
@@ -90,7 +78,7 @@ func sanitizeNativeRequestBody(spec execution.AttemptSpec, stream bool) ([]byte,
}
return encodeNativeJSONObject(object)
case spec.ClientProtocol == protocol.OpenAIResponses &&
(spec.Operation == execution.OperationResponsesCreate || spec.Operation == execution.OperationProbe):
spec.Operation == execution.OperationResponsesCreate:
object, err := decodeNativeJSONObject(spec.Body)
if err != nil {
return nil, err
@@ -82,31 +82,43 @@ func TestOfficialNativeProbeUsesSelectedModelAndProtocolTarget(t *testing.T) {
name string
channelID channel.ID
protocol protocol.Protocol
path string
body []byte
runtime func(*testing.T, string) *testRuntime
response string
assert func(*testing.T, *http.Request, map[string]any)
}{
{name: "openai", channelID: channel.OpenAI, protocol: protocol.OpenAICompletions, path: "/v1/chat/completions", body: []byte(`{"model":"client","messages":[{"role":"user","content":"ping"}]}`), runtime: func(t *testing.T, base string) *testRuntime {
{name: "openai", channelID: channel.OpenAI, protocol: protocol.OpenAICompletions, runtime: func(t *testing.T, base string) *testRuntime {
return newProtocolTestRuntime(t, testRuntimeOptions{allowPrivateNetwork: true, openAIBaseURL: base})
}, assert: func(t *testing.T, request *http.Request, payload map[string]any) {
if request.URL.Path != "/v1/chat/completions" || payload["model"] != "probe-upstream" || payload["stream"] != false {
}, response: `{"id":"chat_1","object":"chat.completion","created":1,"model":"probe-upstream","choices":[{"index":0,"message":{"role":"assistant","content":"pong"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}`, assert: func(t *testing.T, request *http.Request, payload map[string]any) {
if request.URL.Path != "/v1/chat/completions" || payload["model"] != "probe-upstream" {
t.Errorf("OpenAI probe = %s %#v", request.URL.Path, payload)
}
}},
{name: "anthropic", channelID: channel.Anthropic, protocol: protocol.Anthropic, path: "/v1/messages", body: []byte(`{"model":"client","max_tokens":1,"messages":[{"role":"user","content":"ping"}]}`), runtime: func(t *testing.T, base string) *testRuntime {
return newProtocolTestRuntime(t, testRuntimeOptions{allowPrivateNetwork: true, anthropicBaseURL: base})
}, assert: func(t *testing.T, request *http.Request, payload map[string]any) {
if request.URL.Path != "/v1/messages" || payload["model"] != "probe-upstream" || payload["stream"] != false {
t.Errorf("Anthropic probe = %s %#v", request.URL.Path, payload)
if stream, exists := payload["stream"]; exists && stream != false {
t.Errorf("OpenAI probe stream = %#v", stream)
}
}},
{name: "gemini", channelID: channel.Gemini, protocol: protocol.Gemini, path: "/v1beta/models/client:generateContent", body: []byte(`{"contents":[{"parts":[{"text":"ping"}]}]}`), runtime: func(t *testing.T, base string) *testRuntime {
{name: "anthropic", channelID: channel.Anthropic, protocol: protocol.Anthropic, runtime: func(t *testing.T, base string) *testRuntime {
return newProtocolTestRuntime(t, testRuntimeOptions{allowPrivateNetwork: true, anthropicBaseURL: base})
}, response: `{"id":"msg_1","type":"message","role":"assistant","model":"probe-upstream","content":[{"type":"text","text":"pong"}],"stop_reason":"end_turn","usage":{"input_tokens":1,"output_tokens":1}}`, assert: func(t *testing.T, request *http.Request, payload map[string]any) {
if request.URL.Path != "/v1/messages" || payload["model"] != "probe-upstream" {
t.Errorf("Anthropic probe = %s %#v", request.URL.Path, payload)
}
if stream, exists := payload["stream"]; exists && stream != false {
t.Errorf("Anthropic probe stream = %#v", stream)
}
if payload["max_tokens"] != float64(1) {
t.Errorf("Anthropic max_tokens = %#v", payload["max_tokens"])
}
}},
{name: "gemini", channelID: channel.Gemini, protocol: protocol.Gemini, runtime: func(t *testing.T, base string) *testRuntime {
return newProtocolTestRuntime(t, testRuntimeOptions{allowPrivateNetwork: true, geminiBaseURL: base + "/v1beta"})
}, assert: func(t *testing.T, request *http.Request, payload map[string]any) {
}, response: `{"candidates":[{"content":{"role":"model","parts":[{"text":"pong"}]},"finishReason":"STOP"}],"usageMetadata":{"promptTokenCount":1,"candidatesTokenCount":1,"totalTokenCount":2},"modelVersion":"probe-upstream"}`, assert: func(t *testing.T, request *http.Request, payload map[string]any) {
if request.URL.Path != "/v1beta/models/probe-upstream:generateContent" {
t.Errorf("Gemini probe path = %s", request.URL.Path)
}
generationConfig, _ := payload["generationConfig"].(map[string]any)
if generationConfig["maxOutputTokens"] != float64(1) {
t.Errorf("Gemini generationConfig = %#v", generationConfig)
}
}},
}
for _, test := range tests {
@@ -121,12 +133,12 @@ func TestOfficialNativeProbeUsesSelectedModelAndProtocolTarget(t *testing.T) {
}
test.assert(t, request, payload)
writer.Header().Set("Content-Type", "application/json")
_, _ = io.WriteString(writer, `{"ok":true}`)
_, _ = io.WriteString(writer, test.response)
}))
defer server.Close()
runtime := test.runtime(t, server.URL)
spec := utilitySpec(test.channelID, test.protocol, execution.OperationProbe, http.MethodPost, test.path, test.body)
spec := utilitySpec(test.channelID, test.protocol, execution.OperationProbe, "", "", nil)
spec.ClientModel = "probe-client"
spec.UpstreamModel = "probe-upstream"
result := runtime.Execute(context.Background(), spec)
+35 -2
View File
@@ -303,6 +303,34 @@ func TestAttemptSpecRawQueryPreservesOpaqueBytesAndRejectsUnsafeShapes(t *testin
}
}
func TestProbeAttemptIsSemanticAndDoesNotCarryProviderWireShape(t *testing.T) {
spec := validAttemptSpec([]byte(`{"api_key":"sk-secret-value"}`))
spec.Operation = OperationProbe
spec.RequiredFeatures = FeatureSet{}
spec.Method = ""
spec.Path = ""
spec.Query = nil
spec.RawQuery = ""
spec.Body = nil
if err := spec.Validate(); err != nil {
t.Fatalf("semantic Probe Validate() error = %v", err)
}
for _, mutate := range []func(*AttemptSpec){
func(value *AttemptSpec) { value.Method = http.MethodPost },
func(value *AttemptSpec) { value.Path = "/v1/chat/completions" },
func(value *AttemptSpec) { value.RawQuery = "alt=sse" },
func(value *AttemptSpec) { value.Body = []byte(`{"model":"probe"}`) },
} {
invalid := spec.Clone()
mutate(&invalid)
if err := invalid.Validate(); err == nil {
t.Fatalf("Probe accepted provider wire fields: %#v", invalid)
}
}
}
func TestAttemptAndStreamResultsAreDefensivelyCopied(t *testing.T) {
reasoningBudget := int64(4096)
result := AttemptResult{
@@ -439,7 +467,6 @@ func TestValidationAcceptsValidContractsAndRejectsInvalidFields(t *testing.T) {
OperationResponsesInputItems,
OperationResponsesPassthrough,
OperationListModels,
OperationProbe,
} {
modelOptional := spec.Clone()
modelOptional.Operation = operation
@@ -457,10 +484,16 @@ func TestValidationAcceptsValidContractsAndRejectsInvalidFields(t *testing.T) {
if err := responsesCreate.Validate(); err == nil {
t.Fatal("expected responses create without a model to be rejected")
}
for _, operation := range []Operation{OperationResponsesCompact, OperationResponsesInputTokens} {
for _, operation := range []Operation{OperationResponsesCompact, OperationResponsesInputTokens, OperationProbe} {
modelRequired := spec.Clone()
modelRequired.Operation = operation
modelRequired.UpstreamModel = ""
if operation == OperationProbe {
modelRequired.Method = ""
modelRequired.Path = ""
modelRequired.Query = nil
modelRequired.Body = nil
}
if err := modelRequired.Validate(); err == nil {
t.Fatalf("expected %q without a model to be rejected", operation)
}
+22 -15
View File
@@ -77,22 +77,28 @@ func (s AttemptSpec) Validate() error {
return validationError("upstream_model", "must not be empty for this operation")
}
}
if !validHTTPToken(s.Method) {
return validationError("method", "must be a valid HTTP method")
}
if err := validateRequestPath(s.Path); err != nil {
return validationError("path", "must be an absolute request path without query or fragment")
}
if s.RawQuery != "" {
if len(s.Query) != 0 {
return validationError("raw_query", "must not be combined with query")
if s.Operation == OperationProbe {
if s.Method != "" || s.Path != "" || len(s.Query) != 0 || s.RawQuery != "" || len(s.Body) != 0 {
return validationError("probe", "must not contain provider wire fields")
}
if strings.Contains(s.RawQuery, "#") || containsControl(s.RawQuery) {
return validationError("raw_query", "contains a fragment or control characters")
} else {
if !validHTTPToken(s.Method) {
return validationError("method", "must be a valid HTTP method")
}
if err := validateRequestPath(s.Path); err != nil {
return validationError("path", "must be an absolute request path without query or fragment")
}
if s.RawQuery != "" {
if len(s.Query) != 0 {
return validationError("raw_query", "must not be combined with query")
}
if strings.Contains(s.RawQuery, "#") || containsControl(s.RawQuery) {
return validationError("raw_query", "contains a fragment or control characters")
}
}
if err := validateValues(s.Query); err != nil {
return validationError("query", "contains invalid control characters")
}
}
if err := validateValues(s.Query); err != nil {
return validationError("query", "contains invalid control characters")
}
if err := validateHeader(s.Header); err != nil {
return validationError("header", "contains an invalid name or value")
@@ -288,7 +294,8 @@ func operationRequiresModel(operation Operation) bool {
return operation == OperationChatCompletion ||
operation == OperationResponsesCreate ||
operation == OperationResponsesCompact ||
operation == OperationResponsesInputTokens
operation == OperationResponsesInputTokens ||
operation == OperationProbe
}
func validateRequestPath(path string) error {
+1
View File
@@ -390,6 +390,7 @@ func TestLoaderMapsSystemAndGroupRows(t *testing.T) {
Remove: []string{"X-System-Remove"},
},
InjectUsageOptions: true,
ValidationInterval: 10 * time.Minute,
RequestLogRetentionDays: 7,
ModelsDevAutoSyncEnabled: true,
}
+21 -1
View File
@@ -10,6 +10,7 @@ type Manager struct {
publishMu sync.Mutex
current atomic.Pointer[ConfigSnapshot]
reconciler SnapshotReconciler
updates chan struct{}
}
// SnapshotReconciler synchronizes infrastructure resources derived from a
@@ -19,7 +20,7 @@ type SnapshotReconciler interface {
}
func NewManager() *Manager {
return &Manager{}
return &Manager{updates: make(chan struct{})}
}
// SetSnapshotReconciler installs the process-owned runtime reconciler during
@@ -37,6 +38,21 @@ func (m *Manager) Current() *ConfigSnapshot {
return m.current.Load()
}
// CurrentWithUpdates returns a snapshot and a channel closed by the next
// successful publication. The pair is captured under the publication lock so
// consumers cannot miss a change between reading the snapshot and subscribing.
func (m *Manager) CurrentWithUpdates() (*ConfigSnapshot, <-chan struct{}) {
if m == nil {
return nil, nil
}
m.publishMu.Lock()
defer m.publishMu.Unlock()
if m.updates == nil {
m.updates = make(chan struct{})
}
return m.current.Load(), m.updates
}
// WithCurrentSnapshot runs a short callback while publication is blocked.
// Allowed callback work is limited to loading/reading the current snapshot,
// pure signature recomputation, and coordinated Registry/Stats recovery. It
@@ -99,5 +115,9 @@ func (m *Manager) publishCompiledLocked(next *ConfigSnapshot) *ConfigSnapshot {
next.Revision = current.Revision + 1
}
m.current.Store(next)
if m.updates != nil {
close(m.updates)
}
m.updates = make(chan struct{})
return next
}
+33
View File
@@ -84,6 +84,39 @@ func TestManagerPublishPreservesOldSnapshotReference(t *testing.T) {
}
}
func TestManagerCurrentWithUpdatesSignalsOnlySuccessfulPublication(t *testing.T) {
manager := NewManager()
current, updates := manager.CurrentWithUpdates()
if current != nil || updates == nil {
t.Fatalf("initial CurrentWithUpdates() = %p/%v", current, updates)
}
first, err := manager.Publish(managerCompileInput(1))
if err != nil {
t.Fatal(err)
}
select {
case <-updates:
default:
t.Fatal("successful publication did not signal update")
}
current, nextUpdates := manager.CurrentWithUpdates()
if current != first || nextUpdates == nil || nextUpdates == updates {
t.Fatalf("next CurrentWithUpdates() = %p/%v", current, nextUpdates)
}
invalid := managerCompileInput(2)
invalid.Groups[0].ChannelID = channel.ID("invalid")
if _, err := manager.Publish(invalid); err == nil {
t.Fatal("invalid Publish() error = nil")
}
select {
case <-nextUpdates:
t.Fatal("failed publication signaled an update")
default:
}
}
func TestManagerPublishFailureKeepsCurrentSnapshot(t *testing.T) {
manager := NewManager()
first, err := manager.Publish(managerCompileInput(1))
+12 -1
View File
@@ -19,6 +19,7 @@ const (
SettingStreamIdleTimeout = "stream_idle_timeout"
SettingHeaderRules = "header_rules"
SettingInjectUsageOptions = "inject_usage_options"
SettingValidationInterval = "validation_interval"
SettingRequestLogRetentionDays = "request_log_retention_days"
SettingModelsDevAutoSyncEnabled = "models_dev_auto_sync_enabled"
)
@@ -35,6 +36,7 @@ type RuntimeSettings struct {
StreamIdleTimeout time.Duration
HeaderRules HeaderRules
InjectUsageOptions bool
ValidationInterval time.Duration
RequestLogRetentionDays int
ModelsDevAutoSyncEnabled bool
}
@@ -52,6 +54,7 @@ func DefaultRuntimeSettings() RuntimeSettings {
StreamIdleTimeout: 300 * time.Second,
HeaderRules: HeaderRules{Set: map[string]string{}},
InjectUsageOptions: true,
ValidationInterval: 10 * time.Minute,
RequestLogRetentionDays: defaultRequestLogRetentionDays,
ModelsDevAutoSyncEnabled: true,
}
@@ -64,6 +67,7 @@ func IsRuntimeSettingKey(key string) bool {
SettingStreamIdleTimeout,
SettingHeaderRules,
SettingInjectUsageOptions,
SettingValidationInterval,
SettingRequestLogRetentionDays,
SettingModelsDevAutoSyncEnabled:
return true
@@ -106,6 +110,12 @@ func ResolveRuntimeSettings(settings config.Settings) (RuntimeSettings, error) {
return RuntimeSettings{}, err
}
resolved.InjectUsageOptions = value
case SettingValidationInterval:
seconds, err := positiveWholeSeconds(key, value)
if err != nil {
return RuntimeSettings{}, err
}
resolved.ValidationInterval = time.Duration(seconds) * time.Second
case SettingRequestLogRetentionDays:
days, err := wholeNumberInRange(
key,
@@ -186,7 +196,8 @@ func ValidateRuntimeSetting(key string, value any) error {
switch key {
case SettingFirstByteTimeout,
SettingRequestTimeout,
SettingStreamIdleTimeout:
SettingStreamIdleTimeout,
SettingValidationInterval:
_, err := positiveWholeSeconds(key, value)
return err
case SettingHeaderRules:
+36 -3
View File
@@ -22,6 +22,7 @@ func TestCompilePublishesDefaultRuntimeSettingsWithoutGroups(t *testing.T) {
StreamIdleTimeout: 300 * time.Second,
HeaderRules: HeaderRules{Set: map[string]string{}},
InjectUsageOptions: true,
ValidationInterval: 10 * time.Minute,
RequestLogRetentionDays: 7,
ModelsDevAutoSyncEnabled: true,
}
@@ -59,6 +60,35 @@ func TestModelsDevAutoSyncSettingDefaultsTrueAndIsSystemOnly(t *testing.T) {
}
}
func TestValidationIntervalDefaultsToTenMinutesAndIsSystemOnly(t *testing.T) {
defaults, err := ResolveRuntimeSettings(nil)
if err != nil {
t.Fatal(err)
}
if defaults.ValidationInterval != 10*time.Minute {
t.Fatalf("ValidationInterval = %v, want 10m", defaults.ValidationInterval)
}
overridden, err := ResolveRuntimeSettings(config.Settings{
SettingValidationInterval: json.Number("900"),
})
if err != nil {
t.Fatal(err)
}
if overridden.ValidationInterval != 15*time.Minute {
t.Fatalf("ValidationInterval = %v, want 15m", overridden.ValidationInterval)
}
if !IsRuntimeSettingKey(SettingValidationInterval) {
t.Fatal("validation_interval is not a public runtime setting")
}
if _, err := ResolveGroupRuntimeSettings(
defaults,
config.Settings{SettingValidationInterval: json.Number("900")},
); err == nil {
t.Fatal("Group override accepted system-only validation_interval")
}
}
func TestCompileRejectsInvalidGlobalSettingsWithoutGroups(t *testing.T) {
for _, input := range []config.Settings{
{"request_log_retention_days": json.Number("0")},
@@ -222,9 +252,10 @@ func TestParseHeaderRulesRejectsSDKOwnedCredentialHeaders(t *testing.T) {
func TestResolveRuntimeSettingsAppliesSystemOverrides(t *testing.T) {
got, err := ResolveRuntimeSettings(config.Settings{
SettingFirstByteTimeout: json.Number("180"),
SettingRequestTimeout: json.Number("900"),
SettingStreamIdleTimeout: json.Number("45"),
SettingFirstByteTimeout: json.Number("180"),
SettingRequestTimeout: json.Number("900"),
SettingStreamIdleTimeout: json.Number("45"),
SettingValidationInterval: json.Number("900"),
SettingHeaderRules: map[string]any{
"set": map[string]any{"x-test": "value"},
"remove": []any{"x-old"},
@@ -237,6 +268,7 @@ func TestResolveRuntimeSettingsAppliesSystemOverrides(t *testing.T) {
if got.FirstByteTimeout != 180*time.Second ||
got.RequestTimeout != 900*time.Second ||
got.StreamIdleTimeout != 45*time.Second ||
got.ValidationInterval != 15*time.Minute ||
got.RequestLogRetentionDays != 30 {
t.Fatalf("settings = %#v", got)
}
@@ -308,6 +340,7 @@ func TestIsRuntimeSettingKeyRecognizesOnlyPublicRuntimeKeys(t *testing.T) {
SettingStreamIdleTimeout,
SettingHeaderRules,
SettingInjectUsageOptions,
SettingValidationInterval,
SettingRequestLogRetentionDays,
} {
if !IsRuntimeSettingKey(key) {
+4
View File
@@ -21,6 +21,7 @@ export const runtimeSettingKeys = [
'stream_idle_timeout',
'header_rules',
'inject_usage_options',
'validation_interval',
'request_log_retention_days',
'models_dev_auto_sync_enabled',
] as const
@@ -40,6 +41,7 @@ export interface SettingsValues {
stream_idle_timeout: number
header_rules: HeaderRulesDto
inject_usage_options: boolean
validation_interval: number
request_log_retention_days: number
models_dev_auto_sync_enabled: boolean
}
@@ -56,6 +58,7 @@ export type SettingsPatch = Partial<{
stream_idle_timeout: number | null
header_rules: HeaderRulesDto | null
inject_usage_options: boolean | null
validation_interval: number | null
request_log_retention_days: number | null
models_dev_auto_sync_enabled: boolean | null
}>
@@ -121,6 +124,7 @@ export function projectSettings(value: unknown): SettingsDto {
stream_idle_timeout: projectSafeInteger(values.stream_idle_timeout, { minimum: 1 }),
header_rules: projectHeaderRules(values.header_rules),
inject_usage_options: projectBoolean(values.inject_usage_options),
validation_interval: projectSafeInteger(values.validation_interval, { minimum: 1 }),
request_log_retention_days: projectSafeInteger(values.request_log_retention_days, {
minimum: 1,
maximum: 365,
@@ -38,6 +38,7 @@ const timeoutKeys: TimeoutSettingKey[] = [
'first_byte_timeout',
'request_timeout',
'stream_idle_timeout',
'validation_interval',
]
function cloneDraft(): SettingsDraft {
return createSettingsDraft({
+6 -1
View File
@@ -104,7 +104,12 @@ const valid = computed(
controllerValid.value &&
(!draft.value?.overrides.has('header_rules') || headerRulesValid.value),
)
const timeoutKeys = ['first_byte_timeout', 'request_timeout', 'stream_idle_timeout'] as const
const timeoutKeys = [
'first_byte_timeout',
'request_timeout',
'stream_idle_timeout',
'validation_interval',
] as const
const changedKeys = computed(() => {
const changed = runtimeSettingKeys.filter((key) =>
Object.prototype.hasOwnProperty.call(patch.value, key),
@@ -29,6 +29,7 @@ const requestForwardingKeys: RuntimeSettingKey[] = [
'stream_idle_timeout',
'header_rules',
'inject_usage_options',
'validation_interval',
]
const logsMaintenanceKeys: RuntimeSettingKey[] = ['request_log_retention_days']
const modelPriceKeys: RuntimeSettingKey[] = ['models_dev_auto_sync_enabled']
@@ -273,6 +274,7 @@ export function validateSettingsSection(draft: SettingsDraft, section: SettingsS
'first_byte_timeout',
'request_timeout',
'stream_idle_timeout',
'validation_interval',
]
return timeouts.every((key) => !draft.overrides.has(key) || isValidTimeout(draft.values[key]))
}
+2 -1
View File
@@ -58,10 +58,11 @@ export default {
runtime: {
title: 'Request and forwarding',
description:
'Explicit values override built-in defaults. Restoring a default uses the current version; all timeouts are in seconds.',
'Explicit values override built-in defaults. Restoring a default uses the current version; all time values are in seconds.',
first_byte_timeout: 'Native response / stream first-event timeout',
request_timeout: 'Total request timeout',
stream_idle_timeout: 'Stream idle timeout',
validation_interval: 'Credential validation interval',
inject_usage_options: 'Inject usage options for streaming responses',
injectUsageHelp:
'Applies only to capable openai-completions streaming requests so the final usage can be returned.',
+2 -1
View File
@@ -58,10 +58,11 @@ export default {
runtime: {
title: 'リクエストと転送',
description:
'明示値は組み込み既定値を上書きします。既定値の復元後は現在のバージョンが有効値を決めます。すべての時間単位は秒です。',
'明示値は組み込み既定値を上書きします。既定値の復元後は現在のバージョンが有効値を決めます。すべての時間設定は秒単位です。',
first_byte_timeout: 'ネイティブ応答 / ストリーム初回イベントのタイムアウト',
request_timeout: 'リクエスト全体のタイムアウト',
stream_idle_timeout: 'ストリームアイドルタイムアウト',
validation_interval: '認証情報の検証間隔',
inject_usage_options: 'ストリーミング応答へ usage オプションを注入',
injectUsageHelp:
'最終 usage を取得するため、この機能をサポートする openai-completions のストリーミング要求でのみ有効です。',
+2 -1
View File
@@ -53,10 +53,11 @@ export default {
},
runtime: {
title: '请求与转发',
description: '显式值覆盖内置默认;恢复默认后由当前版本决定有效值。所有时间单位均为秒。',
description: '显式值覆盖内置默认;恢复默认后由当前版本决定有效值。所有时间设置均以秒为单位。',
first_byte_timeout: '原生响应 / 流式首事件超时',
request_timeout: '请求总时长',
stream_idle_timeout: '流空闲超时',
validation_interval: '检测间隔',
inject_usage_options: '为流式响应注入 usage 选项',
injectUsageHelp: '只在支持该能力的 openai-completions 流式请求中生效,用于获取最终 usage。',
models_dev_auto_sync_enabled: '自动同步 Models.dev 目录与价格',