diff --git a/internal/control/discover_executor.go b/internal/control/discover_executor.go index 4893a75d..c78b9cf5 100644 --- a/internal/control/discover_executor.go +++ b/internal/control/discover_executor.go @@ -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 } diff --git a/internal/control/discover_executor_test.go b/internal/control/discover_executor_test.go index 54b419e2..19023a95 100644 --- a/internal/control/discover_executor_test.go +++ b/internal/control/discover_executor_test.go @@ -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") } }) } diff --git a/internal/control/runtime.go b/internal/control/runtime.go index d866a755..e9c289bd 100644 --- a/internal/control/runtime.go +++ b/internal/control/runtime.go @@ -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 { diff --git a/internal/control/runtime_test.go b/internal/control/runtime_test.go index 12422b4e..2db474b1 100644 --- a/internal/control/runtime_test.go +++ b/internal/control/runtime_test.go @@ -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) diff --git a/internal/control/server_test.go b/internal/control/server_test.go index 667f3d34..becd1683 100644 --- a/internal/control/server_test.go +++ b/internal/control/server_test.go @@ -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) } diff --git a/internal/control/settings.go b/internal/control/settings.go index c5fd616b..05009392 100644 --- a/internal/control/settings.go +++ b/internal/control/settings.go @@ -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, }, diff --git a/internal/control/settings_etag_test.go b/internal/control/settings_etag_test.go index c4c628cc..af979e87 100644 --- a/internal/control/settings_etag_test.go +++ b/internal/control/settings_etag_test.go @@ -37,6 +37,7 @@ func TestSettingsWireETagMatchesCheckedInFixture(t *testing.T) { Remove: []string{}, }, InjectUsageOptions: true, + ValidationInterval: 600, RequestLogRetentionDays: 7, }, Overrides: []string{}, diff --git a/internal/control/settings_occ_test.go b/internal/control/settings_occ_test.go index ddff3215..00ed0136 100644 --- a/internal/control/settings_occ_test.go +++ b/internal/control/settings_occ_test.go @@ -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) } diff --git a/internal/control/settings_test.go b/internal/control/settings_test.go index 8a456690..2877b0ad 100644 --- a/internal/control/settings_test.go +++ b/internal/control/settings_test.go @@ -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 diff --git a/internal/control/testdata/settings-etag-v1.json b/internal/control/testdata/settings-etag-v1.json index 01b28449..4bbc43c0 100644 --- a/internal/control/testdata/settings-etag-v1.json +++ b/internal/control/testdata/settings-etag-v1.json @@ -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" } diff --git a/internal/control/validation.go b/internal/control/validation.go index eb2dfdd2..428c79d7 100644 --- a/internal/control/validation.go +++ b/internal/control/validation.go @@ -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( diff --git a/internal/control/validation_test.go b/internal/control/validation_test.go index e5c11433..306fa123 100644 --- a/internal/control/validation_test.go +++ b/internal/control/validation_test.go @@ -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) { diff --git a/internal/execution/bifrost/compatible_matrix_test.go b/internal/execution/bifrost/compatible_matrix_test.go index 8aff53f3..4630efce 100644 --- a/internal/execution/bifrost/compatible_matrix_test.go +++ b/internal/execution/bifrost/compatible_matrix_test.go @@ -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" diff --git a/internal/execution/bifrost/converted.go b/internal/execution/bifrost/converted.go index f28eeae1..64243330 100644 --- a/internal/execution/bifrost/converted.go +++ b/internal/execution/bifrost/converted.go @@ -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 diff --git a/internal/execution/bifrost/executor.go b/internal/execution/bifrost/executor.go index 1f1df1bc..7b14e4a6 100644 --- a/internal/execution/bifrost/executor.go +++ b/internal/execution/bifrost/executor.go @@ -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) } diff --git a/internal/execution/bifrost/passthrough.go b/internal/execution/bifrost/passthrough.go index 96e7b072..a6ddaf30 100644 --- a/internal/execution/bifrost/passthrough.go +++ b/internal/execution/bifrost/passthrough.go @@ -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 diff --git a/internal/execution/bifrost/utility_operations_test.go b/internal/execution/bifrost/utility_operations_test.go index 23ee68c8..595a349d 100644 --- a/internal/execution/bifrost/utility_operations_test.go +++ b/internal/execution/bifrost/utility_operations_test.go @@ -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) diff --git a/internal/execution/contracts_test.go b/internal/execution/contracts_test.go index 64b4724f..593a4754 100644 --- a/internal/execution/contracts_test.go +++ b/internal/execution/contracts_test.go @@ -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) } diff --git a/internal/execution/validation.go b/internal/execution/validation.go index 09755df4..ce501ed0 100644 --- a/internal/execution/validation.go +++ b/internal/execution/validation.go @@ -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 { diff --git a/internal/state/loader/loader_test.go b/internal/state/loader/loader_test.go index 76b606b2..a283c4e5 100644 --- a/internal/state/loader/loader_test.go +++ b/internal/state/loader/loader_test.go @@ -390,6 +390,7 @@ func TestLoaderMapsSystemAndGroupRows(t *testing.T) { Remove: []string{"X-System-Remove"}, }, InjectUsageOptions: true, + ValidationInterval: 10 * time.Minute, RequestLogRetentionDays: 7, ModelsDevAutoSyncEnabled: true, } diff --git a/internal/state/manager.go b/internal/state/manager.go index 7fc14d87..5cd87e47 100644 --- a/internal/state/manager.go +++ b/internal/state/manager.go @@ -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 } diff --git a/internal/state/manager_test.go b/internal/state/manager_test.go index 933b49d1..6d598471 100644 --- a/internal/state/manager_test.go +++ b/internal/state/manager_test.go @@ -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)) diff --git a/internal/state/runtime_settings.go b/internal/state/runtime_settings.go index 62491174..3b3f8518 100644 --- a/internal/state/runtime_settings.go +++ b/internal/state/runtime_settings.go @@ -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: diff --git a/internal/state/runtime_settings_test.go b/internal/state/runtime_settings_test.go index 4ea6ff02..540a994e 100644 --- a/internal/state/runtime_settings_test.go +++ b/internal/state/runtime_settings_test.go @@ -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) { diff --git a/web/src/app/resources/settings.ts b/web/src/app/resources/settings.ts index 44e71dde..62d81783 100644 --- a/web/src/app/resources/settings.ts +++ b/web/src/app/resources/settings.ts @@ -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, diff --git a/web/src/features/settings/RuntimeSettingsSection.vue b/web/src/features/settings/RuntimeSettingsSection.vue index 132a08fe..fbe4889e 100644 --- a/web/src/features/settings/RuntimeSettingsSection.vue +++ b/web/src/features/settings/RuntimeSettingsSection.vue @@ -38,6 +38,7 @@ const timeoutKeys: TimeoutSettingKey[] = [ 'first_byte_timeout', 'request_timeout', 'stream_idle_timeout', + 'validation_interval', ] function cloneDraft(): SettingsDraft { return createSettingsDraft({ diff --git a/web/src/features/settings/SettingsView.vue b/web/src/features/settings/SettingsView.vue index 65310ebf..d503ad9b 100644 --- a/web/src/features/settings/SettingsView.vue +++ b/web/src/features/settings/SettingsView.vue @@ -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), diff --git a/web/src/features/settings/settings-patch.ts b/web/src/features/settings/settings-patch.ts index 10c14855..abb29231 100644 --- a/web/src/features/settings/settings-patch.ts +++ b/web/src/features/settings/settings-patch.ts @@ -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])) } diff --git a/web/src/i18n/locales/en-US/settings.ts b/web/src/i18n/locales/en-US/settings.ts index 177a437c..239d2f8c 100644 --- a/web/src/i18n/locales/en-US/settings.ts +++ b/web/src/i18n/locales/en-US/settings.ts @@ -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.', diff --git a/web/src/i18n/locales/ja-JP/settings.ts b/web/src/i18n/locales/ja-JP/settings.ts index 7cd8ef22..31447ec1 100644 --- a/web/src/i18n/locales/ja-JP/settings.ts +++ b/web/src/i18n/locales/ja-JP/settings.ts @@ -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 のストリーミング要求でのみ有効です。', diff --git a/web/src/i18n/locales/zh-CN/settings.ts b/web/src/i18n/locales/zh-CN/settings.ts index 0f10ecfc..0defd24e 100644 --- a/web/src/i18n/locales/zh-CN/settings.ts +++ b/web/src/i18n/locales/zh-CN/settings.ts @@ -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 目录与价格',