Merge branch 'v2' into codex/codex-subscription
This commit is contained in:
21 files changed
+758
-36
No files matched your search
@@ -6,7 +6,7 @@ the complete resolved dependency inventory.
|
||||
## Bifrost Core
|
||||
|
||||
- Module: `github.com/maximhq/bifrost/core`
|
||||
- Version: `v1.7.7`
|
||||
- Version: `v1.7.11`
|
||||
- Copyright: 2025 H3 Labs Inc.
|
||||
- License: Apache License 2.0
|
||||
|
||||
|
||||
@@ -8,7 +8,7 @@ require (
|
||||
github.com/glebarez/sqlite v1.11.0
|
||||
github.com/joho/godotenv v1.5.1
|
||||
github.com/klauspost/compress v1.19.1
|
||||
github.com/maximhq/bifrost/core v1.7.7
|
||||
github.com/maximhq/bifrost/core v1.7.11
|
||||
github.com/nicksnyder/go-i18n/v2 v2.6.1
|
||||
github.com/sirupsen/logrus v1.9.4
|
||||
go.uber.org/dig v1.19.0
|
||||
|
||||
@@ -174,8 +174,8 @@ github.com/mattn/go-isatty v0.0.24 h1:tGZZoVgT/KiqK1c8ocVLeDS8BSWMRd47J3Lbz7vsRe
|
||||
github.com/mattn/go-isatty v0.0.24/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A=
|
||||
github.com/mattn/go-sqlite3 v1.14.22 h1:2gZY6PC6kBnID23Tichd1K+Z0oS6nE/XwU+Vz/5o4kU=
|
||||
github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
|
||||
github.com/maximhq/bifrost/core v1.7.7 h1:5mMspsjvP/yJ4XBvuuAZ99R0BF+Dyy2l6FFO6UZxLfk=
|
||||
github.com/maximhq/bifrost/core v1.7.7/go.mod h1:XQGQ99V6iW2yRbq4zg5+LfDcJMR06x/Ie4HuC8FWEDI=
|
||||
github.com/maximhq/bifrost/core v1.7.11 h1:9dhHNM9IN3DPM7q+Xhg06LW+WGITz9d9m0RyeVHOQ4Y=
|
||||
github.com/maximhq/bifrost/core v1.7.11/go.mod h1:ftRe80M+N8GTP7OvEG2G4yNK0jr/s/T4XFSA/YuuOTA=
|
||||
github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
|
||||
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
|
||||
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
@@ -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"},
|
||||
@@ -343,6 +343,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,
|
||||
|
||||
@@ -543,8 +543,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 {
|
||||
|
||||
@@ -173,6 +173,9 @@ func (service *Service) ListRequestLogs(
|
||||
}
|
||||
page, err := service.requestLogs.List(ctx, query)
|
||||
if err != nil {
|
||||
if requestWasCanceled(ctx, err) {
|
||||
return requestlog.Page{}, err
|
||||
}
|
||||
return requestlog.Page{}, app_errors.ParseDBError(err)
|
||||
}
|
||||
return page, nil
|
||||
@@ -184,6 +187,9 @@ func (service *Service) GetRequestLog(ctx context.Context, requestID string) (re
|
||||
}
|
||||
record, err := service.requestLogs.Get(ctx, requestID)
|
||||
if err != nil {
|
||||
if requestWasCanceled(ctx, err) {
|
||||
return requestlog.Record{}, err
|
||||
}
|
||||
return requestlog.Record{}, app_errors.ParseDBError(err)
|
||||
}
|
||||
return record, nil
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
@@ -15,6 +16,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/sirupsen/logrus"
|
||||
|
||||
"gpt-load/internal/channel"
|
||||
"gpt-load/internal/execution"
|
||||
@@ -27,6 +29,95 @@ import (
|
||||
"gpt-load/internal/usage"
|
||||
)
|
||||
|
||||
func TestRequestLogReadsPreserveRequestCancellation(t *testing.T) {
|
||||
fixture := newServiceFixture(t)
|
||||
reader := &recordingRequestLogReader{err: context.Canceled, getErr: context.Canceled}
|
||||
fixture.service.requestLogs = reader
|
||||
requestCtx, cancel := context.WithCancel(t.Context())
|
||||
cancel()
|
||||
|
||||
if _, err := fixture.service.ListRequestLogs(requestCtx, requestlog.ListQuery{}); !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("ListRequestLogs() error = %v, want context.Canceled", err)
|
||||
}
|
||||
if _, err := fixture.service.GetRequestLog(requestCtx, "request-id"); !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("GetRequestLog() error = %v, want context.Canceled", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListRequestLogsSuppressesOnlyCanceledHTTPRequestErrors(t *testing.T) {
|
||||
initControlI18n(t)
|
||||
tests := []struct {
|
||||
name string
|
||||
requestCtx func() context.Context
|
||||
readerErr error
|
||||
wantError bool
|
||||
}{
|
||||
{
|
||||
name: "canceled request",
|
||||
requestCtx: func() context.Context {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
return ctx
|
||||
},
|
||||
readerErr: context.Canceled,
|
||||
},
|
||||
{
|
||||
name: "deadline exceeded",
|
||||
requestCtx: func() context.Context {
|
||||
ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
|
||||
cancel()
|
||||
return ctx
|
||||
},
|
||||
readerErr: context.DeadlineExceeded,
|
||||
wantError: true,
|
||||
},
|
||||
{
|
||||
name: "database failure after request cancellation",
|
||||
requestCtx: func() context.Context {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
return ctx
|
||||
},
|
||||
readerErr: errors.New("database unavailable"),
|
||||
wantError: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
fixture := newServiceFixture(t)
|
||||
fixture.service.requestLogs = &recordingRequestLogReader{err: test.readerErr}
|
||||
server := NewServer(&config.Config{AuthKey: "test-auth-key"}, fixture.service)
|
||||
recorder := httptest.NewRecorder()
|
||||
ginContext, _ := gin.CreateTestContext(recorder)
|
||||
ginContext.Request = httptest.NewRequest(http.MethodGet, "/api/logs", nil).
|
||||
WithContext(test.requestCtx())
|
||||
|
||||
var logs bytes.Buffer
|
||||
logger := logrus.StandardLogger()
|
||||
previousOutput, previousFormatter := logger.Out, logger.Formatter
|
||||
logrus.SetOutput(&logs)
|
||||
logrus.SetFormatter(&logrus.JSONFormatter{DisableTimestamp: true})
|
||||
t.Cleanup(func() {
|
||||
logrus.SetOutput(previousOutput)
|
||||
logrus.SetFormatter(previousFormatter)
|
||||
})
|
||||
|
||||
server.handleListRequestLogs(ginContext)
|
||||
loggedFailure := strings.Contains(logs.String(), "[CONTROL] Operation failed")
|
||||
if test.wantError {
|
||||
if recorder.Body.Len() == 0 || !loggedFailure {
|
||||
t.Fatalf("error response/log = %q / %q", recorder.Body.String(), logs.String())
|
||||
}
|
||||
return
|
||||
}
|
||||
if recorder.Body.Len() != 0 || loggedFailure {
|
||||
t.Fatalf("canceled response/log = %q / %q", recorder.Body.String(), logs.String())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequestLogEndpointRejectsUnknownDuplicateAndMalformedQueries(t *testing.T) {
|
||||
validCursor := encodeTestCursorPayload(
|
||||
`{"v":2,"completed_at_ms":1784894400000,"request_id":"00000000-0000-4000-8000-000000000001"}`,
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package control
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"crypto/subtle"
|
||||
"encoding/json"
|
||||
@@ -794,6 +795,10 @@ func writeServiceErrorResponse(
|
||||
operation string,
|
||||
err error,
|
||||
) {
|
||||
if requestWasCanceled(c.Request.Context(), err) {
|
||||
return
|
||||
}
|
||||
|
||||
var apiErr *app_errors.APIError
|
||||
if errors.As(err, &apiErr) {
|
||||
setMutationErrorCode(c, apiErr.Code)
|
||||
@@ -813,6 +818,12 @@ func writeServiceErrorResponse(
|
||||
response.ErrorI18nFromAPIError(c, app_errors.ErrInternalServer, "internal_error")
|
||||
}
|
||||
|
||||
func requestWasCanceled(ctx context.Context, err error) bool {
|
||||
return ctx != nil &&
|
||||
errors.Is(ctx.Err(), context.Canceled) &&
|
||||
errors.Is(err, context.Canceled)
|
||||
}
|
||||
|
||||
func serviceErrorMessageID(
|
||||
operation string,
|
||||
err error,
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -3,7 +3,9 @@ package storage
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net/url"
|
||||
"os"
|
||||
@@ -25,13 +27,51 @@ const (
|
||||
databaseMaxIdleConnections = 5
|
||||
)
|
||||
|
||||
var databaseLogger = logger.New(log.New(os.Stdout, "\r\n", log.LstdFlags), logger.Config{
|
||||
SlowThreshold: 200 * time.Millisecond,
|
||||
LogLevel: logger.Warn,
|
||||
IgnoreRecordNotFoundError: true,
|
||||
ParameterizedQueries: true,
|
||||
Colorful: true,
|
||||
})
|
||||
var databaseLogger = newDatabaseLogger(os.Stdout)
|
||||
|
||||
func newDatabaseLogger(output io.Writer) logger.Interface {
|
||||
base := logger.New(log.New(output, "\r\n", log.LstdFlags), logger.Config{
|
||||
SlowThreshold: 200 * time.Millisecond,
|
||||
LogLevel: logger.Warn,
|
||||
IgnoreRecordNotFoundError: true,
|
||||
ParameterizedQueries: true,
|
||||
Colorful: true,
|
||||
})
|
||||
return databaseLogFilter{Interface: base}
|
||||
}
|
||||
|
||||
type databaseLogFilter struct {
|
||||
logger.Interface
|
||||
}
|
||||
|
||||
func (filter databaseLogFilter) LogMode(level logger.LogLevel) logger.Interface {
|
||||
return databaseLogFilter{Interface: filter.Interface.LogMode(level)}
|
||||
}
|
||||
|
||||
func (filter databaseLogFilter) Trace(
|
||||
ctx context.Context,
|
||||
begin time.Time,
|
||||
query func() (string, int64),
|
||||
err error,
|
||||
) {
|
||||
if ctx != nil &&
|
||||
errors.Is(ctx.Err(), context.Canceled) &&
|
||||
errors.Is(err, context.Canceled) {
|
||||
return
|
||||
}
|
||||
filter.Interface.Trace(ctx, begin, query, err)
|
||||
}
|
||||
|
||||
func (filter databaseLogFilter) ParamsFilter(
|
||||
ctx context.Context,
|
||||
query string,
|
||||
params ...interface{},
|
||||
) (string, []interface{}) {
|
||||
if paramsFilter, ok := filter.Interface.(gorm.ParamsFilter); ok {
|
||||
return paramsFilter.ParamsFilter(ctx, query, params...)
|
||||
}
|
||||
return query, nil
|
||||
}
|
||||
|
||||
// Open opens a database using a fully resolved DSN.
|
||||
// Resolving an empty DSN to DATA_DIR belongs to platform/config.
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestDatabaseLoggerSuppressesOnlyCanceledRequestQueries(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
context func() context.Context
|
||||
err error
|
||||
wantLogged bool
|
||||
wantMessage string
|
||||
}{
|
||||
{
|
||||
name: "canceled request",
|
||||
context: func() context.Context {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
return ctx
|
||||
},
|
||||
err: context.Canceled,
|
||||
},
|
||||
{
|
||||
name: "deadline exceeded",
|
||||
context: func() context.Context {
|
||||
ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
|
||||
cancel()
|
||||
return ctx
|
||||
},
|
||||
err: context.DeadlineExceeded,
|
||||
wantLogged: true,
|
||||
wantMessage: context.DeadlineExceeded.Error(),
|
||||
},
|
||||
{
|
||||
name: "database failure after request cancellation",
|
||||
context: func() context.Context {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
return ctx
|
||||
},
|
||||
err: errors.New("database unavailable"),
|
||||
wantLogged: true,
|
||||
wantMessage: "database unavailable",
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
var output bytes.Buffer
|
||||
queryEvaluated := false
|
||||
newDatabaseLogger(&output).Trace(
|
||||
test.context(),
|
||||
time.Now(),
|
||||
func() (string, int64) {
|
||||
queryEvaluated = true
|
||||
return "SELECT * FROM request_logs", 20
|
||||
},
|
||||
test.err,
|
||||
)
|
||||
|
||||
if test.wantLogged {
|
||||
if !queryEvaluated || !strings.Contains(output.String(), test.wantMessage) {
|
||||
t.Fatalf("database log = %q, query evaluated = %t", output.String(), queryEvaluated)
|
||||
}
|
||||
return
|
||||
}
|
||||
if queryEvaluated || output.Len() != 0 {
|
||||
t.Fatalf("canceled query log = %q, query evaluated = %t", output.String(), queryEvaluated)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -235,7 +235,7 @@ async function submit(): Promise<void> {
|
||||
</AppButton>
|
||||
|
||||
<p id="auth-session-note" class="ledger-login__session-note">
|
||||
{{ t('auth.sessionNotePrefix') }}<code>sessionStorage</code
|
||||
{{ t('auth.sessionNotePrefix') }}<code>localStorage</code
|
||||
>{{ t('auth.sessionNoteSuffix') }}
|
||||
</p>
|
||||
</form>
|
||||
|
||||
@@ -85,8 +85,8 @@ export default {
|
||||
network: 'Unable to reach the management API. Check the service and try again.',
|
||||
invalidResponse: 'The management API returned an unrecognized response.',
|
||||
checking: 'Verifying the current session…',
|
||||
sessionNotePrefix: 'The sign-in key is stored only in ',
|
||||
sessionNoteSuffix: ' for this browser session',
|
||||
sessionNotePrefix: "The sign-in key is stored in this site's ",
|
||||
sessionNoteSuffix: ' and cleared on sign-out or authentication failure',
|
||||
help: {
|
||||
title: 'Which key can I use?',
|
||||
accessKeyTitle: 'AccessKey',
|
||||
|
||||
@@ -84,8 +84,8 @@ export default {
|
||||
network: '管理 API に接続できません。サービスを確認して再試行してください。',
|
||||
invalidResponse: '管理 API から認識できない応答が返されました。',
|
||||
checking: '現在のセッションを確認しています…',
|
||||
sessionNotePrefix: 'ログインキーは、このブラウザーセッションの ',
|
||||
sessionNoteSuffix: ' にのみ保存されます',
|
||||
sessionNotePrefix: 'ログインキーは、このサイトの ',
|
||||
sessionNoteSuffix: ' に保存され、ログアウトまたは認証失敗時に削除されます',
|
||||
help: {
|
||||
title: 'どのキーを使用できますか?',
|
||||
accessKeyTitle: 'AccessKey',
|
||||
|
||||
@@ -84,8 +84,8 @@ export default {
|
||||
network: '无法连接到管理 API,请检查服务后重试。',
|
||||
invalidResponse: '管理 API 返回了无法识别的响应。',
|
||||
checking: '正在验证当前会话…',
|
||||
sessionNotePrefix: '登录密钥仅保存在当前浏览器会话的 ',
|
||||
sessionNoteSuffix: ' 中',
|
||||
sessionNotePrefix: '登录密钥保存在当前站点的 ',
|
||||
sessionNoteSuffix: ' 中,退出登录或认证失效时清除',
|
||||
help: {
|
||||
title: '可以使用哪种密钥?',
|
||||
accessKeyTitle: 'AccessKey',
|
||||
|
||||
+1
-1
@@ -88,7 +88,7 @@ async function bootstrap(): Promise<void> {
|
||||
})
|
||||
|
||||
authSession = createAuthSession({
|
||||
storage: getBrowserStorage('sessionStorage'),
|
||||
storage: getBrowserStorage('localStorage'),
|
||||
queryClient,
|
||||
onClear: clearEphemeralState,
|
||||
validate: (key, globalUnauthorized, signal) =>
|
||||
|
||||
Reference in new issue
Block a user