From 58e147fba6f6343ae5c2078de96803635b2cab4f Mon Sep 17 00:00:00 2001 From: Kingsley <10992364+kingsleydon@users.noreply.github.com> Date: Wed, 19 Aug 2026 04:40:32 +0000 Subject: [PATCH] feat(composite): support Codex endpoints --- .../internal/handler/openai_alpha_search.go | 8 ++- .../handler/openai_codex_models_handler.go | 4 +- .../openai_codex_models_handler_test.go | 19 ++++- backend/internal/handler/openai_live.go | 17 ++++- backend/internal/handler/openai_live_test.go | 3 + .../server/routes/composite_platform_test.go | 72 +++++++++++++++++++ backend/internal/server/routes/gateway.go | 50 +++++++++---- .../internal/server/routes/gateway_test.go | 4 +- backend/internal/service/admin_group.go | 4 +- .../service/admin_service_group_test.go | 38 ++++++++++ docs/COMPOSITE_GROUPS.md | 6 ++ frontend/src/views/admin/GroupsView.vue | 19 +++-- 12 files changed, 216 insertions(+), 28 deletions(-) diff --git a/backend/internal/handler/openai_alpha_search.go b/backend/internal/handler/openai_alpha_search.go index c5766e5669..6e0066682a 100644 --- a/backend/internal/handler/openai_alpha_search.go +++ b/backend/internal/handler/openai_alpha_search.go @@ -30,8 +30,8 @@ func (h *OpenAIGatewayHandler) AlphaSearch(c *gin.Context) { h.errorResponse(c, http.StatusUnauthorized, "authentication_error", "Invalid API key") return } - if apiKey.Group.Platform != service.PlatformOpenAI { - h.errorResponse(c, http.StatusNotFound, "not_found_error", "Codex alpha search is only available for OpenAI groups") + if apiKey.Group.Platform != service.PlatformOpenAI && apiKey.Group.Platform != service.PlatformComposite { + h.errorResponse(c, http.StatusNotFound, "not_found_error", "Codex alpha search is only available for OpenAI and Composite groups") return } subject, ok := middleware2.GetAuthSubjectFromContext(c) @@ -75,6 +75,10 @@ func (h *OpenAIGatewayHandler) AlphaSearch(c *gin.Context) { return } requestedModel := strings.TrimSpace(modelResult.String()) + if !compositeTargetPlatformAllowed(c, apiKey, requestedModel, service.PlatformOpenAI) { + h.errorResponse(c, http.StatusNotFound, "not_found_error", "Codex alpha search only supports OpenAI models for Composite groups") + return + } reqLog = reqLog.With(zap.String("model", requestedModel)) setOpsRequestContext(c, requestedModel, false) setOpsEndpointContext(c, "", int16(service.RequestTypeSync)) diff --git a/backend/internal/handler/openai_codex_models_handler.go b/backend/internal/handler/openai_codex_models_handler.go index 690cba9ce6..2714dc294f 100644 --- a/backend/internal/handler/openai_codex_models_handler.go +++ b/backend/internal/handler/openai_codex_models_handler.go @@ -27,8 +27,8 @@ func (h *OpenAIGatewayHandler) CodexModels(c *gin.Context) { h.errorResponse(c, http.StatusUnauthorized, "invalid_request_error", "API key group is required") return } - if apiKey.Group.Platform != service.PlatformOpenAI { - h.errorResponse(c, http.StatusNotFound, "not_found_error", "Codex models manifest is only available for OpenAI groups") + if apiKey.Group.Platform != service.PlatformOpenAI && apiKey.Group.Platform != service.PlatformComposite { + h.errorResponse(c, http.StatusNotFound, "not_found_error", "Codex models manifest is only available for OpenAI and Composite groups") return } diff --git a/backend/internal/handler/openai_codex_models_handler_test.go b/backend/internal/handler/openai_codex_models_handler_test.go index 7636fa5fd4..f8d1fada6c 100644 --- a/backend/internal/handler/openai_codex_models_handler_test.go +++ b/backend/internal/handler/openai_codex_models_handler_test.go @@ -116,6 +116,19 @@ func TestCodexModelsCanceledRequestDoesNotWriteResponse(t *testing.T) { } } +func TestCompositeCodexModelsReusesExistingManifestSelection(t *testing.T) { + handler, upstream, groupID := newCodexModelsFailoverTestHandler(http.StatusServiceUnavailable) + + recorder := performCodexModelsRequestForPlatform(t, handler, groupID, service.PlatformComposite) + + if got, want := upstream.calls(), []int64{1, 2}; !equalInt64Slices(got, want) { + t.Fatalf("upstream account calls: got %v, want %v", got, want) + } + if recorder.Code != http.StatusOK { + t.Fatalf("status: got %d, want %d; body=%s", recorder.Code, http.StatusOK, recorder.Body.String()) + } +} + func TestCodexModelsFailsOverFromRetryableUpstreamStatus(t *testing.T) { retryableStatuses := []int{ http.StatusTooManyRequests, @@ -287,13 +300,17 @@ func newCodexModelsFailoverTestHandlerWithAccountCount(firstStatus, accountCount } func performCodexModelsRequest(t *testing.T, handler *OpenAIGatewayHandler, groupID int64) *httptest.ResponseRecorder { + return performCodexModelsRequestForPlatform(t, handler, groupID, service.PlatformOpenAI) +} + +func performCodexModelsRequestForPlatform(t *testing.T, handler *OpenAIGatewayHandler, groupID int64, platform string) *httptest.ResponseRecorder { t.Helper() recorder := httptest.NewRecorder() c, _ := gin.CreateTestContext(recorder) c.Request = httptest.NewRequest(http.MethodGet, "/v1/models?client_version=0.144.0", nil) c.Set(string(middleware2.ContextKeyAPIKey), &service.APIKey{ GroupID: &groupID, - Group: &service.Group{ID: groupID, Platform: service.PlatformOpenAI}, + Group: &service.Group{ID: groupID, Platform: platform}, }) handler.CodexModels(c) diff --git a/backend/internal/handler/openai_live.go b/backend/internal/handler/openai_live.go index 0e4a5af45b..1f6b6c49a5 100644 --- a/backend/internal/handler/openai_live.go +++ b/backend/internal/handler/openai_live.go @@ -15,6 +15,7 @@ import ( coderws "github.com/coder/websocket" "github.com/gin-gonic/gin" "github.com/tidwall/gjson" + "github.com/tidwall/sjson" "go.uber.org/zap" ) @@ -29,7 +30,7 @@ func (h *OpenAIGatewayHandler) Live(c *gin.Context) { h.errorResponse(c, http.StatusInternalServerError, "api_error", "User context not found") return } - if apiKey.Group == nil || apiKey.Group.Platform != service.PlatformOpenAI { + if apiKey.Group == nil || (apiKey.Group.Platform != service.PlatformOpenAI && apiKey.Group.Platform != service.PlatformComposite) { h.errorResponse(c, http.StatusNotFound, "not_found_error", "Live is not supported for this platform") return } @@ -43,6 +44,18 @@ func (h *OpenAIGatewayHandler) Live(c *gin.Context) { return } model := strings.TrimSpace(gjson.GetBytes(request.Session, "model").String()) + if !compositeTargetPlatformAllowed(c, apiKey, model, service.PlatformOpenAI) { + h.errorResponse(c, http.StatusNotFound, "not_found_error", "Live only supports OpenAI models for Composite groups") + return + } + if upstreamModel, ok := service.ResolvedUpstreamModelFromContext(c.Request.Context()); ok && upstreamModel != model { + rewrittenSession, rewriteErr := sjson.SetBytes(request.Session, "model", upstreamModel) + if rewriteErr != nil { + h.errorResponse(c, http.StatusInternalServerError, "api_error", "Failed to apply Composite model route") + return + } + request.Session = rewrittenSession + } reqLog := requestLogger( c, "handler.openai_gateway.live", @@ -231,6 +244,6 @@ func (h *OpenAIGatewayHandler) LiveSideband(c *gin.Context) { func liveEnabledForAPIKey(apiKey *service.APIKey) bool { return apiKey != nil && apiKey.Group != nil && - apiKey.Group.Platform == service.PlatformOpenAI && + (apiKey.Group.Platform == service.PlatformOpenAI || apiKey.Group.Platform == service.PlatformComposite) && apiKey.Group.AllowLive } diff --git a/backend/internal/handler/openai_live_test.go b/backend/internal/handler/openai_live_test.go index 9df49654c8..93fc10fff9 100644 --- a/backend/internal/handler/openai_live_test.go +++ b/backend/internal/handler/openai_live_test.go @@ -87,6 +87,9 @@ func TestLiveEnabledForAPIKey(t *testing.T) { require.True(t, liveEnabledForAPIKey(&service.APIKey{ Group: &service.Group{Platform: service.PlatformOpenAI, AllowLive: true}, })) + require.True(t, liveEnabledForAPIKey(&service.APIKey{ + Group: &service.Group{Platform: service.PlatformComposite, AllowLive: true}, + })) } func TestLiveAttestationErrorIsExplicit(t *testing.T) { diff --git a/backend/internal/server/routes/composite_platform_test.go b/backend/internal/server/routes/composite_platform_test.go index 682d129ebc..3e64d2fbb8 100644 --- a/backend/internal/server/routes/composite_platform_test.go +++ b/backend/internal/server/routes/composite_platform_test.go @@ -133,6 +133,78 @@ func TestCompositeTargetPlatformMiddlewareUsesExplicitRouteAndRewritesBody(t *te require.Equal(t, http.StatusNoContent, w.Code) } +func TestCompositeTargetPlatformMiddlewareRewritesNestedLiveModel(t *testing.T) { + gin.SetMode(gin.TestMode) + router := gin.New() + resolver := service.NewCompositeRouteResolver(compositeRouteRepoStub{ + routes: []service.CompositeModelRoute{ + { + ID: 1, + GroupID: 1, + PublicModel: "live-alias", + MatchType: service.CompositeRouteMatchExact, + TargetPlatform: service.PlatformOpenAI, + UpstreamModel: "gpt-live", + Endpoint: service.CompositeRouteEndpointAny, + Priority: 100, + Enabled: true, + }, + }, + }) + router.Use(gin.HandlerFunc(servermiddleware.APIKeyAuthMiddleware(func(c *gin.Context) { + groupID := int64(1) + c.Set(string(servermiddleware.ContextKeyAPIKey), &service.APIKey{ + GroupID: &groupID, + Group: &service.Group{ID: groupID, Platform: service.PlatformComposite}, + }) + c.Next() + }))) + router.Use(compositeTargetPlatformMiddleware(resolver)) + router.POST("/backend-api/codex/realtime/calls", func(c *gin.Context) { + platform, ok := service.ResolvedTargetPlatformFromContext(c.Request.Context()) + require.True(t, ok) + require.Equal(t, service.PlatformOpenAI, platform) + + body, err := io.ReadAll(c.Request.Body) + require.NoError(t, err) + require.JSONEq(t, `{"session":{"model":"gpt-live"},"sdp":"v=0"}`, string(body)) + c.Status(http.StatusNoContent) + }) + + req := httptest.NewRequest( + http.MethodPost, + "/backend-api/codex/realtime/calls", + strings.NewReader(`{"session":{"model":"live-alias"},"sdp":"v=0"}`), + ) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + + router.ServeHTTP(w, req) + + require.Equal(t, http.StatusNoContent, w.Code) +} + +func TestCompositeRequestModelFromMultipartLiveSession(t *testing.T) { + var body bytes.Buffer + writer := multipart.NewWriter(&body) + require.NoError(t, writer.WriteField("sdp", "v=0")) + require.NoError(t, writer.WriteField("session", `{"model":"live-alias"}`)) + require.NoError(t, writer.Close()) + + require.Equal(t, "live-alias", compositeRequestModelFromBody(writer.FormDataContentType(), body.Bytes())) +} + +func TestCompositeCodexControlPathsUseResponsesRoutes(t *testing.T) { + for _, path := range []string{ + "/v1/alpha/search", + "/backend-api/codex/alpha/search", + "/v1/live", + "/backend-api/codex/realtime/calls", + } { + require.Equal(t, service.CompositeRouteEndpointResponses, compositeRouteEndpointForPath(path), "path=%s", path) + } +} + func TestCompositeTargetPlatformMiddlewareUsesExplicitRouteForMultipartImages(t *testing.T) { gin.SetMode(gin.TestMode) router := gin.New() diff --git a/backend/internal/server/routes/gateway.go b/backend/internal/server/routes/gateway.go index 3c4b33f519..9a06ae9c79 100644 --- a/backend/internal/server/routes/gateway.go +++ b/backend/internal/server/routes/gateway.go @@ -55,9 +55,6 @@ func RegisterGatewayRoutes( return false } } - isOpenAIGatewayPlatform := func(c *gin.Context) bool { - return getGroupPlatform(c) == service.PlatformOpenAI - } countTokensHandler := func(c *gin.Context) { switch getGroupPlatform(c) { case service.PlatformOpenAI, service.PlatformKimi, service.PlatformZhipu, service.PlatformDeepseek: @@ -69,9 +66,12 @@ func RegisterGatewayRoutes( } } modelsHandler := func(c *gin.Context) { - if isOpenAIGatewayPlatform(c) && c.Query("client_version") != "" { - h.OpenAIGateway.CodexModels(c) - return + if c.Query("client_version") != "" { + switch getGroupPlatform(c) { + case service.PlatformOpenAI, service.PlatformComposite: + h.OpenAIGateway.CodexModels(c) + return + } } h.Gateway.Models(c) } @@ -551,8 +551,10 @@ func compositeTargetPlatformMiddleware(resolver *service.CompositeRouteResolver) if decision.Matched { c.Request = c.Request.WithContext(service.WithCompositeRouteDecision(c.Request.Context(), decision)) if upstreamModel := strings.TrimSpace(decision.UpstreamModel); upstreamModel != "" && upstreamModel != model && gjson.ValidBytes(body) { - if rewritten, rewriteErr := sjson.SetBytes(body, "model", upstreamModel); rewriteErr == nil { - body = rewritten + if _, modelPath := compositeJSONRequestModel(body); modelPath != "" { + if rewritten, rewriteErr := sjson.SetBytes(body, modelPath, upstreamModel); rewriteErr == nil { + body = rewritten + } } } } @@ -563,12 +565,25 @@ func compositeTargetPlatformMiddleware(resolver *service.CompositeRouteResolver) } func compositeRequestModelFromBody(contentType string, body []byte) string { - if model := strings.TrimSpace(gjson.GetBytes(body, "model").String()); model != "" { + if model, _ := compositeJSONRequestModel(body); model != "" { return model } return compositeMultipartModelFromBody(contentType, body) } +func compositeJSONRequestModel(body []byte) (string, string) { + for _, path := range []string{"model", "session.model"} { + model := gjson.GetBytes(body, path) + if model.Type != gjson.String { + continue + } + if value := strings.TrimSpace(model.String()); value != "" { + return value, path + } + } + return "", "" +} + func compositeMultipartModelFromBody(contentType string, body []byte) string { mediaType, params, err := mime.ParseMediaType(strings.TrimSpace(contentType)) if err != nil || !strings.EqualFold(mediaType, "multipart/form-data") { @@ -587,14 +602,22 @@ func compositeMultipartModelFromBody(contentType string, body []byte) string { if err != nil { return "" } - if part.FormName() != "model" || part.FileName() != "" { + fieldName := part.FormName() + if part.FileName() != "" || (fieldName != "model" && fieldName != "session") { continue } data, err := io.ReadAll(part) if err != nil { return "" } - return strings.TrimSpace(string(data)) + switch fieldName { + case "model": + return strings.TrimSpace(string(data)) + case "session": + if model, _ := compositeJSONRequestModel(data); model != "" { + return model + } + } } } @@ -670,7 +693,10 @@ func compositeRouteEndpointForPath(path string) string { return service.CompositeRouteEndpointCountTokens case strings.Contains(path, "/messages"): return service.CompositeRouteEndpointMessages - case strings.Contains(path, "/responses"): + case strings.Contains(path, "/responses"), + strings.Contains(path, "/alpha/search"), + strings.Contains(path, "/realtime/calls"), + strings.HasSuffix(strings.TrimRight(path, "/"), "/live"): return service.CompositeRouteEndpointResponses case strings.Contains(path, "/chat/completions"): return service.CompositeRouteEndpointChatCompletions diff --git a/backend/internal/server/routes/gateway_test.go b/backend/internal/server/routes/gateway_test.go index ded7bd2989..362054a7de 100644 --- a/backend/internal/server/routes/gateway_test.go +++ b/backend/internal/server/routes/gateway_test.go @@ -94,7 +94,7 @@ func TestGatewayRoutesOpenAIAlphaSearchPathsAreRegistered(t *testing.T) { } } -func TestGatewayRoutesAlphaSearchRejectsNonOpenAIGroup(t *testing.T) { +func TestGatewayRoutesAlphaSearchRejectsUnsupportedGroup(t *testing.T) { router := newGatewayRoutesTestRouter(service.PlatformGrok) req := httptest.NewRequest(http.MethodPost, "/v1/alpha/search", strings.NewReader(`{"model":"gpt-5.6-sol"}`)) req.Header.Set("Content-Type", "application/json") @@ -103,7 +103,7 @@ func TestGatewayRoutesAlphaSearchRejectsNonOpenAIGroup(t *testing.T) { router.ServeHTTP(w, req) require.Equal(t, http.StatusNotFound, w.Code) - require.Contains(t, w.Body.String(), "only available for OpenAI groups") + require.Contains(t, w.Body.String(), "only available for OpenAI and Composite groups") } func TestGatewayRoutesOpenAIImagesPathsAreRegistered(t *testing.T) { diff --git a/backend/internal/service/admin_group.go b/backend/internal/service/admin_group.go index f8f9acb4e7..83467c9849 100644 --- a/backend/internal/service/admin_group.go +++ b/backend/internal/service/admin_group.go @@ -510,7 +510,7 @@ func (s *adminServiceImpl) CreateGroup(ctx context.Context, input *CreateGroupIn ReasoningEffortMappings: reasoningEffortMappings, } sanitizeGroupMessagesDispatchFields(group) - if group.Platform != PlatformOpenAI { + if group.Platform != PlatformOpenAI && group.Platform != PlatformComposite { group.AllowLive = false } sanitizeGroupReasoningEffortPolicy(group) @@ -890,7 +890,7 @@ func (s *adminServiceImpl) UpdateGroup(ctx context.Context, id int64, input *Upd group.ReasoningEffortMappings = reasoningEffortMappings } sanitizeGroupMessagesDispatchFields(group) - if group.Platform != PlatformOpenAI { + if group.Platform != PlatformOpenAI && group.Platform != PlatformComposite { group.AllowLive = false } sanitizeGroupReasoningEffortPolicy(group) diff --git a/backend/internal/service/admin_service_group_test.go b/backend/internal/service/admin_service_group_test.go index 64bfdaeb02..8b0374e6be 100644 --- a/backend/internal/service/admin_service_group_test.go +++ b/backend/internal/service/admin_service_group_test.go @@ -1029,6 +1029,44 @@ func TestAdminService_CreateGroup_ClearsMessagesDispatchFieldsForNonOpenAIPlatfo require.Equal(t, OpenAIMessagesDispatchModelConfig{}, repo.created.MessagesDispatchModelConfig) } +func TestAdminService_CreateCompositeGroupPreservesLive(t *testing.T) { + repo := &groupRepoStubForAdmin{} + svc := &adminServiceImpl{groupRepo: repo} + + group, err := svc.CreateGroup(context.Background(), &CreateGroupInput{ + Name: "composite-group", + Platform: PlatformComposite, + RateMultiplier: 1.0, + AllowLive: true, + }) + + require.NoError(t, err) + require.NotNil(t, group) + require.NotNil(t, repo.created) + require.True(t, repo.created.AllowLive) +} + +func TestAdminService_UpdateCompositeGroupPreservesLive(t *testing.T) { + existingGroup := &Group{ + ID: 1, + Name: "composite-group", + Platform: PlatformComposite, + Status: StatusActive, + } + repo := &groupRepoStubForAdmin{getByID: existingGroup} + svc := &adminServiceImpl{groupRepo: repo} + allowLive := true + + group, err := svc.UpdateGroup(context.Background(), existingGroup.ID, &UpdateGroupInput{ + AllowLive: &allowLive, + }) + + require.NoError(t, err) + require.NotNil(t, group) + require.NotNil(t, repo.updated) + require.True(t, repo.updated.AllowLive) +} + func TestAdminService_UpdateGroup_ClearsMessagesDispatchFieldsWhenPlatformChangesAwayFromOpenAI(t *testing.T) { existingGroup := &Group{ ID: 1, diff --git a/docs/COMPOSITE_GROUPS.md b/docs/COMPOSITE_GROUPS.md index c0e849e538..de6d0cf071 100644 --- a/docs/COMPOSITE_GROUPS.md +++ b/docs/COMPOSITE_GROUPS.md @@ -53,6 +53,12 @@ route's `upstream_model` before dispatch. For Gemini native paths such as `/v1beta/models/{model}:generateContent`, the gateway resolves `{model}` and the handler forwards the resolved upstream model. +Codex Alpha Search and Live requests use the `responses` route domain. Live +requests resolve the model from `session.model`, including multipart `session` +payloads, and apply the configured `upstream_model` before dispatch. +Codex model manifest requests reuse the existing OpenAI account selection and +failover path within the Composite group. + ## Built-In Detection Composite routing detects common public model IDs and provider-prefixed IDs: diff --git a/frontend/src/views/admin/GroupsView.vue b/frontend/src/views/admin/GroupsView.vue index 73d75cb4e0..b0144f4686 100644 --- a/frontend/src/views/admin/GroupsView.vue +++ b/frontend/src/views/admin/GroupsView.vue @@ -1581,9 +1581,9 @@ - +