From e9daf96865d88f7a0b14aa901a67fde900528a86 Mon Sep 17 00:00:00 2001 From: tbphp Date: Thu, 13 Aug 2026 23:33:06 +0800 Subject: [PATCH] =?UTF-8?q?feat(channel):=20=E6=94=AF=E6=8C=81=20DeepSeek?= =?UTF-8?q?=20=E4=B8=89=E7=A7=8D=E5=8E=9F=E7=94=9F=E5=8D=8F=E8=AE=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../channel/final_provider_contract_test.go | 17 +- internal/channel/presets.go | 42 ++- internal/channel/registry_test.go | 13 +- internal/execution/bifrost/converted.go | 56 +++- .../bifrost/converted_protocols_test.go | 27 +- internal/execution/bifrost/executor.go | 21 +- internal/execution/bifrost/runtime_manager.go | 79 ++++- .../execution/bifrost/runtime_manager_test.go | 275 +++++++++++++++++- 8 files changed, 513 insertions(+), 17 deletions(-) diff --git a/internal/channel/final_provider_contract_test.go b/internal/channel/final_provider_contract_test.go index bd165137..e23b68db 100644 --- a/internal/channel/final_provider_contract_test.go +++ b/internal/channel/final_provider_contract_test.go @@ -45,12 +45,13 @@ func TestNativeProviderChannelsResolveWithoutCompatibleFallback(t *testing.T) { channelID ID providerKind ProviderKind responsesMode RouteMode + anthropicMode RouteMode catalogProviderID string }{ - {channelID: DeepSeek, providerKind: ProviderKind("deepseek"), responsesMode: RouteConverted, catalogProviderID: "deepseek"}, - {channelID: OpenRouter, providerKind: ProviderKind("openrouter"), responsesMode: RouteNative, catalogProviderID: "openrouter"}, - {channelID: Groq, providerKind: ProviderKind("groq"), responsesMode: RouteConverted, catalogProviderID: "groq"}, - {channelID: XAI, providerKind: ProviderKind("xai"), responsesMode: RouteNative, catalogProviderID: "xai"}, + {channelID: DeepSeek, providerKind: ProviderKind("deepseek"), responsesMode: RouteNative, anthropicMode: RouteNative, catalogProviderID: "deepseek"}, + {channelID: OpenRouter, providerKind: ProviderKind("openrouter"), responsesMode: RouteNative, anthropicMode: RouteConverted, catalogProviderID: "openrouter"}, + {channelID: Groq, providerKind: ProviderKind("groq"), responsesMode: RouteConverted, anthropicMode: RouteConverted, catalogProviderID: "groq"}, + {channelID: XAI, providerKind: ProviderKind("xai"), responsesMode: RouteNative, anthropicMode: RouteConverted, catalogProviderID: "xai"}, } for _, test := range tests { t.Run(string(test.channelID), func(t *testing.T) { @@ -70,6 +71,14 @@ func TestNativeProviderChannelsResolveWithoutCompatibleFallback(t *testing.T) { if mode, ok := target.Mode(protocol.OpenAIResponses, execution.OperationResponsesCreate); !ok || mode != test.responsesMode { t.Fatalf("Responses mode = %q, %t, want %q", mode, ok, test.responsesMode) } + if mode, ok := target.Mode(protocol.Anthropic, execution.OperationChatCompletion); !ok || mode != test.anthropicMode { + t.Fatalf("Anthropic mode = %q, %t, want %q", mode, ok, test.anthropicMode) + } + if test.channelID == DeepSeek { + if mode, ok := target.Mode(protocol.Anthropic, execution.OperationListModels); !ok || mode != RouteConverted { + t.Fatalf("DeepSeek Anthropic model-list mode = %q, %t", mode, ok) + } + } if _, ok := target.Mode(protocol.OpenAIResponses, execution.OperationResponsesRetrieve); ok { t.Fatal("target unexpectedly supports Responses lifecycle") } diff --git a/internal/channel/presets.go b/internal/channel/presets.go index e56f4b28..1cca2503 100644 --- a/internal/channel/presets.go +++ b/internal/channel/presets.go @@ -104,9 +104,9 @@ func builtinDefinitions() []definition { azureDefinition, bedrockDefinition, vertexDefinition, - newNativeOpenAIChannelDefinition( + newDeepSeekDefinition( DeepSeek, "DeepSeek", "DS", "deepseek", []string{"deep seek"}, - "deepseek", ProviderDeepSeek, apiKeySchema, false, + "deepseek", ProviderDeepSeek, apiKeySchema, ), newFixedCompatibleDefinition( MoonshotAI, "Moonshot AI", "MS", "moonshot", []string{"kimi", "moonshot"}, @@ -330,6 +330,44 @@ func newNativeOpenAIChannelDefinition( ) } +func newDeepSeekDefinition( + id ID, + name string, + mark string, + icon string, + searchTerms []string, + catalogProviderID string, + providerKind ProviderKind, + credentials objectSchema, +) definition { + deepSeek := newDefinition( + id, + name, + mark, + icon, + "Managed API preset", + searchTerms, + objectSchema{{ + descriptor: FieldDescriptor{Key: "base_url", Label: "Base URL", InputKind: InputURL}, + normalize: normalizeBaseURL, + }}, + credentials, + catalogProviderID, + providerKind, + false, + map[protocol.Protocol]bool{ + protocol.OpenAICompletions: true, + protocol.OpenAIResponses: true, + protocol.Anthropic: true, + }, + false, + ) + // DeepSeek's Anthropic-compatible surface exposes Messages, while its model + // catalog remains OpenAI-shaped and is converted for Anthropic clients. + deepSeek.modes[protocol.Anthropic][execution.OperationListModels] = RouteConverted + return deepSeek +} + func newDefinition( id ID, name string, diff --git a/internal/channel/registry_test.go b/internal/channel/registry_test.go index 57fdbee1..528b0fa7 100644 --- a/internal/channel/registry_test.go +++ b/internal/channel/registry_test.go @@ -501,8 +501,17 @@ func TestCuratedChannelResolvesCodeOwnedTargetAndCatalogMapping(t *testing.T) { if mode, ok := target.Mode(protocol.OpenAICompletions, execution.OperationChatCompletion); !ok || mode != RouteNative { t.Fatalf("DeepSeek native completions mode = %q, %t", mode, ok) } - if mode, ok := target.Mode(protocol.Anthropic, execution.OperationChatCompletion); !ok || mode != RouteConverted { - t.Fatalf("DeepSeek converted Anthropic mode = %q, %t", mode, ok) + if mode, ok := target.Mode(protocol.OpenAIResponses, execution.OperationResponsesCreate); !ok || mode != RouteNative { + t.Fatalf("DeepSeek native Responses mode = %q, %t", mode, ok) + } + if mode, ok := target.Mode(protocol.Anthropic, execution.OperationChatCompletion); !ok || mode != RouteNative { + t.Fatalf("DeepSeek native Anthropic mode = %q, %t", mode, ok) + } + if mode, ok := target.Mode(protocol.Anthropic, execution.OperationListModels); !ok || mode != RouteConverted { + t.Fatalf("DeepSeek Anthropic model-list mode = %q, %t", mode, ok) + } + if mode, ok := target.Mode(protocol.Gemini, execution.OperationChatCompletion); !ok || mode != RouteConverted { + t.Fatalf("DeepSeek converted Gemini mode = %q, %t", mode, ok) } descriptor, ok := registry.Get(DeepSeek) if !ok || len(descriptor.ParamFields) != 1 || len(descriptor.CredentialFields) != 1 { diff --git a/internal/execution/bifrost/converted.go b/internal/execution/bifrost/converted.go index 4ae64424..80d30841 100644 --- a/internal/execution/bifrost/converted.go +++ b/internal/execution/bifrost/converted.go @@ -30,6 +30,7 @@ type responsesStreamSDKResult struct { } type targetConversionError struct { + code string summary string } @@ -40,14 +41,24 @@ func (err *targetConversionError) Error() string { return err.summary } -func (*targetConversionError) ConversionCode() string { - return execution.ErrorCodeTargetConversionNotSupported +func (err *targetConversionError) ConversionCode() string { + if err == nil || err.code == "" { + return execution.ErrorCodeTargetConversionNotSupported + } + return err.code } func unsupportedTargetConversion(summary string) error { return &targetConversionError{summary: summary} } +func criticalSemanticLoss(summary string) error { + return &targetConversionError{ + code: execution.ErrorCodeCriticalSemanticLoss, + summary: summary, + } +} + func buildConvertedResponsesRequest(spec execution.AttemptSpec, provider schemas.ModelProvider) (*schemas.BifrostResponsesRequest, error) { conversionContext := schemas.NewBifrostContext(context.Background(), schemas.NoDeadline) defer conversionContext.Cancel() @@ -71,7 +82,11 @@ func buildConvertedResponsesRequest(spec execution.AttemptSpec, provider schemas if err := json.Unmarshal(spec.Body, &wire); err != nil { return nil, fmt.Errorf("invalid Anthropic request body") } + if provider == schemas.DeepSeek && wire.Container != nil { + return nil, criticalSemanticLoss("DeepSeek Anthropic route cannot preserve container semantics") + } request = wire.ToBifrostResponsesRequest(conversionContext) + preserveDeepSeekAnthropicEffort(request, provider, wire.OutputConfig) case protocol.Gemini: if spec.Operation != execution.OperationChatCompletion { return nil, unsupportedTargetConversion("converted Gemini operation is not supported") @@ -96,6 +111,25 @@ func buildConvertedResponsesRequest(spec execution.AttemptSpec, provider schemas return request, nil } +func preserveDeepSeekAnthropicEffort( + request *schemas.BifrostResponsesRequest, + provider schemas.ModelProvider, + outputConfig *anthropic.AnthropicOutputConfig, +) { + if request == nil || provider != schemas.DeepSeek || outputConfig == nil || outputConfig.Effort == nil { + return + } + if request.Params == nil { + request.Params = &schemas.ResponsesParameters{} + } + if request.Params.ExtraParams == nil { + request.Params.ExtraParams = make(map[string]any) + } + request.Params.ExtraParams["output_config"] = map[string]any{ + "effort": *outputConfig.Effort, + } +} + func stripResponsesControlParams(params *schemas.ResponsesParameters) { if params == nil || len(params.ExtraParams) == 0 { return @@ -306,6 +340,24 @@ func convertedTypedTarget( return appendTypedQuery(resourcePath, rawQuery), nil } +func deepSeekNativeTypedTarget(baseURL string, clientProtocol protocol.Protocol, rawQuery string) (string, error) { + resourcePath := "" + switch clientProtocol { + case protocol.OpenAICompletions: + resourcePath = "/chat/completions" + case protocol.OpenAIResponses: + resourcePath = "/responses" + case protocol.Anthropic: + resourcePath = "/anthropic/v1/messages" + default: + return "", fmt.Errorf("unsupported DeepSeek native protocol") + } + if baseURL != "" { + return resolveTypedTargetURL(baseURL, resourcePath, rawQuery) + } + return appendTypedQuery(resourcePath, rawQuery), nil +} + func convertedListModelsTarget(providerKind channel.ProviderKind, baseURL string, rawQuery string) (string, error) { resourcePath := "/models" switch providerKind { diff --git a/internal/execution/bifrost/converted_protocols_test.go b/internal/execution/bifrost/converted_protocols_test.go index 042484ec..90a7d356 100644 --- a/internal/execution/bifrost/converted_protocols_test.go +++ b/internal/execution/bifrost/converted_protocols_test.go @@ -335,7 +335,7 @@ func TestNativeOpenAIStyleTypedTargetsPreserveSafeQuery(t *testing.T) { wantTarget string }{ {name: "deepseek chat", provider: channel.ProviderDeepSeek, wantTarget: "/chat/completions?" + rawQuery}, - {name: "deepseek responses stream", provider: channel.ProviderDeepSeek, responses: true, stream: true, wantTarget: "/chat/completions?" + rawQuery}, + {name: "deepseek converted responses stream", provider: channel.ProviderDeepSeek, responses: true, stream: true, wantTarget: "/chat/completions?" + rawQuery}, {name: "openrouter chat stream", provider: channel.ProviderOpenRouter, stream: true, wantTarget: "/v1/chat/completions?" + rawQuery}, {name: "openrouter responses", provider: channel.ProviderOpenRouter, responses: true, wantTarget: "/v1/responses?" + rawQuery}, {name: "groq chat", provider: channel.ProviderGroq, wantTarget: "/v1/chat/completions?" + rawQuery}, @@ -371,6 +371,31 @@ func TestNativeOpenAIStyleTypedTargetsPreserveSafeQuery(t *testing.T) { } } +func TestDeepSeekNativeTypedTargetsPreserveSafeQuery(t *testing.T) { + const rawQuery = "trace=%2F&cursor=next" + for _, test := range []struct { + clientProtocol protocol.Protocol + wantTarget string + }{ + {clientProtocol: protocol.OpenAICompletions, wantTarget: "/chat/completions?" + rawQuery}, + {clientProtocol: protocol.OpenAIResponses, wantTarget: "/responses?" + rawQuery}, + {clientProtocol: protocol.Anthropic, wantTarget: "/anthropic/v1/messages?" + rawQuery}, + } { + t.Run(string(test.clientProtocol), func(t *testing.T) { + target, err := deepSeekNativeTypedTarget("", test.clientProtocol, rawQuery) + if err != nil { + t.Fatal(err) + } + if target != test.wantTarget { + t.Fatalf("target = %q, want %q", target, test.wantTarget) + } + }) + } + if _, err := deepSeekNativeTypedTarget("", protocol.Gemini, rawQuery); err == nil { + t.Fatal("Gemini native target error = nil") + } +} + func TestNativeOpenAIStyleListModelsTargetsPreserveSafeQuery(t *testing.T) { const rawQuery = "cursor=%2F&limit=10" for _, test := range []struct { diff --git a/internal/execution/bifrost/executor.go b/internal/execution/bifrost/executor.go index 52212826..9c6afda1 100644 --- a/internal/execution/bifrost/executor.go +++ b/internal/execution/bifrost/executor.go @@ -323,6 +323,11 @@ func (r *Runtime) ExecuteStream( func preparedUpstreamAPI(prepared preparedAttempt, spec execution.AttemptSpec) execution.UpstreamAPI { providerKind := channel.ProviderKind(spec.TargetKind) + if providerKind == channel.ProviderDeepSeek && + spec.ClientProtocol == protocol.Anthropic && + spec.Operation != execution.OperationListModels { + return execution.UpstreamAPIAnthropicMessages + } switch providerKind { case channel.ProviderAzureOpenAI: return execution.UpstreamAPIAzureOpenAI @@ -397,7 +402,7 @@ func (r *Runtime) prepare(spec execution.AttemptSpec, stream bool) (preparedAtte failure := notSentUnaryFailure(execution.ErrorKindInternal, "provider runtime config is unavailable") return preparedAttempt{}, &failure } - effective, effectiveErr := buildEffectiveProviderConfig(resolved, r.allowPrivate) + effective, effectiveErr := buildEffectiveProviderConfigForAttempt(resolved, spec, r.allowPrivate) fixedFingerprint := r.fixedConfig.fingerprint fixedCanonical := r.fixedConfig.canonical if r.fixedConfig.baseFingerprint != "" { @@ -475,6 +480,9 @@ func (r *Runtime) prepare(spec execution.AttemptSpec, stream bool) (preparedAtte failure := notSentUnaryFailure(execution.ErrorKindInvalidRequest, "invalid credential") return preparedAttempt{}, &failure } + if providerKind == channel.ProviderDeepSeek && spec.ClientProtocol == protocol.Anthropic { + directKey.UseAnthropicEndpoints = schemas.Ptr(true) + } if spec.Operation == execution.OperationProbe { if mode == channel.RouteNative && providerKind == channel.ProviderGoogleVertex { passthroughPath, pathErr := vertexNativeGeminiPath(spec.UpstreamModel, "generateContent") @@ -509,6 +517,9 @@ func (r *Runtime) prepare(spec execution.AttemptSpec, stream bool) (preparedAtte false, "", ) + if targetErr == nil && mode == channel.RouteNative && providerKind == channel.ProviderDeepSeek { + typedURL, targetErr = deepSeekNativeTypedTarget(targetBaseURL, spec.ClientProtocol, "") + } if targetErr != nil { failure := notSentUnaryFailure(execution.ErrorKindInvalidRequest, "invalid channel probe target") return preparedAttempt{}, &failure @@ -588,6 +599,9 @@ func (r *Runtime) prepare(spec execution.AttemptSpec, stream bool) (preparedAtte return preparedAttempt{}, &failure } typedURL, targetErr := convertedTypedTarget(providerKind, targetBaseURL, spec.UpstreamModel, true, stream, safeQuery) + if targetErr == nil && mode == channel.RouteNative && providerKind == channel.ProviderDeepSeek { + typedURL, targetErr = deepSeekNativeTypedTarget(targetBaseURL, spec.ClientProtocol, safeQuery) + } if targetErr != nil { failure := notSentUnaryFailure(execution.ErrorKindInvalidRequest, "invalid compatible channel target") return preparedAttempt{}, &failure @@ -1053,6 +1067,11 @@ 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.OperationChatCompletion && + channel.ProviderKind(spec.TargetKind) == channel.ProviderDeepSeek && + spec.ClientProtocol == protocol.Anthropic { + bifrostContext.SetValue(schemas.BifrostContextKeyPassthroughExtraParams, true) + } if spec.Operation == execution.OperationProbe && channel.ProviderKind(spec.TargetKind) == channel.ProviderOpenAICompatible { bifrostContext.SetValue(schemas.BifrostContextKeyPassthroughExtraParams, true) diff --git a/internal/execution/bifrost/runtime_manager.go b/internal/execution/bifrost/runtime_manager.go index 70cff238..7aa8ad74 100644 --- a/internal/execution/bifrost/runtime_manager.go +++ b/internal/execution/bifrost/runtime_manager.go @@ -14,6 +14,7 @@ import ( "gpt-load/internal/channel" "gpt-load/internal/execution" + "gpt-load/internal/protocol" ) type effectiveProviderConfig struct { @@ -36,11 +37,31 @@ func buildEffectiveProviderConfig( if err != nil { return effectiveProviderConfig{}, err } - config := providerConfig(baseURL, custom, schemas.OpenAI, allowPrivateNetwork) + config := buildProviderConfig(provider, baseURL, custom, schemas.OpenAI, allowPrivateNetwork) + return newEffectiveProviderConfig(provider, baseURL, custom, config) +} + +func buildProviderConfig( + provider schemas.ModelProvider, + baseURL string, + custom bool, + baseProvider schemas.ModelProvider, + allowPrivateNetwork bool, +) *schemas.ProviderConfig { + config := providerConfig(baseURL, custom, baseProvider, allowPrivateNetwork) if !custom { config = providerConfig(baseURL, false, provider, allowPrivateNetwork) } config.CheckAndSetDefaults() + return config +} + +func newEffectiveProviderConfig( + provider schemas.ModelProvider, + baseURL string, + custom bool, + config *schemas.ProviderConfig, +) (effectiveProviderConfig, error) { canonical, err := json.Marshal(struct { Provider schemas.ModelProvider `json:"provider"` Config *schemas.ProviderConfig `json:"config"` @@ -59,6 +80,48 @@ func buildEffectiveProviderConfig( }, nil } +func buildEffectiveProviderConfigForAttempt( + resolved channel.ResolvedTarget, + spec execution.AttemptSpec, + allowPrivateNetwork bool, +) (effectiveProviderConfig, error) { + base, err := buildEffectiveProviderConfig(resolved, allowPrivateNetwork) + if err != nil || resolved.ProviderKind != channel.ProviderDeepSeek { + return base, err + } + if spec.ClientProtocol != protocol.OpenAIResponses || + (spec.Operation != execution.OperationResponsesCreate && spec.Operation != execution.OperationProbe) || + channel.RouteMode(spec.RouteMode) != channel.RouteNative { + return base, nil + } + + provider := customProviderKey(schemas.OpenAI, base.targetBaseURL) + config := buildProviderConfig(provider, base.targetBaseURL, true, schemas.OpenAI, allowPrivateNetwork) + config.CustomProviderConfig.AllowedRequests.Responses = true + config.CustomProviderConfig.AllowedRequests.ResponsesStream = true + config.CustomProviderConfig.AllowedRequests.ChatCompletion = false + config.CustomProviderConfig.AllowedRequests.ChatCompletionStream = false + config.CustomProviderConfig.RequestPathOverrides[schemas.ResponsesRequest] = base.targetBaseURL + "/responses" + config.CustomProviderConfig.RequestPathOverrides[schemas.ResponsesStreamRequest] = base.targetBaseURL + "/responses" + config.CheckAndSetDefaults() + return newEffectiveProviderConfig(provider, base.targetBaseURL, true, config) +} + +func buildDeepSeekResponsesConfig( + resolved channel.ResolvedTarget, + allowPrivateNetwork bool, +) (effectiveProviderConfig, error) { + return buildEffectiveProviderConfigForAttempt( + resolved, + execution.AttemptSpec{ + ClientProtocol: protocol.OpenAIResponses, + Operation: execution.OperationResponsesCreate, + RouteMode: execution.RouteNative, + }, + allowPrivateNetwork, + ) +} + func resolveSDKProviderConfig(resolved channel.ResolvedTarget) (schemas.ModelProvider, string, bool, error) { if preset, ok := sdkProviderSpecFor(resolved.ProviderKind); ok { baseURL, configured, err := targetBaseURL(resolved.TargetConfig) @@ -471,7 +534,7 @@ func (manager *RuntimeManager) configForAttempt(spec execution.AttemptSpec) (eff failure := notSentUnaryFailure(execution.ErrorKindInvalidRequest, "invalid channel target") return effectiveProviderConfig{}, &failure } - config, err := buildEffectiveProviderConfig(resolved, manager.options.allowPrivateNetwork) + config, err := buildEffectiveProviderConfigForAttempt(resolved, spec, manager.options.allowPrivateNetwork) if err != nil { failure := notSentUnaryFailure(execution.ErrorKindInvalidRequest, "invalid provider runtime config") return effectiveProviderConfig{}, &failure @@ -528,13 +591,23 @@ func (manager *RuntimeManager) Reconcile(targets []channel.ResolvedTarget) error if manager == nil || manager.pool == nil { return fmt.Errorf("reconcile provider runtimes: manager is unavailable") } - configs := make([]effectiveProviderConfig, 0, len(targets)) + configs := make([]effectiveProviderConfig, 0, len(targets)*2) for _, target := range targets { config, err := buildEffectiveProviderConfig(target, manager.options.allowPrivateNetwork) if err != nil { return fmt.Errorf("reconcile provider runtime %q: %w", target.ChannelID, err) } configs = append(configs, config) + if target.ProviderKind == channel.ProviderDeepSeek { + responsesMode, ok := target.Mode(protocol.OpenAIResponses, execution.OperationResponsesCreate) + if ok && responsesMode == channel.RouteNative { + responsesConfig, responsesErr := buildDeepSeekResponsesConfig(target, manager.options.allowPrivateNetwork) + if responsesErr != nil { + return fmt.Errorf("reconcile provider runtime %q Responses: %w", target.ChannelID, responsesErr) + } + configs = append(configs, responsesConfig) + } + } } manager.pool.reconcile(configs) return nil diff --git a/internal/execution/bifrost/runtime_manager_test.go b/internal/execution/bifrost/runtime_manager_test.go index 2e4a4d35..43bb814b 100644 --- a/internal/execution/bifrost/runtime_manager_test.go +++ b/internal/execution/bifrost/runtime_manager_test.go @@ -9,6 +9,7 @@ import ( "io" "net/http" "net/http/httptest" + "strings" "sync" "sync/atomic" "testing" @@ -55,6 +56,35 @@ func TestEffectiveProviderConfigUsesSDKProviderAndCanonicalDefaultBaseURL(t *tes } } +func TestDeepSeekResponsesUsesDedicatedOpenAIRuntimeConfig(t *testing.T) { + registry := channel.NewRegistry() + resolved, err := registry.Resolve(channel.DeepSeek, json.RawMessage(`{"base_url":"https://deepseek.example/api"}`)) + if err != nil { + t.Fatal(err) + } + base, err := buildEffectiveProviderConfig(resolved, true) + if err != nil { + t.Fatal(err) + } + responses, err := buildDeepSeekResponsesConfig(resolved, true) + if err != nil { + t.Fatal(err) + } + if base.provider != schemas.DeepSeek || base.custom || responses.provider == base.provider || !responses.custom { + t.Fatalf("base/responses configs = %+v/%+v", base, responses) + } + custom := responses.providerConfig.CustomProviderConfig + if custom == nil || custom.BaseProviderType != schemas.OpenAI || custom.AllowedRequests == nil || + !custom.AllowedRequests.Responses || !custom.AllowedRequests.ResponsesStream || + custom.AllowedRequests.ChatCompletion || custom.AllowedRequests.ChatCompletionStream { + t.Fatalf("DeepSeek Responses custom config = %+v", custom) + } + if custom.RequestPathOverrides[schemas.ResponsesRequest] != "https://deepseek.example/api/responses" || + custom.RequestPathOverrides[schemas.ResponsesStreamRequest] != "https://deepseek.example/api/responses" { + t.Fatalf("DeepSeek Responses path overrides = %#v", custom.RequestPathOverrides) + } +} + func TestRuntimeManagerPartitionsAzureEntraByCredentialIdentity(t *testing.T) { registry := channel.NewRegistry() target, err := registry.Resolve( @@ -583,8 +613,8 @@ func TestProductionRuntimeManagerUsesDeclaredResponsesOperation(t *testing.T) { }{ { channelID: channel.DeepSeek, - wantPath: "/chat/completions", - response: `{"id":"chat","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}}`, + wantPath: "/responses", + response: `{"id":"resp","object":"response","created_at":1,"status":"completed","model":"served","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}`, }, { channelID: channel.OpenRouter, @@ -640,6 +670,216 @@ func TestProductionRuntimeManagerUsesDeclaredResponsesOperation(t *testing.T) { } } +func TestProductionRuntimeManagerUsesDeepSeekNativeAnthropicEndpoint(t *testing.T) { + var calls atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + calls.Add(1) + if request.Method != http.MethodPost || request.URL.Path != "/anthropic/v1/messages" { + t.Errorf("request target = %s %s", request.Method, request.URL.Path) + } + if request.Header.Get("X-Api-Key") != "native-key" || request.Header.Get("Authorization") != "" { + t.Errorf("credential headers = x-api-key:%q authorization:%q", request.Header.Get("X-Api-Key"), request.Header.Get("Authorization")) + } + writer.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(writer, `{"id":"msg","type":"message","role":"assistant","model":"served","content":[{"type":"text","text":"OK"}],"stop_reason":"end_turn","usage":{"input_tokens":1,"output_tokens":1}}`) + })) + defer server.Close() + + manager, err := newRuntimeManager(runtimeOptions{allowPrivateNetwork: true}, channel.NewRegistry()) + if err != nil { + t.Fatal(err) + } + if err := manager.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(manager.Shutdown) + + spec := deepSeekAnthropicAttempt(t, server.URL) + result := manager.Execute(context.Background(), spec) + if validationErr := result.Validate(); validationErr != nil || result.Error != nil { + t.Fatalf("Execute() = %+v validation=%v", result, validationErr) + } + if calls.Load() != 1 || result.UpstreamAPI != execution.UpstreamAPIAnthropicMessages { + t.Fatalf("calls/result = %d/%+v", calls.Load(), result) + } +} + +func TestProductionRuntimeManagerRejectsDeepSeekAnthropicContainerBeforeDispatch(t *testing.T) { + var calls atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + calls.Add(1) + })) + defer server.Close() + + manager, err := newRuntimeManager(runtimeOptions{allowPrivateNetwork: true}, channel.NewRegistry()) + if err != nil { + t.Fatal(err) + } + if err := manager.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(manager.Shutdown) + + spec := deepSeekAnthropicAttempt(t, server.URL) + spec.RouteRequirement = execution.RouteRequirementNative + spec.Body = []byte(`{"model":"client-model","max_tokens":16,"container":{"id":"container_123"},"messages":[{"role":"user","content":"hello"}]}`) + result := manager.Execute(context.Background(), spec) + if validationErr := result.Validate(); validationErr != nil { + t.Fatalf("Execute() validation = %v; result=%+v", validationErr, result) + } + if result.DispatchState != execution.DispatchNotSent || result.ResponseStarted || result.Error == nil || + result.Error.Kind != execution.ErrorKindConversionUnsupported || + result.Error.Code != execution.ErrorCodeCriticalSemanticLoss { + t.Fatalf("Execute() = %+v", result) + } + if calls.Load() != 0 { + t.Fatalf("upstream calls = %d, want 0", calls.Load()) + } +} + +func TestProductionRuntimeManagerPreservesDeepSeekAnthropicEffort(t *testing.T) { + var calls atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + calls.Add(1) + body, err := io.ReadAll(request.Body) + if err != nil { + t.Fatal(err) + } + var payload struct { + OutputConfig *struct { + Effort string `json:"effort"` + } `json:"output_config"` + } + if err := json.Unmarshal(body, &payload); err != nil { + t.Fatalf("decode request body: %v; body=%s", err, body) + } + if payload.OutputConfig == nil || payload.OutputConfig.Effort != "high" { + t.Errorf("output_config = %+v; body=%s", payload.OutputConfig, body) + } + writer.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(writer, `{"id":"msg","type":"message","role":"assistant","model":"served","content":[{"type":"text","text":"OK"}],"stop_reason":"end_turn","usage":{"input_tokens":1,"output_tokens":1}}`) + })) + defer server.Close() + + manager, err := newRuntimeManager(runtimeOptions{allowPrivateNetwork: true}, channel.NewRegistry()) + if err != nil { + t.Fatal(err) + } + if err := manager.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(manager.Shutdown) + + spec := deepSeekAnthropicAttempt(t, server.URL) + spec.Body = []byte(`{"model":"client-model","max_tokens":8192,"thinking":{"type":"enabled","budget_tokens":4096},"output_config":{"effort":"high"},"messages":[{"role":"user","content":"hello"}]}`) + result := manager.Execute(context.Background(), spec) + if validationErr := result.Validate(); validationErr != nil || result.Error != nil { + t.Fatalf("Execute() = %+v validation=%v", result, validationErr) + } + if calls.Load() != 1 { + t.Fatalf("upstream calls = %d, want 1", calls.Load()) + } +} + +func TestProductionRuntimeManagerPreservesDeepSeekAnthropicEffortStream(t *testing.T) { + var calls atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + calls.Add(1) + body, err := io.ReadAll(request.Body) + if err != nil { + t.Error(err) + return + } + var payload struct { + Stream bool `json:"stream"` + OutputConfig *struct { + Effort string `json:"effort"` + } `json:"output_config"` + } + if err := json.Unmarshal(body, &payload); err != nil { + t.Errorf("decode request body: %v; body=%s", err, body) + return + } + if !payload.Stream || payload.OutputConfig == nil || payload.OutputConfig.Effort != "high" { + t.Errorf("stream/output_config = %t/%+v; body=%s", payload.Stream, payload.OutputConfig, body) + } + writer.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(writer, anthropicResponsesStreamFixture) + })) + defer server.Close() + + manager, err := newRuntimeManager(runtimeOptions{allowPrivateNetwork: true}, channel.NewRegistry()) + if err != nil { + t.Fatal(err) + } + if err := manager.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(manager.Shutdown) + + spec := deepSeekAnthropicAttempt(t, server.URL) + spec.Body = []byte(`{"model":"client-model","max_tokens":8192,"stream":true,"thinking":{"type":"enabled","budget_tokens":4096},"output_config":{"effort":"high"},"messages":[{"role":"user","content":"hello"}]}`) + result := manager.ExecuteStream(context.Background(), spec, func(execution.StreamEvent) error { return nil }) + if validationErr := result.Validate(); validationErr != nil || result.Error != nil { + t.Fatalf("ExecuteStream() = %+v validation=%v", result, validationErr) + } + if calls.Load() != 1 { + t.Fatalf("upstream calls = %d, want 1", calls.Load()) + } +} + +func TestProductionRuntimeManagerUsesDeepSeekNativeResponsesStream(t *testing.T) { + const upstreamStream = "event: response.created\n" + + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_stream\",\"object\":\"response\",\"status\":\"in_progress\",\"model\":\"served\",\"output\":[]}}\n\n" + + "event: response.completed\n" + + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_stream\",\"object\":\"response\",\"status\":\"completed\",\"model\":\"served\",\"output\":[],\"usage\":{\"input_tokens\":1,\"output_tokens\":1,\"total_tokens\":2}}}\n\n" + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.Method != http.MethodPost || request.URL.Path != "/responses" { + t.Errorf("request target = %s %s", request.Method, request.URL.Path) + } + if request.Header.Get("Authorization") != "Bearer native-key" { + t.Errorf("Authorization = %q", request.Header.Get("Authorization")) + } + body, err := io.ReadAll(request.Body) + if err != nil { + t.Fatal(err) + } + var payload map[string]any + if err := json.Unmarshal(body, &payload); err != nil || payload["stream"] != true { + t.Errorf("stream request = %s err=%v", body, err) + } + writer.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(writer, upstreamStream) + })) + defer server.Close() + + manager, err := newRuntimeManager(runtimeOptions{allowPrivateNetwork: true}, channel.NewRegistry()) + if err != nil { + t.Fatal(err) + } + if err := manager.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(manager.Shutdown) + + spec := openAIResponsesAttempt(t, channel.DeepSeek, server.URL) + var data bytes.Buffer + result := manager.ExecuteStream(context.Background(), spec, func(event execution.StreamEvent) error { + if event.Kind == execution.StreamEventData { + data.Write(event.Data) + } + return nil + }) + if validationErr := result.Validate(); validationErr != nil || result.Error != nil { + t.Fatalf("ExecuteStream() = %+v validation=%v", result, validationErr) + } + if !strings.Contains(data.String(), "event: response.created") || + !strings.Contains(data.String(), "event: response.completed") || + result.UpstreamAPI != execution.UpstreamAPIOpenAIResponses { + t.Fatalf("stream/result = %q/%+v", data.String(), result) + } +} + func TestProductionRuntimeManagerUsesNativeProviderListModelsPaths(t *testing.T) { tests := []struct { channelID channel.ID @@ -785,6 +1025,37 @@ func openAIResponsesAttempt(t *testing.T, channelID channel.ID, baseURL string) }) } +func deepSeekAnthropicAttempt(t *testing.T, baseURL string) execution.AttemptSpec { + t.Helper() + registry := channel.NewRegistry() + target, err := registry.Resolve(channel.DeepSeek, json.RawMessage(`{"base_url":"`+baseURL+`"}`)) + if err != nil { + t.Fatal(err) + } + mode, ok := target.Mode(protocol.Anthropic, execution.OperationChatCompletion) + if !ok { + t.Fatal("DeepSeek Anthropic route is missing") + } + return execution.NewAttemptSpec(execution.AttemptSpec{ + RequestID: "deepseek-anthropic-request", + AttemptID: "deepseek-anthropic-attempt", + Sequence: 1, + ChannelID: string(channel.DeepSeek), + TargetKind: string(target.ProviderKind), + TargetConfig: target.TargetConfig, + RouteMode: execution.RouteMode(mode), + ClientProtocol: protocol.Anthropic, + Operation: execution.OperationChatCompletion, + ClientModel: "client-model", + UpstreamModel: "upstream-model", + Method: http.MethodPost, + Path: "/v1/messages", + Header: make(http.Header), + Body: []byte(`{"model":"client-model","max_tokens":16,"messages":[{"role":"user","content":"hello"}]}`), + Credential: execution.NewCredentialSnapshot(1, 1, 1, []byte(`{"api_key":"native-key"}`)), + }) +} + func listModelsAttempt(t *testing.T, channelID channel.ID, baseURL string) execution.AttemptSpec { t.Helper() registry := channel.NewRegistry()