From 2d8a3868110d81c5e91119a504b7d0537a284cde Mon Sep 17 00:00:00 2001 From: shaw Date: Fri, 2 Oct 2026 13:08:40 +0800 Subject: [PATCH] =?UTF-8?q?fix(typesafe):=20=E8=A1=A5=E9=BD=90=E5=B9=B3?= =?UTF-8?q?=E5=8F=B0=E8=BF=81=E7=A7=BB=E3=80=81=E5=AE=A1=E8=AE=A1=E8=A6=86?= =?UTF-8?q?=E7=9B=96=E3=80=81=E7=AB=AF=E7=82=B9=E9=9A=94=E7=A6=BB=E4=B8=8E?= =?UTF-8?q?=E9=94=99=E8=AF=AF=E7=AD=96=E7=95=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 241 迁移：user_platform_quotas / composite_model_routes 的 CHECK 约束 加入 typesafe。此前设置 typesafe 默认配额会让注册时的多行配额快照整体 违约（fail-open 后新用户所有平台配额丢失），单用户配额与 Composite 路由保存 500。 - Prompt 审计新增 typesafe_systemone 协议提取（state + 各问题 instructions/ criteria/选项标签，键排序保证哈希稳定）；此前提取为空，阻断与异步审计均被放行。 旧内容审核同样覆盖问题文本，且不再丢弃含 的 System One 文本。 - TypeSafe 分组及路由到 TypeSafe 的 Composite 请求访问 Messages / count_tokens / Chat Completions / Responses 时返回 404，避免以 x-api-key 打到错误上游路径并 污染账号状态；TypeSafe 账号 base_url 为空时不再回退 api.anthropic.com。 - SystemOne 补在途余额预留、利润控制准入终检、wrapReleaseOnDone，失败切换改用 共享 FailoverState（同账号重试 / 池模式 / 临时封禁）。 - 上游错误沿用共享账号错误策略：402/403、自定义错误码、临时不可调度规则生效， 记录 ops 上游错误；400/422 仍不切换且永不改变账号状态。 - 账号测试改走原生 System One 探测，可用模型只返回 jev-latest； 响应超限显式报错，响应 Content-Type 仅透传 JSON 类型。 --- README.md | 2 +- README_CN.md | 2 +- .../internal/handler/admin/account_handler.go | 7 + .../account_handler_available_models_test.go | 33 ++++ backend/internal/handler/gateway_handler.go | 6 + .../gateway_handler_chat_completions.go | 3 + .../handler/gateway_handler_responses.go | 3 + backend/internal/handler/gateway_systemone.go | 181 +++++++++++++----- .../handler/gateway_systemone_test.go | 66 +++++++ backend/internal/pkg/typesafe/client.go | 16 +- backend/internal/pkg/typesafe/systemone.go | 14 +- .../internal/pkg/typesafe/systemone_test.go | 2 + .../internal/securityaudit/prompt_snapshot.go | 60 ++++++ .../securityaudit/prompt_snapshot_test.go | 39 ++++ backend/internal/service/account.go | 4 + .../internal/service/account_test_service.go | 4 + .../service/account_test_service_typesafe.go | 95 +++++++++ .../account_test_service_typesafe_test.go | 92 +++++++++ .../service/content_moderation_input.go | 29 ++- .../content_moderation_systemone_test.go | 8 + backend/internal/service/gateway_systemone.go | 105 +++++++--- .../service/gateway_systemone_test.go | 137 +++++++++++++ .../migrations/241_add_typesafe_platform.sql | 26 +++ .../typesafe_platform_migration_test.go | 21 ++ 24 files changed, 859 insertions(+), 96 deletions(-) create mode 100644 backend/internal/service/account_test_service_typesafe.go create mode 100644 backend/internal/service/account_test_service_typesafe_test.go create mode 100644 backend/migrations/241_add_typesafe_platform.sql create mode 100644 backend/migrations/typesafe_platform_migration_test.go diff --git a/README.md b/README.md index 48a6f0cdb3..d602bf3950 100644 --- a/README.md +++ b/README.md @@ -848,7 +848,7 @@ curl https://your-sub2api.example.com/v1/systemone \ --data '{"model":"jev-latest","state":"Text to evaluate","questions":{"safety":{"type":"noul","instructions":"Evaluate whether the text is unsafe"}}}' ``` -The built-in `jev-latest` price is `$0.042` per million input tokens and `$0` for output tokens. Channel pricing can override both values. Credential, rate-limit, overload, server, and network failures use the existing account cooldown and failover path; request errors (`400` and `422`) are returned without retrying another account. +The built-in `jev-latest` price is `$0.042` per million input tokens and `$0` for output tokens. Channel pricing can override both values. Credential, billing, permission, rate-limit, overload, server, and network failures (`401`, `402`, `403`, `429`, `529`, `5xx`, transport errors) use the existing account error policy (including custom error codes and temporary-unschedulable rules) and fail over to another account; request errors (`400` and `422`) are returned without retrying another account and never change account state. TypeSafe groups (and Composite requests routed to TypeSafe) reject Messages, Chat Completions, Responses, and count_tokens requests with `404`. --- diff --git a/README_CN.md b/README_CN.md index 4680298f82..8fe4bc5d83 100644 --- a/README_CN.md +++ b/README_CN.md @@ -764,7 +764,7 @@ curl https://your-sub2api.example.com/v1/systemone \ --data '{"model":"jev-latest","state":"待评估文本","questions":{"safety":{"type":"noul","instructions":"评估文本是否不安全"}}}' ``` -`jev-latest` 内置价格为输入 `$0.042/百万 tokens`、输出 `$0`，渠道定价可以覆盖。凭据、限流、过载、服务端和网络错误沿用现有账号冷却与故障切换链路；请求错误（`400`、`422`）不会切换账号重试。 +`jev-latest` 内置价格为输入 `$0.042/百万 tokens`、输出 `$0`，渠道定价可以覆盖。凭据、欠费、权限、限流、过载、服务端和网络错误（`401`、`402`、`403`、`429`、`529`、`5xx`、传输错误）沿用现有账号错误策略（含自定义错误码与临时不可调度规则）并切换账号；请求错误（`400`、`422`）不会切换账号重试，也不会改变账号状态。TypeSafe 分组（以及路由到 TypeSafe 的 Composite 请求）调用 Messages、Chat Completions、Responses、count_tokens 时返回 `404`。 --- diff --git a/backend/internal/handler/admin/account_handler.go b/backend/internal/handler/admin/account_handler.go index 3d0d2ec510..6a0ef4fdf4 100644 --- a/backend/internal/handler/admin/account_handler.go +++ b/backend/internal/handler/admin/account_handler.go @@ -27,6 +27,7 @@ import ( "github.com/Wei-Shaw/sub2api/internal/pkg/openai" "github.com/Wei-Shaw/sub2api/internal/pkg/response" "github.com/Wei-Shaw/sub2api/internal/pkg/timezone" + "github.com/Wei-Shaw/sub2api/internal/pkg/typesafe" "github.com/Wei-Shaw/sub2api/internal/pkg/xai" "github.com/Wei-Shaw/sub2api/internal/service" @@ -2950,6 +2951,12 @@ func (h *AccountHandler) GetAvailableModels(c *gin.Context) { return } + // TypeSafe accounts serve only the native System One model. + if account.IsTypeSafe() { + response.Success(c, []claude.Model{{ID: typesafe.JevLatestModel, Type: "model", DisplayName: typesafe.JevLatestModel}}) + return + } + // Handle Claude/Anthropic accounts // For OAuth and Setup-Token accounts: return default models if account.IsOAuth() { diff --git a/backend/internal/handler/admin/account_handler_available_models_test.go b/backend/internal/handler/admin/account_handler_available_models_test.go index 28dc6d1492..3bcfce11d0 100644 --- a/backend/internal/handler/admin/account_handler_available_models_test.go +++ b/backend/internal/handler/admin/account_handler_available_models_test.go @@ -526,3 +526,36 @@ func TestAccountHandlerSyncUpstreamModels_MetadataEnrichmentFailureReturnsWarnin require.Len(t, resp.Data.Warnings, 1) require.Equal(t, "upstream_model_metadata_incomplete", resp.Data.Warnings[0].Code) } + +func TestAccountHandlerGetAvailableModels_TypeSafeOnlyReturnsJev(t *testing.T) { + for _, credentials := range []map[string]any{ + {"api_key": "ts-secret"}, + {"api_key": "ts-secret", "model_mapping": map[string]any{"jev-latest": "jev-latest"}}, + } { + svc := &availableModelsAdminService{ + stubAdminService: newStubAdminService(), + account: service.Account{ + ID: 46, + Name: "typesafe", + Platform: service.PlatformTypeSafe, + Type: service.AccountTypeAPIKey, + Status: service.StatusActive, + Credentials: credentials, + }, + } + router := setupAvailableModelsRouter(svc) + + rec := httptest.NewRecorder() + router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/admin/accounts/46/models", nil)) + require.Equal(t, http.StatusOK, rec.Code) + + var resp struct { + Data []struct { + ID string `json:"id"` + } `json:"data"` + } + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &resp)) + require.Len(t, resp.Data, 1) + require.Equal(t, "jev-latest", resp.Data[0].ID) + } +} diff --git a/backend/internal/handler/gateway_handler.go b/backend/internal/handler/gateway_handler.go index a25a65b576..afd77297f3 100644 --- a/backend/internal/handler/gateway_handler.go +++ b/backend/internal/handler/gateway_handler.go @@ -223,6 +223,9 @@ func (h *GatewayHandler) Messages(c *gin.Context) { h.errorResponse(c, http.StatusBadRequest, "invalid_request_error", "Model is not supported by composite groups") return } + if rejectSystemOneOnlyPlatform(c, apiKey, h.errorResponse) { + return + } if decision := h.checkSecurityAudit(c, reqLog, apiKey, subject, service.ContentModerationProtocolAnthropicMessages, reqModel, body); decision != nil && !decision.AllowNextStage { h.anthropicSecurityAuditError(c, decision) @@ -2169,6 +2172,9 @@ func (h *GatewayHandler) CountTokens(c *gin.Context) { h.errorResponse(c, http.StatusBadRequest, "invalid_request_error", "Model is not supported by composite groups") return } + if rejectSystemOneOnlyPlatform(c, apiKey, h.errorResponse) { + return + } setOpsRequestContext(c, parsedReq.Model, parsedReq.Stream) setOpsEndpointContext(c, "", int16(service.RequestTypeFromLegacy(parsedReq.Stream, false))) diff --git a/backend/internal/handler/gateway_handler_chat_completions.go b/backend/internal/handler/gateway_handler_chat_completions.go index 7ede3bd3ee..561ed1bca0 100644 --- a/backend/internal/handler/gateway_handler_chat_completions.go +++ b/backend/internal/handler/gateway_handler_chat_completions.go @@ -82,6 +82,9 @@ func (h *GatewayHandler) ChatCompletions(c *gin.Context) { h.chatCompletionsErrorResponse(c, http.StatusBadRequest, "invalid_request_error", "Model is not supported by composite groups") return } + if rejectSystemOneOnlyPlatform(c, apiKey, h.chatCompletionsErrorResponse) { + return + } reqStream, ok := parseOpenAICompatibleStream(body) if !ok { h.chatCompletionsErrorResponse(c, http.StatusBadRequest, "invalid_request_error", invalidStreamFieldTypeMessage) diff --git a/backend/internal/handler/gateway_handler_responses.go b/backend/internal/handler/gateway_handler_responses.go index 7fdcd8e9d9..5ef6bace91 100644 --- a/backend/internal/handler/gateway_handler_responses.go +++ b/backend/internal/handler/gateway_handler_responses.go @@ -82,6 +82,9 @@ func (h *GatewayHandler) Responses(c *gin.Context) { h.responsesErrorResponse(c, http.StatusBadRequest, "invalid_request_error", "Model is not supported by composite groups") return } + if rejectSystemOneOnlyPlatform(c, apiKey, h.responsesErrorResponse) { + return + } reqStream, ok := parseOpenAICompatibleStream(body) if !ok { h.responsesErrorResponse(c, http.StatusBadRequest, "invalid_request_error", invalidStreamFieldTypeMessage) diff --git a/backend/internal/handler/gateway_systemone.go b/backend/internal/handler/gateway_systemone.go index 7c2713e79d..77ade84e95 100644 --- a/backend/internal/handler/gateway_systemone.go +++ b/backend/internal/handler/gateway_systemone.go @@ -16,6 +16,25 @@ import ( "go.uber.org/zap" ) +const systemOneOnlyPlatformMessage = "TypeSafe models are only available through POST /v1/systemone" + +// rejectSystemOneOnlyPlatform stops TypeSafe traffic from entering a non System +// One protocol chain. TypeSafe accounts only speak the native System One +// protocol; letting them reach the Anthropic/OpenAI converters would send +// foreign payloads (and the account key) to the wrong upstream path and feed +// the resulting auth failures back into account state. +func rejectSystemOneOnlyPlatform(c *gin.Context, apiKey *service.APIKey, writeError func(*gin.Context, int, string, string)) bool { + if _, forced := middleware2.GetForcePlatformFromContext(c); forced { + return false + } + if effectiveAPIKeyPlatform(c, apiKey) != service.PlatformTypeSafe { + return false + } + service.MarkOpsClientBusinessLimited(c, service.OpsClientBusinessLimitedReasonLocalFeatureGate) + writeError(c, http.StatusNotFound, "not_found_error", systemOneOnlyPlatformMessage) + return true +} + // SystemOne proxies TypeSafe's native, non-streaming System One protocol. func (h *GatewayHandler) SystemOne(c *gin.Context) { requestStart := time.Now() @@ -71,13 +90,17 @@ func (h *GatewayHandler) SystemOne(c *gin.Context) { streamStarted := false userRelease, err := h.concurrencyHelper.AcquireUserSlotWithWait(c, subject.UserID, subject.Concurrency, false, &streamStarted) if err != nil { + reqLog.Warn("systemone.user_slot_acquire_failed", zap.Error(err)) h.handleConcurrencyError(c, err, "user", false) return } + // 在请求结束或 Context 取消时确保释放槽位，避免客户端断开造成泄漏。 + userRelease = wrapReleaseOnDone(c.Request.Context(), userRelease) if userRelease != nil { defer userRelease() } if err := h.billingCacheService.CheckBillingEligibility(c.Request.Context(), apiKey.User, apiKey, apiKey.Group, subscription, service.QuotaPlatform(c.Request.Context(), apiKey)); err != nil { + reqLog.Info("systemone.billing_eligibility_check_failed", zap.Error(err)) status, code, message, retryAfter := billingErrorDetails(err) if retryAfter > 0 { c.Header("Retry-After", strconv.Itoa(retryAfter)) @@ -85,64 +108,122 @@ func (h *GatewayHandler) SystemOne(c *gin.Context) { h.errorResponse(c, status, code, message) return } - - failedAccountIDs := make(map[int64]struct{}) - var lastFailoverErr *service.UpstreamFailoverError - maxSwitches := h.maxAccountSwitches - if maxSwitches <= 0 { - maxSwitches = 3 + // 余额模式在途预留：防止并发请求在预检时看到同一份余额而集体透支。 + inflightRelease, err := reserveInflightBalance(c, h.billingCacheService, h.gatewayService, apiKey, subscription, tokenInflightEstimate(model, body)) + if err != nil { + reqLog.Info("systemone.inflight_reservation_rejected", zap.Error(err)) + status, code, message, retryAfter := billingErrorDetails(err) + if retryAfter > 0 { + c.Header("Retry-After", strconv.Itoa(retryAfter)) + } + h.errorResponse(c, status, code, message) + return } - for switchCount := 0; ; { - selection, err := h.gatewayService.SelectAccountWithLoadAwareness(c.Request.Context(), apiKey.GroupID, "", model, failedAccountIDs, "", subject.UserID) - if err != nil || selection == nil || selection.Account == nil { - if failoverClientGone(c) { - return - } - if lastFailoverErr != nil { - h.handleFailoverExhausted(c, lastFailoverErr, service.PlatformTypeSafe, false) - return - } - cls := classifyNoAccountErrorFromGin(c, h.gatewayService, apiKey, model, model, service.PlatformTypeSafe) - h.errorResponse(c, cls.Status, cls.ErrType, cls.Message) + defer inflightRelease() + + fs := NewFailoverState(h.maxAccountSwitches, false) + for { + if failoverClientGone(c) { return } + selection, err := h.gatewayService.SelectAccountWithLoadAwareness(c.Request.Context(), apiKey.GroupID, "", model, fs.FailedAccountIDs, "", subject.UserID) + if err == nil && (selection == nil || selection.Account == nil) { + err = service.ErrNoAvailableAccounts + } + if err != nil { + if failoverClientGone(c) { + reqLog.Info("systemone.account_select_aborted_client_disconnected", zap.Error(err)) + return + } + if len(fs.FailedAccountIDs) == 0 { + cls := classifyNoAccountErrorFromGin(c, h.gatewayService, apiKey, model, model, service.PlatformTypeSafe) + cls = classifySelectionFailureError(err, cls) + if !cls.ModelNotFound { + markOpsRoutingCapacityLimitedIfNoAvailable(c, err) + } + reqLog.Warn("systemone.select_account_no_available", zap.Bool("model_not_found", cls.ModelNotFound), zap.Error(err)) + h.errorResponse(c, cls.Status, cls.ErrType, cls.Message) + return + } + switch fs.HandleSelectionExhausted(c.Request.Context()) { + case FailoverContinue: + continue + case FailoverCanceled: + failoverClientGone(c) + return + default: + if fs.LastFailoverErr != nil { + h.handleFailoverExhausted(c, fs.LastFailoverErr, service.PlatformTypeSafe, false) + } else { + h.handleFailoverExhaustedSimple(c, http.StatusBadGateway, false) + } + return + } + } account := selection.Account + + accountRelease := selection.ReleaseFunc + if !selection.Acquired { + if selection.WaitPlan == nil { + markOpsRoutingCapacityLimited(c) + h.errorResponse(c, http.StatusServiceUnavailable, "api_error", "No available accounts") + return + } + accountRelease, err = h.concurrencyHelper.AcquireAccountSlotWithWaitTimeout(c, account.ID, selection.WaitPlan.MaxConcurrency, selection.WaitPlan.Timeout, false, &streamStarted) + if err != nil { + reqLog.Warn("systemone.account_slot_acquire_failed", zap.Int64("account_id", account.ID), zap.Error(err)) + h.handleConcurrencyError(c, err, "account", false) + return + } + } + // 准入终检：与其他网关入口一致，利润控制否决的账号不得承接本次请求。 + admissionCtx := service.ContextWithSelectionProfitGate(c.Request.Context(), selection) + latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(admissionCtx, account) + if vetoed { + if accountRelease != nil { + accountRelease() + } + reqLog.Debug("systemone.account_slot_profit_vetoed", zap.Int64("account_id", account.ID), zap.String("reason", reason)) + if fs.RecordProfitVeto(account.ID) == FailoverExhausted { + reqLog.Warn("systemone.profit_veto_attempts_exhausted", zap.Int("profit_veto_count", fs.ProfitVetoCount())) + h.errorResponse(c, http.StatusServiceUnavailable, "api_error", profitVetoExhaustedMessage) + return + } + continue + } + account = latest + accountRelease = wrapReleaseOnDone(c.Request.Context(), accountRelease) setOpsSelectedAccount(c, account.ID, account.Platform) service.SetOpsUpstreamModel(c, model) - accountRelease, err := h.acquireSystemOneAccountSlot(c, selection) - if err != nil { - h.handleConcurrencyError(c, err, "account", false) - return - } forwardStart := time.Now() - result, forwardErr := func() (*service.SystemOneForwardResult, error) { - if accountRelease != nil { - defer accountRelease() - } - return h.gatewayService.ForwardSystemOne(c.Request.Context(), c, account, body) - }() + result, forwardErr := h.gatewayService.ForwardSystemOne(c.Request.Context(), c, account, body) + if accountRelease != nil { + accountRelease() + } service.SetOpsLatencyMs(c, service.OpsResponseLatencyMsKey, time.Since(forwardStart).Milliseconds()) if forwardErr != nil { var failoverErr *service.UpstreamFailoverError if errors.As(forwardErr, &failoverErr) { - if failoverClientGone(c) { + switch fs.HandleFailoverError(c.Request.Context(), h.gatewayService, account.ID, account.Platform, account.GetPoolModeRetryCount(), failoverErr) { + case FailoverContinue: + reqLog.Warn("systemone.upstream_failover_switching", + zap.Int64("account_id", account.ID), + zap.Int("upstream_status", failoverErr.StatusCode), + zap.Int("switch_count", fs.SwitchCount), + ) + continue + case FailoverExhausted: + h.handleFailoverExhausted(c, fs.LastFailoverErr, service.PlatformTypeSafe, false) + return + case FailoverCanceled: + failoverClientGone(c) return } - failedAccountIDs[account.ID] = struct{}{} - lastFailoverErr = failoverErr - if switchCount >= maxSwitches { - h.handleFailoverExhausted(c, failoverErr, service.PlatformTypeSafe, false) - return - } - switchCount++ - reqLog.Warn("systemone.upstream_failover_switching", - zap.Int64("account_id", account.ID), - zap.Int("upstream_status", failoverErr.StatusCode), - zap.Int("switch_count", switchCount), - ) - continue + } + if failoverClientGone(c) { + return } var upstreamErr *service.SystemOneUpstreamError if errors.As(forwardErr, &upstreamErr) { @@ -154,6 +235,10 @@ func (h *GatewayHandler) SystemOne(c *gin.Context) { return } reqLog.Warn("systemone.forward_failed", zap.Int64("account_id", account.ID), zap.Error(forwardErr)) + if errors.Is(forwardErr, typesafe.ErrSystemOneResponseTooLarge) { + h.errorResponse(c, http.StatusBadGateway, "upstream_error", "TypeSafe response exceeds the gateway size limit") + return + } h.errorResponse(c, http.StatusBadGateway, "upstream_error", "TypeSafe upstream request failed") return } @@ -164,16 +249,6 @@ func (h *GatewayHandler) SystemOne(c *gin.Context) { } } -func (h *GatewayHandler) acquireSystemOneAccountSlot(c *gin.Context, selection *service.AccountSelectionResult) (func(), error) { - if selection.Acquired { - return selection.ReleaseFunc, nil - } - if selection.WaitPlan == nil { - return nil, errors.New("account concurrency unavailable") - } - return h.concurrencyHelper.AcquireAccountSlotWithWaitTimeout(c, selection.Account.ID, selection.WaitPlan.MaxConcurrency, selection.WaitPlan.Timeout, false, new(bool)) -} - func (h *GatewayHandler) recordSystemOneUsage(c *gin.Context, apiKey *service.APIKey, account *service.Account, subscription *service.UserSubscription, mapping service.ChannelMappingResult, model string, body []byte, result *service.SystemOneForwardResult, userID int64, pricingAt time.Time) { userAgent := c.GetHeader("User-Agent") clientIP := ip.GetClientIP(c) diff --git a/backend/internal/handler/gateway_systemone_test.go b/backend/internal/handler/gateway_systemone_test.go index bf60bf4990..084b37ed9f 100644 --- a/backend/internal/handler/gateway_systemone_test.go +++ b/backend/internal/handler/gateway_systemone_test.go @@ -41,3 +41,69 @@ func TestSystemOneRejectsNonTypeSafeGroupBeforeScheduling(t *testing.T) { require.Equal(t, http.StatusNotFound, recorder.Code) require.Contains(t, recorder.Body.String(), "only available for TypeSafe") } + +func newTypeSafeGroupContext(t *testing.T, path, body, groupPlatform string) (*gin.Context, *httptest.ResponseRecorder) { + t.Helper() + c, recorder := newSystemOneHandlerContext(body) + c.Request = httptest.NewRequest(http.MethodPost, path, strings.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + groupID := int64(9) + c.Set(string(middleware2.ContextKeyAPIKey), &service.APIKey{ID: 4, UserID: 5, GroupID: &groupID, Group: &service.Group{ID: groupID, Platform: groupPlatform}}) + c.Set(string(middleware2.ContextKeyUser), middleware2.AuthSubject{UserID: 5, Concurrency: 1}) + return c, recorder +} + +func TestRejectSystemOneOnlyPlatform(t *testing.T) { + write := func(c *gin.Context, status int, errType, message string) { + c.JSON(status, gin.H{"type": errType, "message": message}) + } + + c, recorder := newTypeSafeGroupContext(t, "/v1/messages", `{}`, service.PlatformTypeSafe) + require.True(t, rejectSystemOneOnlyPlatform(c, mustAPIKey(t, c), write)) + require.Equal(t, http.StatusNotFound, recorder.Code) + require.Contains(t, recorder.Body.String(), systemOneOnlyPlatformMessage) + + c, _ = newTypeSafeGroupContext(t, "/v1/messages", `{}`, service.PlatformComposite) + require.False(t, rejectSystemOneOnlyPlatform(c, mustAPIKey(t, c), write)) + c.Request = c.Request.WithContext(service.WithResolvedTargetPlatform(c.Request.Context(), service.PlatformTypeSafe)) + require.True(t, rejectSystemOneOnlyPlatform(c, mustAPIKey(t, c), write)) + + c, _ = newTypeSafeGroupContext(t, "/v1/messages", `{}`, service.PlatformAnthropic) + require.False(t, rejectSystemOneOnlyPlatform(c, mustAPIKey(t, c), write)) + + c, _ = newTypeSafeGroupContext(t, "/antigravity/v1/messages", `{}`, service.PlatformTypeSafe) + c.Set(string(middleware2.ContextKeyForcePlatform), service.PlatformAntigravity) + require.False(t, rejectSystemOneOnlyPlatform(c, mustAPIKey(t, c), write)) +} + +func mustAPIKey(t *testing.T, c *gin.Context) *service.APIKey { + t.Helper() + apiKey, ok := middleware2.GetAPIKeyFromContext(c) + require.True(t, ok) + return apiKey +} + +func TestTypeSafeGroupsRejectNonSystemOneProtocolsBeforeScheduling(t *testing.T) { + h := &GatewayHandler{ + cfg: &config.Config{Gateway: config.GatewayConfig{MaxBodySize: 1 << 20}}, + gatewayService: &service.GatewayService{}, + } + for _, tc := range []struct { + name string + path string + body string + handler func(*gin.Context) + }{ + {"messages", "/v1/messages", `{"model":"jev-latest","max_tokens":1,"messages":[{"role":"user","content":"hi"}]}`, h.Messages}, + {"count_tokens", "/v1/messages/count_tokens", `{"model":"jev-latest","messages":[{"role":"user","content":"hi"}]}`, h.CountTokens}, + {"chat_completions", "/v1/chat/completions", `{"model":"jev-latest","messages":[{"role":"user","content":"hi"}]}`, h.ChatCompletions}, + {"responses", "/v1/responses", `{"model":"jev-latest","input":"hi"}`, h.Responses}, + } { + t.Run(tc.name, func(t *testing.T) { + c, recorder := newTypeSafeGroupContext(t, tc.path, tc.body, service.PlatformTypeSafe) + tc.handler(c) + require.Equal(t, http.StatusNotFound, recorder.Code, recorder.Body.String()) + require.Contains(t, recorder.Body.String(), systemOneOnlyPlatformMessage) + }) + } +} diff --git a/backend/internal/pkg/typesafe/client.go b/backend/internal/pkg/typesafe/client.go index 3c66d8c314..7acbd9b4a4 100644 --- a/backend/internal/pkg/typesafe/client.go +++ b/backend/internal/pkg/typesafe/client.go @@ -61,9 +61,21 @@ func NewSystemOneRequest(ctx context.Context, baseURL, key string, body []byte) return req, nil } +// MaxSystemOneResponseBytes bounds a buffered System One response body. +const MaxSystemOneResponseBytes = 4 << 20 + +var ErrSystemOneResponseTooLarge = errors.New("typesafe response exceeds size limit") + func DecodeSystemOneResponse(r io.Reader) (*SystemOneResponse, error) { - body, err := io.ReadAll(io.LimitReader(r, 4<<20)) - if err != nil || !json.Valid(body) { + body, err := io.ReadAll(io.LimitReader(r, MaxSystemOneResponseBytes+1)) + if err != nil { + return nil, errors.New("typesafe invalid response") + } + // A truncated body would otherwise surface as a misleading "invalid JSON". + if len(body) > MaxSystemOneResponseBytes { + return nil, ErrSystemOneResponseTooLarge + } + if !json.Valid(body) { return nil, errors.New("typesafe invalid response") } var envelope struct { diff --git a/backend/internal/pkg/typesafe/systemone.go b/backend/internal/pkg/typesafe/systemone.go index 898a0e5c24..14093b2e24 100644 --- a/backend/internal/pkg/typesafe/systemone.go +++ b/backend/internal/pkg/typesafe/systemone.go @@ -11,10 +11,10 @@ import ( var ErrStreamingUnsupported = errors.New("typesafe system one does not support streaming") type systemOneEnvelope struct { - Model json.RawMessage `json:"model"` - State json.RawMessage `json:"state"` - Questions map[string]json.RawMessage `json:"questions"` - Stream json.RawMessage `json:"stream"` + Model json.RawMessage `json:"model"` + State json.RawMessage `json:"state"` + Questions json.RawMessage `json:"questions"` + Stream json.RawMessage `json:"stream"` } func ValidateSystemOneRequest(body []byte) (string, error) { @@ -33,7 +33,9 @@ func ValidateSystemOneRequest(body []byte) (string, error) { if err := validateStringObjectOrArray(envelope.State, "state"); err != nil { return "", err } - if len(envelope.Questions) == 0 { + questionsRaw := bytes.TrimSpace(envelope.Questions) + var questions map[string]json.RawMessage + if len(questionsRaw) == 0 || questionsRaw[0] != '{' || json.Unmarshal(questionsRaw, &questions) != nil || len(questions) == 0 { return "", errors.New("questions must be a non-empty object") } if len(envelope.Stream) > 0 && string(envelope.Stream) != "null" { @@ -45,7 +47,7 @@ func ValidateSystemOneRequest(body []byte) (string, error) { return "", ErrStreamingUnsupported } } - for id, raw := range envelope.Questions { + for id, raw := range questions { if err := validateQuestion(id, raw); err != nil { return "", err } diff --git a/backend/internal/pkg/typesafe/systemone_test.go b/backend/internal/pkg/typesafe/systemone_test.go index c4233a97f0..3f6d3d9c00 100644 --- a/backend/internal/pkg/typesafe/systemone_test.go +++ b/backend/internal/pkg/typesafe/systemone_test.go @@ -46,6 +46,8 @@ func TestValidateSystemOneRequestRejectsInvalidRequests(t *testing.T) { {"missing state", `{"model":"jev-latest","questions":{"q":{"type":"noul","instructions":"x"}}}`, "state"}, {"scalar state", `{"model":"jev-latest","state":42,"questions":{"q":{"type":"noul","instructions":"x"}}}`, "state"}, {"empty questions", `{"model":"jev-latest","state":"x","questions":{}}`, "questions"}, + {"array questions", `{"model":"jev-latest","state":"x","questions":[{"type":"noul"}]}`, "questions must be a non-empty object"}, + {"null questions", `{"model":"jev-latest","state":"x","questions":null}`, "questions must be a non-empty object"}, {"unknown question type", `{"model":"jev-latest","state":"x","questions":{"q":{"type":"boolean","instructions":"x"}}}`, "unsupported type"}, {"question type with whitespace", `{"model":"jev-latest","state":"x","questions":{"q":{"type":" noul ","instructions":"x"}}}`, "unsupported type"}, {"noul criteria array", `{"model":"jev-latest","state":"x","questions":{"q":{"type":"noul","instructions":"x","criteria":[]}}}`, "noul criteria"}, diff --git a/backend/internal/securityaudit/prompt_snapshot.go b/backend/internal/securityaudit/prompt_snapshot.go index b47e846741..12c1630b50 100644 --- a/backend/internal/securityaudit/prompt_snapshot.go +++ b/backend/internal/securityaudit/prompt_snapshot.go @@ -106,6 +106,8 @@ func extractProtocolSegments(protocol string, document any) []promptSegment { return append(extractInstructions(root["instructions"]), extractResponses(root["input"])...) case "openai_images", "grok_media", "media", "images": return userPromptSegments(extractMediaPrompts(root)) + case "typesafe_systemone": + return extractSystemOneSegments(root) default: if segments := extractChatLikeSegments(root); len(segments) > 0 { return segments @@ -120,6 +122,64 @@ func extractProtocolSegments(protocol string, document any) []promptSegment { } } +// extractSystemOneSegments collects every client-controlled text of a TypeSafe +// System One request. Question instructions and criteria come first and the +// evaluated state last, so the state is the prioritized segment. Object keys are +// visited in sorted order to keep the prompt hash stable across requests. +func extractSystemOneSegments(root map[string]any) []promptSegment { + if root == nil { + return nil + } + texts := make([]string, 0, 4) + questions, _ := root["questions"].(map[string]any) + for _, id := range sortedJSONKeys(questions) { + question, ok := questions[id].(map[string]any) + if !ok { + continue + } + texts = appendJSONStringLeaves(texts, question["instructions"]) + criteria, isObject := question["criteria"].(map[string]any) + if !isObject || stringValue(question["type"]) != "choice" { + texts = appendJSONStringLeaves(texts, question["criteria"]) + continue + } + // Choice option labels are client-chosen text sent to the model as well. + for _, label := range sortedJSONKeys(criteria) { + texts = appendJSONStringLeaves(texts, label) + texts = appendJSONStringLeaves(texts, criteria[label]) + } + } + texts = appendJSONStringLeaves(texts, root["state"]) + return userPromptSegments(texts) +} + +func appendJSONStringLeaves(texts []string, value any) []string { + switch typed := value.(type) { + case string: + if text := strings.TrimSpace(typed); text != "" { + texts = append(texts, text) + } + case []any: + for _, item := range typed { + texts = appendJSONStringLeaves(texts, item) + } + case map[string]any: + for _, key := range sortedJSONKeys(typed) { + texts = appendJSONStringLeaves(texts, typed[key]) + } + } + return texts +} + +func sortedJSONKeys(values map[string]any) []string { + keys := make([]string, 0, len(values)) + for key := range values { + keys = append(keys, key) + } + sort.Strings(keys) + return keys +} + // clientInstructionRoles are roles a client may freely populate. Attackers can // place jailbreak/PII text in assistant/tool turns, so blocking audit must scan // them too—not only user/system/developer instructions. diff --git a/backend/internal/securityaudit/prompt_snapshot_test.go b/backend/internal/securityaudit/prompt_snapshot_test.go index 70d26f925c..317cd98df7 100644 --- a/backend/internal/securityaudit/prompt_snapshot_test.go +++ b/backend/internal/securityaudit/prompt_snapshot_test.go @@ -412,3 +412,42 @@ func mustJSON(t *testing.T, value string) []byte { func metadataTextForTest(scanText string) string { return strings.Replace(scanText, promptAuditPrioritySeparator, "\n\n", 1) } + +func TestPromptSnapshotTypeSafeSystemOneCollectsStateAndQuestions(t *testing.T) { + body := `{"model":"jev-latest","state":{"title":"STATE_TITLE","items":["STATE_ITEM",{"text":"STATE_NESTED"},3]},` + + `"questions":{"b":{"type":"choice","instructions":"CHOICE_INSTRUCTIONS","criteria":{"OPTION_LABEL":{"description":"OPTION_DESC"},"empty":null}},` + + `"a":{"type":"noul","instructions":["NOUL_INSTRUCTIONS"],"criteria":{"true":"NOUL_TRUE","false":"NOUL_FALSE"}},` + + `"c":{"type":"score","criteria":["SCORE_LOW",{"description":"SCORE_HIGH"}]}}}` + + snapshot, err := ExtractPromptSnapshot(Request{Protocol: "typesafe_systemone", Body: []byte(body)}) + require.NoError(t, err) + for _, text := range []string{"STATE_TITLE", "STATE_ITEM", "STATE_NESTED", "CHOICE_INSTRUCTIONS", "OPTION_LABEL", "OPTION_DESC", + "NOUL_INSTRUCTIONS", "NOUL_TRUE", "NOUL_FALSE", "SCORE_LOW", "SCORE_HIGH"} { + require.Contains(t, snapshot.ScanText, text) + } + require.NotContains(t, snapshot.ScanText, "jev-latest") + require.Equal(t, 12, snapshot.MessageCount) + + // Map iteration order must not change the audited text or its hash. + for range 20 { + again, err := ExtractPromptSnapshot(Request{Protocol: "typesafe_systemone", Body: []byte(body)}) + require.NoError(t, err) + require.Equal(t, snapshot.PromptHash, again.PromptHash) + require.Equal(t, snapshot.ScanText, again.ScanText) + } + + blocking, err := ExtractBlockingPromptSnapshot(Request{Protocol: "typesafe_systemone", Body: []byte(body)}, true) + require.NoError(t, err) + require.Contains(t, blocking.ScanText, "STATE_TITLE") + require.Contains(t, blocking.ScanText, "CHOICE_INSTRUCTIONS") +} + +func TestPromptSnapshotTypeSafeSystemOneStringStateIsAudited(t *testing.T) { + snapshot, err := ExtractPromptSnapshot(Request{Protocol: "typesafe_systemone", Body: []byte(`{"model":"jev-latest","state":"plain state","questions":{"q":{"type":"noul"}}}`)}) + require.NoError(t, err) + require.True(t, strings.HasPrefix(snapshot.ScanText, "plain state")) + require.Equal(t, 1, snapshot.MessageCount) + + _, err = ExtractPromptSnapshot(Request{Protocol: "typesafe_systemone", Body: []byte(`{"model":"jev-latest","state":{"n":1},"questions":{"q":{"type":"noul"}}}`)}) + require.ErrorIs(t, err, ErrNoPromptText) +} diff --git a/backend/internal/service/account.go b/backend/internal/service/account.go index 38efbfba86..ca8246677e 100644 --- a/backend/internal/service/account.go +++ b/backend/internal/service/account.go @@ -985,6 +985,10 @@ func (a *Account) GetBaseURL() string { } baseURL := a.GetCredential("base_url") if baseURL == "" { + // TypeSafe keys must never fall back to the Anthropic host. + if a.Platform == PlatformTypeSafe { + return typesafe.DefaultBaseURL + } return "https://api.anthropic.com" } if a.Platform == PlatformAntigravity { diff --git a/backend/internal/service/account_test_service.go b/backend/internal/service/account_test_service.go index 7d16eeaaec..4da30d610b 100644 --- a/backend/internal/service/account_test_service.go +++ b/backend/internal/service/account_test_service.go @@ -416,6 +416,10 @@ func (s *AccountTestService) TestAccountConnection(c *gin.Context, accountID int return s.testOpenCodeGoAccountConnection(c, account, modelID, prompt) } + if account.IsTypeSafe() { + return s.testTypeSafeAccountConnection(c, account, prompt) + } + return s.testClaudeAccountConnection(c, account, modelID) } diff --git a/backend/internal/service/account_test_service_typesafe.go b/backend/internal/service/account_test_service_typesafe.go new file mode 100644 index 0000000000..6c9501d892 --- /dev/null +++ b/backend/internal/service/account_test_service_typesafe.go @@ -0,0 +1,95 @@ +package service + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + + "github.com/Wei-Shaw/sub2api/internal/pkg/typesafe" + "github.com/gin-gonic/gin" +) + +const ( + typeSafeTestDefaultState = "Sub2API connection test" + typeSafeTestQuestionID = "connection_test" + typeSafeTestMaxPreviewBytes = 2000 +) + +// testTypeSafeAccountConnection probes a TypeSafe account with a minimal native +// System One request. TypeSafe accounts never speak the Claude protocol, so +// they must not fall through to testClaudeAccountConnection (which would send +// the key to /v1/messages and could misclassify the account). +func (s *AccountTestService) testTypeSafeAccountConnection(c *gin.Context, account *Account, prompt string) error { + ctx := c.Request.Context() + if account.Type != AccountTypeAPIKey { + return s.sendErrorAndEnd(c, fmt.Sprintf("Unsupported account type: %s", account.Type)) + } + apiKey := account.GetTypeSafeAPIKey() + if apiKey == "" { + return s.sendErrorAndEnd(c, "No API key available") + } + baseURL, err := s.validateUpstreamBaseURL(account.GetTypeSafeBaseURL()) + if err != nil { + return s.sendErrorAndEnd(c, fmt.Sprintf("Invalid base URL: %s", err.Error())) + } + + state := strings.TrimSpace(prompt) + if state == "" { + state = typeSafeTestDefaultState + } + payload, err := json.Marshal(map[string]any{ + "model": typesafe.JevLatestModel, + "state": state, + "questions": map[string]any{ + typeSafeTestQuestionID: map[string]any{ + "type": "noul", + "instructions": "Is this text a connection test?", + }, + }, + }) + if err != nil { + return s.sendErrorAndEnd(c, "Failed to create test payload") + } + + c.Writer.Header().Set("Content-Type", "text/event-stream") + c.Writer.Header().Set("Cache-Control", "no-cache") + c.Writer.Header().Set("Connection", "keep-alive") + c.Writer.Header().Set("X-Accel-Buffering", "no") + c.Writer.Flush() + + s.sendEvent(c, TestEvent{Type: "test_start", Model: typesafe.JevLatestModel}) + + req, err := typesafe.NewSystemOneRequest(ctx, baseURL, apiKey, payload) + if err != nil { + return s.sendErrorAndEnd(c, "Failed to create request") + } + proxyURL := "" + if account.ProxyID != nil && account.Proxy != nil { + proxyURL = account.Proxy.URL() + } + resp, err := s.httpUpstream.Do(req, proxyURL, account.ID, account.Concurrency) + if err != nil { + return s.sendErrorAndEnd(c, fmt.Sprintf("Request failed: %s", sanitizeUpstreamErrorMessage(err.Error()))) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10)) + errMsg := fmt.Sprintf("API returned %d: %s", resp.StatusCode, truncateString(string(body), typeSafeTestMaxPreviewBytes)) + // 401/403 表示 API Key 无效或被上游拒绝，标记为 error 状态。 + if (resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden) && s.accountRepo != nil { + _ = s.accountRepo.SetError(ctx, account.ID, errMsg) + } + return s.sendErrorAndEnd(c, errMsg) + } + + decoded, err := typesafe.DecodeSystemOneResponse(resp.Body) + if err != nil { + return s.sendErrorAndEnd(c, fmt.Sprintf("Invalid System One response: %s", err.Error())) + } + s.sendEvent(c, TestEvent{Type: "content", Text: truncateString(string(decoded.Body), typeSafeTestMaxPreviewBytes)}) + s.sendEvent(c, TestEvent{Type: "test_complete", Success: true}) + return nil +} diff --git a/backend/internal/service/account_test_service_typesafe_test.go b/backend/internal/service/account_test_service_typesafe_test.go new file mode 100644 index 0000000000..c60abf42c3 --- /dev/null +++ b/backend/internal/service/account_test_service_typesafe_test.go @@ -0,0 +1,92 @@ +package service + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/Wei-Shaw/sub2api/internal/pkg/typesafe" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +func newTypeSafeAccountTestFixture(t *testing.T, status int, responseBody string) (*AccountTestService, *systemOnePolicyAccountRepo, *[]*http.Request, *gin.Context, *httptest.ResponseRecorder) { + t.Helper() + account := &Account{ID: 31, Platform: PlatformTypeSafe, Type: AccountTypeAPIKey, Credentials: map[string]any{"api_key": "ts-secret"}} + repo := &systemOnePolicyAccountRepo{account: account} + var requests []*http.Request + upstream := &systemOneHTTPUpstream{do: func(req *http.Request) (*http.Response, error) { + requests = append(requests, req) + return &http.Response{StatusCode: status, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(responseBody))}, nil + }} + svc := &AccountTestService{ + accountRepo: repo, + httpUpstream: upstream, + cfg: &config.Config{}, + } + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/api/v1/admin/accounts/31/test", nil) + return svc, repo, &requests, c, rec +} + +func TestTypeSafeAccountTestSendsNativeSystemOneRequest(t *testing.T) { + svc, repo, requests, c, rec := newTypeSafeAccountTestFixture(t, http.StatusOK, `{"model":"jev-1.13.0","answers":{"connection_test":{"type":"noul"}},"usage":{"input_tokens":9}}`) + + // The Claude default model must be ignored: TypeSafe only serves jev-latest. + require.NoError(t, svc.TestAccountConnection(c, 31, "claude-sonnet-4-5", "", AccountTestModeDefault)) + + require.Len(t, *requests, 1) + req := (*requests)[0] + require.Equal(t, typesafe.DefaultBaseURL+typesafe.SystemOnePath, req.URL.String()) + require.Equal(t, "Bearer ts-secret", req.Header.Get("Authorization")) + require.Empty(t, req.Header.Get("x-api-key")) + body, err := io.ReadAll(req.Body) + require.NoError(t, err) + model, err := typesafe.ValidateSystemOneRequest(body) + require.NoError(t, err) + require.Equal(t, typesafe.JevLatestModel, model) + + output := rec.Body.String() + require.Contains(t, output, `"type":"test_start"`) + require.Contains(t, output, `"model":"jev-latest"`) + require.Contains(t, output, `"type":"test_complete"`) + require.Contains(t, output, "jev-1.13.0") + require.Zero(t, repo.errorCalls) +} + +func TestTypeSafeAccountTestMarksRejectedKey(t *testing.T) { + for _, status := range []int{http.StatusUnauthorized, http.StatusForbidden} { + t.Run(http.StatusText(status), func(t *testing.T) { + svc, repo, _, c, rec := newTypeSafeAccountTestFixture(t, status, `{"detail":"invalid key"}`) + require.Error(t, svc.TestAccountConnection(c, 31, "", "", AccountTestModeDefault)) + require.Equal(t, 1, repo.errorCalls) + responseText, errMsg := parseTestSSEOutput(rec.Body.String()) + require.Empty(t, responseText) + require.Contains(t, errMsg, "API returned") + }) + } +} + +func TestTypeSafeAccountTestTransientFailureKeepsAccountState(t *testing.T) { + svc, repo, _, c, _ := newTypeSafeAccountTestFixture(t, http.StatusServiceUnavailable, `{"detail":"busy"}`) + require.Error(t, svc.TestAccountConnection(c, 31, "", "", AccountTestModeDefault)) + require.Zero(t, repo.errorCalls) +} + +func TestTypeSafeAccountTestUsesPromptAsState(t *testing.T) { + svc, _, requests, c, _ := newTypeSafeAccountTestFixture(t, http.StatusOK, `{"answers":{}}`) + require.NoError(t, svc.TestAccountConnection(c, 31, "", "custom state", AccountTestModeDefault)) + body, err := io.ReadAll((*requests)[0].Body) + require.NoError(t, err) + var payload struct { + State string `json:"state"` + } + require.NoError(t, json.Unmarshal(body, &payload)) + require.Equal(t, "custom state", payload.State) +} diff --git a/backend/internal/service/content_moderation_input.go b/backend/internal/service/content_moderation_input.go index 7a03288d4e..6ef090a100 100644 --- a/backend/internal/service/content_moderation_input.go +++ b/backend/internal/service/content_moderation_input.go @@ -47,7 +47,9 @@ func extractContentModerationInput(protocol string, body []byte, filterReminders collector.addModerationText(&parts, gjson.GetBytes(body, "prompt").String()) collector.collectContentValue(gjson.GetBytes(body, "images"), &parts, &images) case ContentModerationProtocolTypeSafeSystemOne: - collector.collectSystemOneState(gjson.GetBytes(body, "state"), &parts) + // System One carries no client-harness reminder blocks, so a literal + // is ordinary user text and must never be skipped. + moderationTextCollector{}.collectSystemOneInput(body, &parts) default: collector.collectLastResponsesInput(gjson.GetBytes(body, "input"), &parts, &images) collector.collectLastRoleMessage(gjson.GetBytes(body, "messages"), "user", &parts, &images) @@ -63,7 +65,28 @@ func extractContentModerationInput(protocol string, body []byte, filterReminders return out } -func (collector moderationTextCollector) collectSystemOneState(value gjson.Result, parts *[]string) { +// collectSystemOneInput moderates the evaluated state plus every question's +// instructions and criteria: all of them are client text sent to Jev. +func (collector moderationTextCollector) collectSystemOneInput(body []byte, parts *[]string) { + collector.collectSystemOneText(gjson.GetBytes(body, "state"), parts) + gjson.GetBytes(body, "questions").ForEach(func(_, question gjson.Result) bool { + collector.collectSystemOneText(question.Get("instructions"), parts) + criteria := question.Get("criteria") + if question.Get("type").String() == "choice" && criteria.IsObject() { + // Choice option labels are client-chosen text as well. + criteria.ForEach(func(label, description gjson.Result) bool { + collector.addModerationText(parts, label.String()) + collector.collectSystemOneText(description, parts) + return true + }) + return true + } + collector.collectSystemOneText(criteria, parts) + return true + }) +} + +func (collector moderationTextCollector) collectSystemOneText(value gjson.Result, parts *[]string) { switch { case !value.Exists(): return @@ -71,7 +94,7 @@ func (collector moderationTextCollector) collectSystemOneState(value gjson.Resul collector.addModerationText(parts, value.String()) case value.IsArray(), value.IsObject(): value.ForEach(func(_, child gjson.Result) bool { - collector.collectSystemOneState(child, parts) + collector.collectSystemOneText(child, parts) return true }) } diff --git a/backend/internal/service/content_moderation_systemone_test.go b/backend/internal/service/content_moderation_systemone_test.go index 993a71fa7e..c9a6ae9505 100644 --- a/backend/internal/service/content_moderation_systemone_test.go +++ b/backend/internal/service/content_moderation_systemone_test.go @@ -15,6 +15,7 @@ func TestExtractContentModerationInputTypeSafeSystemOne(t *testing.T) { {"string", `{"state":"plain text"}`, "plain text"}, {"object", `{"state":{"title":"hello","nested":{"body":"world"}}}`, "hello world"}, {"array", `{"state":["first",{"text":"second"},3]}`, "first second"}, + {"questions", `{"state":"state text","questions":{"c":{"type":"choice","instructions":"pick one","criteria":{"label a":{"description":"desc a"},"label b":null}},"s":{"type":"score","criteria":["low",{"text":"high"}]},"n":{"type":"noul","instructions":["judge"],"criteria":{"true":"yes"}}}}`, "state text pick one label a desc a label b low high judge yes"}, } { t.Run(tc.name, func(t *testing.T) { input := ExtractContentModerationInput(ContentModerationProtocolTypeSafeSystemOne, []byte(tc.body)) @@ -23,3 +24,10 @@ func TestExtractContentModerationInputTypeSafeSystemOne(t *testing.T) { }) } } + +func TestExtractContentModerationInputTypeSafeSystemOneKeepsReminderText(t *testing.T) { + body := `{"state":"hidden payload","questions":{"q":{"type":"noul","instructions":"hidden instructions"}}}` + input := ExtractContentModerationInput(ContentModerationProtocolTypeSafeSystemOne, []byte(body)) + require.Contains(t, input.Text, "hidden payload") + require.Contains(t, input.Text, "hidden instructions") +} diff --git a/backend/internal/service/gateway_systemone.go b/backend/internal/service/gateway_systemone.go index f9201f91e8..261cb74b9e 100644 --- a/backend/internal/service/gateway_systemone.go +++ b/backend/internal/service/gateway_systemone.go @@ -4,7 +4,7 @@ import ( "context" "errors" "fmt" - "io" + "mime" "net/http" "strings" "time" @@ -47,6 +47,7 @@ func (s *GatewayService) ForwardSystemOne(ctx context.Context, c *gin.Context, a if err != nil { return nil, err } + upstreamURL := req.URL.Scheme + "://" + req.URL.Host + req.URL.Path proxyURL := "" if account.ProxyID != nil && account.Proxy != nil { proxyURL = account.Proxy.URL() @@ -55,45 +56,19 @@ func (s *GatewayService) ForwardSystemOne(ctx context.Context, c *gin.Context, a if err != nil { return nil, s.handleUpstreamTransportError(ctx, c, account, err, OpsUpstreamErrorEvent{ Passthrough: true, - UpstreamURL: req.URL.Scheme + "://" + req.URL.Host + req.URL.Path, + UpstreamURL: upstreamURL, }) } defer func() { _ = resp.Body.Close() }() if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { - _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 64<<10)) - if resp.StatusCode == http.StatusBadRequest || resp.StatusCode == http.StatusUnprocessableEntity { - return nil, &SystemOneUpstreamError{StatusCode: resp.StatusCode} - } - if resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode == 529 || resp.StatusCode >= 500 { - if s.rateLimitService != nil { - s.rateLimitService.HandleUpstreamError(ctx, account, resp.StatusCode, resp.Header, nil, typesafe.JevLatestModel) - } - failoverErr := &UpstreamFailoverError{ - StatusCode: resp.StatusCode, - ResponseHeaders: resp.Header.Clone(), - ClientStatusCode: http.StatusBadGateway, - ClientMessage: "TypeSafe upstream request failed", - } - if resp.StatusCode == http.StatusUnauthorized { - failoverErr.Stage = GatewayFailureStageAccountAuth - failoverErr.Scope = GatewayFailureScopeAccount - failoverErr.Reason = TypeSafeCredentialRejectedReason - failoverErr.NextAccountAction = NextAccountRetry - } - return nil, failoverErr - } - return nil, &SystemOneUpstreamError{StatusCode: http.StatusBadGateway} + return nil, s.handleSystemOneErrorResponse(ctx, c, account, resp, upstreamURL) } decoded, err := typesafe.DecodeSystemOneResponse(resp.Body) if err != nil { return nil, err } - contentType := strings.TrimSpace(resp.Header.Get("Content-Type")) - if contentType == "" { - contentType = "application/json" - } return &SystemOneForwardResult{ ForwardResult: ForwardResult{ RequestID: resp.Header.Get("x-request-id"), @@ -105,6 +80,76 @@ func (s *GatewayService) ForwardSystemOne(ctx context.Context, c *gin.Context, a }, StatusCode: resp.StatusCode, Body: decoded.Body, - ContentType: contentType, + ContentType: systemOneResponseContentType(resp.Header.Get("Content-Type")), }, nil } + +// handleSystemOneErrorResponse applies the shared account error policy to a +// non-2xx System One response. 400/422 describe the caller's own payload, so +// they never touch account state (a tenant must not be able to disable an +// account with bad input) and are not retried elsewhere. Every other status +// goes through the account error policy (custom error codes, temporary +// unschedulable rules, pool mode) and fails over when the status is retryable +// or the policy took the account out of rotation. +func (s *GatewayService) handleSystemOneErrorResponse(ctx context.Context, c *gin.Context, account *Account, resp *http.Response, upstreamURL string) error { + respBody, _ := s.readUpstreamErrorBody(resp) + upstreamMsg := sanitizeUpstreamErrorMessage(strings.TrimSpace(extractUpstreamErrorMessage(respBody))) + setOpsUpstreamError(c, resp.StatusCode, upstreamMsg, "") + event := OpsUpstreamErrorEvent{ + Passthrough: true, + ProxyID: opsUpstreamProxyID(account), + ProxyName: opsUpstreamProxyName(account), + Platform: account.Platform, + AccountID: account.ID, + AccountName: account.Name, + UpstreamStatusCode: resp.StatusCode, + UpstreamRequestID: resp.Header.Get("x-request-id"), + UpstreamURL: upstreamURL, + Kind: "http_error", + Message: upstreamMsg, + } + + if resp.StatusCode == http.StatusBadRequest || resp.StatusCode == http.StatusUnprocessableEntity { + appendOpsUpstreamError(c, event) + return &SystemOneUpstreamError{StatusCode: resp.StatusCode} + } + + shouldDisable := false + if s.rateLimitService != nil { + shouldDisable = s.rateLimitService.HandleUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody, typesafe.JevLatestModel) + } + if !shouldDisable && !s.shouldFailoverUpstreamError(resp.StatusCode) { + appendOpsUpstreamError(c, event) + return &SystemOneUpstreamError{StatusCode: resp.StatusCode} + } + + event.Kind = "failover" + appendOpsUpstreamError(c, event) + failoverErr := &UpstreamFailoverError{ + StatusCode: resp.StatusCode, + ResponseBody: respBody, + ResponseHeaders: resp.Header.Clone(), + RetryableOnSameAccount: !shouldDisable && account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), + } + if resp.StatusCode == http.StatusUnauthorized { + failoverErr.Stage = GatewayFailureStageAccountAuth + failoverErr.Scope = GatewayFailureScopeAccount + failoverErr.Reason = TypeSafeCredentialRejectedReason + failoverErr.NextAccountAction = NextAccountRetry + } + return failoverErr +} + +// systemOneResponseContentType keeps the upstream JSON media type (and its +// charset) but never relays a non-JSON type for a body already validated as +// JSON, so the gateway origin cannot be made to serve it as HTML. +func systemOneResponseContentType(raw string) string { + mediaType, _, err := mime.ParseMediaType(strings.TrimSpace(raw)) + if err != nil { + return "application/json" + } + if mediaType == "application/json" || (strings.HasPrefix(mediaType, "application/") && strings.HasSuffix(mediaType, "+json")) { + return strings.TrimSpace(raw) + } + return "application/json" +} diff --git a/backend/internal/service/gateway_systemone_test.go b/backend/internal/service/gateway_systemone_test.go index c7a6606b54..62b3051ee4 100644 --- a/backend/internal/service/gateway_systemone_test.go +++ b/backend/internal/service/gateway_systemone_test.go @@ -231,3 +231,140 @@ func TestForwardSystemOneTransportAndTimeoutFailOver(t *testing.T) { require.ErrorAs(t, err, &failoverErr) } } + +func newSystemOneStatusUpstream(status int, body string) *systemOneHTTPUpstream { + return &systemOneHTTPUpstream{do: func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: status, Header: http.Header{"X-Request-Id": []string{"req-err"}}, Body: io.NopCloser(strings.NewReader(body))}, nil + }} +} + +func systemOneOpsEvents(t *testing.T, c *gin.Context) []*OpsUpstreamErrorEvent { + t.Helper() + raw, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + events, ok := raw.([]*OpsUpstreamErrorEvent) + require.True(t, ok) + return events +} + +func TestForwardSystemOneRequestErrorsNeverTouchAccountState(t *testing.T) { + for _, status := range []int{http.StatusBadRequest, http.StatusUnprocessableEntity} { + t.Run(http.StatusText(status), func(t *testing.T) { + // Even a custom error-code rule must not let client input disable the account. + account := &Account{ID: 21, Platform: PlatformTypeSafe, Type: AccountTypeAPIKey, Credentials: map[string]any{ + "base_url": "http://typesafe.test", "api_key": "ts-secret", + "custom_error_codes_enabled": true, "custom_error_codes": []any{float64(status)}, + }} + repo := &systemOnePolicyAccountRepo{account: account} + svc := newSystemOneTestService(newSystemOneStatusUpstream(status, `{"detail":"bad question"}`)) + svc.rateLimitService = NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + c := newSystemOneTestContext() + + _, err := svc.ForwardSystemOne(context.Background(), c, account, []byte(`{}`)) + var upstreamErr *SystemOneUpstreamError + require.ErrorAs(t, err, &upstreamErr) + require.Equal(t, status, upstreamErr.StatusCode) + require.Zero(t, repo.errorCalls+repo.rateLimitedCalls+repo.overloadedCalls) + events := systemOneOpsEvents(t, c) + require.Len(t, events, 1) + require.Equal(t, "http_error", events[0].Kind) + require.Equal(t, status, events[0].UpstreamStatusCode) + require.Equal(t, "req-err", events[0].UpstreamRequestID) + }) + } +} + +func TestForwardSystemOneAccountLevelFailuresFailOver(t *testing.T) { + for _, tc := range []struct { + name string + status int + wantError int + }{ + {name: "payment required", status: http.StatusPaymentRequired, wantError: 1}, + {name: "forbidden", status: http.StatusForbidden, wantError: 1}, + } { + t.Run(tc.name, func(t *testing.T) { + account := &Account{ID: 22, Platform: PlatformTypeSafe, Type: AccountTypeAPIKey, Credentials: map[string]any{"base_url": "http://typesafe.test", "api_key": "ts-secret"}} + repo := &systemOnePolicyAccountRepo{account: account} + svc := newSystemOneTestService(newSystemOneStatusUpstream(tc.status, `{"detail":"account problem"}`)) + svc.rateLimitService = NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + c := newSystemOneTestContext() + + _, err := svc.ForwardSystemOne(context.Background(), c, account, []byte(`{}`)) + var failoverErr *UpstreamFailoverError + require.ErrorAs(t, err, &failoverErr) + require.True(t, failoverErr.ShouldRetryNextAccount()) + require.Equal(t, tc.status, failoverErr.StatusCode) + require.Equal(t, tc.wantError, repo.errorCalls) + require.Equal(t, "failover", systemOneOpsEvents(t, c)[0].Kind) + }) + } +} + +func TestForwardSystemOneHonorsCustomErrorCodes(t *testing.T) { + account := &Account{ID: 23, Platform: PlatformTypeSafe, Type: AccountTypeAPIKey, Credentials: map[string]any{ + "base_url": "http://typesafe.test", "api_key": "ts-secret", + "custom_error_codes_enabled": true, "custom_error_codes": []any{float64(http.StatusNotFound)}, + }} + repo := &systemOnePolicyAccountRepo{account: account} + svc := newSystemOneTestService(newSystemOneStatusUpstream(http.StatusNotFound, `{"detail":"missing"}`)) + svc.rateLimitService = NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + + _, err := svc.ForwardSystemOne(context.Background(), newSystemOneTestContext(), account, []byte(`{}`)) + var failoverErr *UpstreamFailoverError + require.ErrorAs(t, err, &failoverErr) + require.Equal(t, 1, repo.errorCalls) +} + +func TestForwardSystemOneUnhandledStatusDoesNotFailOver(t *testing.T) { + account := &Account{ID: 24, Platform: PlatformTypeSafe, Type: AccountTypeAPIKey, Credentials: map[string]any{"base_url": "http://typesafe.test", "api_key": "ts-secret"}} + repo := &systemOnePolicyAccountRepo{account: account} + svc := newSystemOneTestService(newSystemOneStatusUpstream(http.StatusConflict, `{"detail":"conflict"}`)) + svc.rateLimitService = NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + + _, err := svc.ForwardSystemOne(context.Background(), newSystemOneTestContext(), account, []byte(`{}`)) + var upstreamErr *SystemOneUpstreamError + require.ErrorAs(t, err, &upstreamErr) + require.Equal(t, http.StatusConflict, upstreamErr.StatusCode) + require.Zero(t, repo.errorCalls) +} + +func TestForwardSystemOneNormalizesResponseContentType(t *testing.T) { + for _, tc := range []struct{ upstream, want string }{ + {"application/json; charset=utf-8", "application/json; charset=utf-8"}, + {"application/problem+json", "application/problem+json"}, + {"text/html; charset=utf-8", "application/json"}, + {"", "application/json"}, + {"not a media type;;", "application/json"}, + } { + upstream := &systemOneHTTPUpstream{do: func(*http.Request) (*http.Response, error) { + header := make(http.Header) + if tc.upstream != "" { + header.Set("Content-Type", tc.upstream) + } + return &http.Response{StatusCode: http.StatusOK, Header: header, Body: io.NopCloser(strings.NewReader(`{"answers":{}}`))}, nil + }} + account := &Account{ID: 25, Platform: PlatformTypeSafe, Type: AccountTypeAPIKey, Credentials: map[string]any{"base_url": "http://typesafe.test", "api_key": "ts-secret"}} + result, err := newSystemOneTestService(upstream).ForwardSystemOne(context.Background(), newSystemOneTestContext(), account, []byte(`{}`)) + require.NoError(t, err) + require.Equal(t, tc.want, result.ContentType, tc.upstream) + } +} + +func TestForwardSystemOneRejectsOversizedResponse(t *testing.T) { + oversized := `{"pad":"` + strings.Repeat("a", typesafe.MaxSystemOneResponseBytes) + `"}` + upstream := &systemOneHTTPUpstream{do: func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: http.StatusOK, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(oversized))}, nil + }} + account := &Account{ID: 26, Platform: PlatformTypeSafe, Type: AccountTypeAPIKey, Credentials: map[string]any{"base_url": "http://typesafe.test", "api_key": "ts-secret"}} + _, err := newSystemOneTestService(upstream).ForwardSystemOne(context.Background(), newSystemOneTestContext(), account, []byte(`{}`)) + require.ErrorIs(t, err, typesafe.ErrSystemOneResponseTooLarge) +} + +func TestTypeSafeAccountBaseURLNeverFallsBackToAnthropic(t *testing.T) { + account := &Account{Platform: PlatformTypeSafe, Type: AccountTypeAPIKey, Credentials: map[string]any{"api_key": "ts-secret"}} + require.Equal(t, typesafe.DefaultBaseURL, account.GetBaseURL()) + require.Equal(t, typesafe.DefaultBaseURL, account.GetTypeSafeBaseURL()) + anthropic := &Account{Platform: PlatformAnthropic, Type: AccountTypeAPIKey, Credentials: map[string]any{"api_key": "sk"}} + require.Equal(t, "https://api.anthropic.com", anthropic.GetBaseURL()) +} diff --git a/backend/migrations/241_add_typesafe_platform.sql b/backend/migrations/241_add_typesafe_platform.sql new file mode 100644 index 0000000000..82a81de70d --- /dev/null +++ b/backend/migrations/241_add_typesafe_platform.sql @@ -0,0 +1,26 @@ +-- Add TypeSafe (Jev System One) as a first-class platform. +-- +-- 1. user_platform_quotas.platform CHECK +-- 2. composite_model_routes.target_platform CHECK +-- +-- TypeSafe 不是对话模型，不进入渠道监控 provider，因此 channel_monitors / +-- channel_monitor_request_templates 的约束保持不变。 +-- +-- Runs after 238_opencode_go_platform.sql. DROP ... IF EXISTS 保证可重入； +-- 新约束是 238 的超集，存量行瞬时校验通过。 + +ALTER TABLE user_platform_quotas + DROP CONSTRAINT IF EXISTS user_platform_quotas_platform_check; + +ALTER TABLE user_platform_quotas + ADD CONSTRAINT user_platform_quotas_platform_check + CHECK (platform IN ('anthropic', 'openai', 'gemini', 'antigravity', 'grok', + 'kimi', 'zhipu', 'deepseek', 'minimax', 'opencode_go', 'typesafe')); + +ALTER TABLE composite_model_routes + DROP CONSTRAINT IF EXISTS composite_model_routes_target_platform_check; + +ALTER TABLE composite_model_routes + ADD CONSTRAINT composite_model_routes_target_platform_check + CHECK (target_platform IN ('anthropic', 'openai', 'gemini', 'antigravity', 'grok', + 'kimi', 'zhipu', 'deepseek', 'minimax', 'opencode_go', 'typesafe')); diff --git a/backend/migrations/typesafe_platform_migration_test.go b/backend/migrations/typesafe_platform_migration_test.go new file mode 100644 index 0000000000..7eb2380918 --- /dev/null +++ b/backend/migrations/typesafe_platform_migration_test.go @@ -0,0 +1,21 @@ +package migrations + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestTypeSafePlatformMigration(t *testing.T) { + content, err := FS.ReadFile("241_add_typesafe_platform.sql") + require.NoError(t, err) + + sql := strings.Join(strings.Fields(string(content)), " ") + require.Contains(t, sql, "DROP CONSTRAINT IF EXISTS user_platform_quotas_platform_check") + require.Contains(t, sql, "DROP CONSTRAINT IF EXISTS composite_model_routes_target_platform_check") + require.Contains(t, sql, + "CHECK (platform IN ('anthropic', 'openai', 'gemini', 'antigravity', 'grok', 'kimi', 'zhipu', 'deepseek', 'minimax', 'opencode_go', 'typesafe'))") + require.Contains(t, sql, + "CHECK (target_platform IN ('anthropic', 'openai', 'gemini', 'antigravity', 'grok', 'kimi', 'zhipu', 'deepseek', 'minimax', 'opencode_go', 'typesafe'))") +}