From 6a54bbff8492baacc901c8e7d67fcb2304db6656 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 1 Jul 2026 16:36:17 +0000 Subject: [PATCH 1/3] =?UTF-8?q?feat(routing):=20EndpointModeAuto=20?= =?UTF-8?q?=E2=80=94=20runtime=20protocol=20auto-detection?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an "auto" OpenAI endpoint mode: for OpenAI-compatible providers the gateway tries the scenario-preferred protocol first, falls back to the alternate on a retryable failure, and caches the winning protocol per provider+model. Gated behind the global "auto_endpoint" experimental flag. Rebased onto main and aligned with main's structure: endpoint code lives in package server (protocol_endpoint.go + endpoint_auto.go/endpoint_cache.go), not a separate internal/endpoint package — following main's housekeeping consolidation. The auto path nests as protocol-fallback (outer) over provider-failover (inner) sharing a single first-chunk gate, via the new dispatchWithPriorityFailoverGated variant. - ai: EndpointModeAuto + IsAutoEndpointMode; non-Codex OAuth defaults to auto - server: resolveAutoTarget (override→cache→scenario), dispatchWithAutoFallback, EndpointCache; wired into chat + responses handlers via runOpenAI*Attempt - protocol_endpoint: ResolveOpenAIEndpoint handles Auto (mirror incoming) as the non-auto fallback - probe: E2E probe honors the endpoint cache - client: IsNonRetryableForProtocolSwitch guards pointless protocol swaps - config: auto_endpoint scenario flag - frontend: experimental toggle + i18n (en/zh) https://claude.ai/code/session_01N9JnUTVt41NcGxYyj3xewf --- .design/openai-endpoint-routing.md | 69 +++- ai/provider.go | 21 +- .../components/GlobalExperimentalFeatures.tsx | 54 ++- .../src/components/ProviderFormDialog.tsx | 1 + frontend/src/contexts/FeatureFlagsContext.tsx | 8 +- frontend/src/i18n/locales/en.ts | 5 +- frontend/src/i18n/locales/zh.ts | 5 +- internal/client/error.go | 28 +- internal/client/error_test.go | 44 +++ internal/probe/e2e.go | 81 +++- internal/server/config/config.go | 5 + internal/server/config/extension_keys.go | 1 + internal/server/endpoint_auto.go | 154 ++++++++ internal/server/endpoint_auto_test.go | 353 ++++++++++++++++++ internal/server/endpoint_cache.go | 62 +++ internal/server/endpoint_cache_test.go | 52 +++ internal/server/failover_dispatch.go | 48 ++- internal/server/openai_chat.go | 76 +++- internal/server/openai_responses.go | 65 +++- internal/server/protocol_endpoint.go | 6 +- internal/server/server.go | 5 + internal/server/server_flags.go | 8 + 22 files changed, 1085 insertions(+), 66 deletions(-) create mode 100644 internal/client/error_test.go create mode 100644 internal/server/endpoint_auto.go create mode 100644 internal/server/endpoint_auto_test.go create mode 100644 internal/server/endpoint_cache.go create mode 100644 internal/server/endpoint_cache_test.go diff --git a/.design/openai-endpoint-routing.md b/.design/openai-endpoint-routing.md index 78b81d6fc..831194e4a 100644 --- a/.design/openai-endpoint-routing.md +++ b/.design/openai-endpoint-routing.md @@ -240,7 +240,74 @@ PR #976 引入此设计。涉及行为变更的两个点： --- -## 10. 不在本文档范围 +## 10. Auto 模式（运行时协议自适应） + +### 10.1 问题 + +第三方聚合 provider 在同一个 OpenAI-style API base 下托管多种模型，有的只支持 Chat、有的只支持 Responses。`OpenAIEndpointMode` 是 provider 级别的，无法 per-model 区分，且模型列表不可穷尽。 + +### 10.2 与 AdaptiveProbe（§2.1）的本质区别 + +| | AdaptiveProbe（已废弃）| EndpointModeAuto | +|---|---|---| +| 探测方式 | 独立 probe 请求（烧 token、被限流） | 用户真实请求 | +| 失败处理 | 缓存失败 → 误判固化 | **只缓存成功，失败不缓存** | +| 冷启动 | 阻塞 10s | 零阻塞，首请求直接发 | +| 级联风险 | probe 失败 → 整个 provider 不可用 | 仅影响当前请求，自动 fallback | +| 实现位置 | 运行时缓存层（§2.1 的根因） | handler 层，resolver 保持纯函数 | + +### 10.3 行为 + +``` +EndpointModeAuto = "auto" +``` + +1. **Rule override 最高优先级**：`openai_endpoint_override` 设了 chat/responses 就直接用，跳过 auto。 +2. **查成功缓存**：`provider_uuid:model_name → protocol`，命中则直接用缓存协议。 +3. **缓存未命中**：用 incoming protocol 作为首次尝试。 +4. **首次尝试失败**：排除不可重试的错误（401/403/429/内容相关），其余都做协议 fallback。 +5. **fallback 成功**：缓存 `provider_uuid:model_name → alternate_protocol`（24h TTL）。 +6. **fallback 失败**：返回错误，不缓存。 + +### 10.4 错误分类（排除法） + +不重试的错误（换协议也没用）： +- 401 / 403：认证问题 +- 429：限流 +- `context_length_exceeded`、`content_policy`、`invalid_api_key`、`model_not_found`：内容/模型问题 + +其余所有错误（404、500、未知错误）→ 允许 fallback。 + +### 10.5 实现层次 + +``` +┌─────────────────────────────────────────────────────────────┐ +│ Layer 2 Rule flag — openai_endpoint_override │ +│ 仍然最高优先级，设了就跳过 auto │ +├─────────────────────────────────────────────────────────────┤ +│ Layer 1 Provider mode — EndpointModeAuto │ +│ handler 层管理缓存 + fallback │ +│ resolver 保持纯函数（auto → mirror incoming） │ +├─────────────────────────────────────────────────────────────┤ +│ Cache EndpointCache（in-memory, per provider+model） │ +│ 只写入成功结果，24h TTL，惰性淘汰 │ +└─────────────────────────────────────────────────────────────┘ +``` + +Auto 逻辑在 handler 层（`openai_chat.go`、`openai_responses.go`），使用 `firstChunkGate` 缓冲首次尝试的响应。Gate 与 `dispatchWithPriorityFailover` 共享（通过 `dispatchWithPriorityFailoverGated` 接受外部 gate），避免嵌套。 + +### 10.6 关键文件 + +- `ai/provider.go` — `EndpointModeAuto` 常量 +- `internal/server/endpoint_cache.go` — 成功缓存 +- `internal/server/endpoint_auto.go` — 错误分类 + `dispatchWithAutoFallback` +- `internal/server/failover_dispatch.go` — `dispatchWithPriorityFailoverGated` +- `internal/server/openai_chat.go` — Chat handler 的 auto 分支 +- `internal/server/openai_responses.go` — Responses handler 的 auto 分支 + +--- + +## 11. 不在本文档范围 - Anthropic / Google provider 的路由（走各自原生 endpoint，不进 OpenAI resolver） - Smart routing / load balance 选哪个 service（在 endpoint 选择之前） diff --git a/ai/provider.go b/ai/provider.go index 455757b5a..5d2bc56ec 100644 --- a/ai/provider.go +++ b/ai/provider.go @@ -221,7 +221,7 @@ type OpenAIEndpointMode string const ( EndpointModeUnknown OpenAIEndpointMode = "" - // EndpointModeChat (the default) means the provider only exposes + // EndpointModeChat means the provider only exposes // /chat/completions. An incoming Responses request will be downgraded // to Chat if its features allow; otherwise the request is rejected. EndpointModeChat OpenAIEndpointMode = "chat" @@ -236,18 +236,31 @@ const ( // client's incoming API so native semantics (reasoning blocks, // previous_response_id continuity, etc.) survive the round trip. EndpointModeBoth OpenAIEndpointMode = "both" + + // EndpointModeAuto is for providers where the gateway tries the + // incoming protocol first; on failure it falls back to the alternate + // protocol and caches successful results per model. This is also the + // default behavior when no mode is explicitly set (zero value). + EndpointModeAuto OpenAIEndpointMode = "auto" ) +// IsAutoEndpointMode reports whether the mode represents auto-detection +// behavior. Both the explicit "auto" value and the zero value (no mode +// set) are treated as auto. +func IsAutoEndpointMode(mode OpenAIEndpointMode) bool { + return mode == EndpointModeAuto || mode == EndpointModeUnknown +} + // OpenAIEndpointModeForIssuer returns the OpenAIEndpointMode that an OAuth // provider should carry given its issuer. Currently only Codex needs a -// non-default mode; other issuers fall through to EndpointModeChat (zero -// value). Centralized so the OAuth web handler and the CLI flow agree on +// non-default mode; other issuers use the zero value (auto). +// Centralized so the OAuth web handler and the CLI flow agree on // the same mapping and future issuer-specific defaults land in one place. func OpenAIEndpointModeForIssuer(issuer Issuer) OpenAIEndpointMode { if issuer == IssuerCodex { return EndpointModeResponses } - return EndpointModeChat + return EndpointModeUnknown } // IsVirtual reports whether this provider routes to the in-process vmodel diff --git a/frontend/src/components/GlobalExperimentalFeatures.tsx b/frontend/src/components/GlobalExperimentalFeatures.tsx index 3bce7430b..6e8491217 100644 --- a/frontend/src/components/GlobalExperimentalFeatures.tsx +++ b/frontend/src/components/GlobalExperimentalFeatures.tsx @@ -1,5 +1,5 @@ import {useFeatureFlags} from '@/contexts/FeatureFlagsContext'; -import { Psychology as IconBrain, Shield as IconShield, SettingsApplications } from '@/components/icons'; +import { Psychology as IconBrain, Shield as IconShield, SettingsApplications, Autorenew } from '@/components/icons'; import {Alert, Box, Chip, Tooltip, Typography,} from '@mui/material'; import React, {useEffect, useState} from 'react'; import {useTranslation} from 'react-i18next'; @@ -19,6 +19,7 @@ const GlobalExperimentalFeatures: React.FC = () => { const [features, setFeatures] = useState>({}); const [guardrailsEnabled, setGuardrailsEnabled] = useState(false); const [mcpEnabled, setMCPEnabled] = useState(false); + const [autoEndpointEnabled, setAutoEndpointEnabled] = useState(false); const [loading, setLoading] = useState(true); const {refresh} = useFeatureFlags(); @@ -43,6 +44,10 @@ const GlobalExperimentalFeatures: React.FC = () => { const mcpResult = await api.getScenarioFlag('_global', 'mcp'); setMCPEnabled(mcpResult?.data?.value || false); + // Load Auto Endpoint flag + const autoEndpointResult = await api.getScenarioFlag('_global', 'auto_endpoint'); + setAutoEndpointEnabled(autoEndpointResult?.data?.value || false); + } catch (error) { console.error('Failed to load global experimental features:', error); } finally { @@ -104,6 +109,24 @@ const GlobalExperimentalFeatures: React.FC = () => { }); }; + const toggleAutoEndpoint = () => { + const newValue = !autoEndpointEnabled; + api.setScenarioFlag('_global', 'auto_endpoint', newValue) + .then((result) => { + if (result.success) { + setAutoEndpointEnabled(newValue); + refresh(); + } else { + console.error('Failed to set Auto Endpoint:', result); + loadFeatures(); + } + }) + .catch((err) => { + console.error('Failed to set Auto Endpoint:', err); + loadFeatures(); + }); + }; + useEffect(() => { loadFeatures(); }, []); @@ -218,6 +241,35 @@ const GlobalExperimentalFeatures: React.FC = () => { )} + {/* Auto Endpoint Section */} + + + + + {t('system.experimentalFeatures.autoEndpoint')} + + + + + + + + + + + {autoEndpointEnabled && ( + + + {t('system.experimentalFeatures.autoEndpointEnabledInfo')} + + + )} + ); }; diff --git a/frontend/src/components/ProviderFormDialog.tsx b/frontend/src/components/ProviderFormDialog.tsx index 0fb779fff..cb27f0410 100644 --- a/frontend/src/components/ProviderFormDialog.tsx +++ b/frontend/src/components/ProviderFormDialog.tsx @@ -48,6 +48,7 @@ export interface EnhancedProviderFormData { apiBaseOpenAI?: string; apiBaseAnthropic?: string; createDualProvider?: boolean; + openaiEndpointMode?: string; /** If set, prefer this exact provider ID when resolving the template. * Avoids mismatches when multiple providers share the same base URL. */ selectedProviderId?: string; diff --git a/frontend/src/contexts/FeatureFlagsContext.tsx b/frontend/src/contexts/FeatureFlagsContext.tsx index 48ace7225..be1fa46c7 100644 --- a/frontend/src/contexts/FeatureFlagsContext.tsx +++ b/frontend/src/contexts/FeatureFlagsContext.tsx @@ -8,6 +8,7 @@ interface FeatureFlagsContextType { skillIde: boolean; enableGuardrails: boolean; enableMCP: boolean; + enableAutoEndpoint: boolean; loading: boolean; refresh: () => void; } @@ -32,20 +33,23 @@ export const FeatureFlagsProvider: React.FC = ({ chil const [skillIde, setSkillIde] = useState(false); const [enableGuardrails, setEnableGuardrails] = useState(false); const [enableMCP, setEnableMCP] = useState(false); + const [enableAutoEndpoint, setEnableAutoEndpoint] = useState(false); const [loading, setLoading] = useState(true); const loadFlags = async () => { try { - const [skillUserResult, skillIdeResult, guardrailsResult, mcpResult] = await Promise.all([ + const [skillUserResult, skillIdeResult, guardrailsResult, mcpResult, autoEndpointResult] = await Promise.all([ api.getScenarioFlag('_global', 'skill_user'), api.getScenarioFlag('_global', 'skill_ide'), api.getScenarioFlag('_global', 'guardrails'), api.getScenarioFlag('_global', 'mcp'), + api.getScenarioFlag('_global', 'auto_endpoint'), ]); setSkillUser(skillUserResult?.data?.value || false); setSkillIde(skillIdeResult?.data?.value || false); setEnableGuardrails(guardrailsResult?.data?.value || false); setEnableMCP(mcpResult?.data?.value || false); + setEnableAutoEndpoint(autoEndpointResult?.data?.value || false); } catch (error) { // Silently fail - flags will default to false // Don't log to console to avoid noise during initial auth @@ -67,7 +71,7 @@ export const FeatureFlagsProvider: React.FC = ({ chil }; return ( - + {children} ); diff --git a/frontend/src/i18n/locales/en.ts b/frontend/src/i18n/locales/en.ts index ed29b845a..affa0833d 100644 --- a/frontend/src/i18n/locales/en.ts +++ b/frontend/src/i18n/locales/en.ts @@ -735,7 +735,10 @@ export default { "enabled": "enabled", "disabled": "disabled - Click to enable", "guardrailsEnabledInfo": "Guardrails is enabled. A \"Guardrails\" page is available in the sidebar for rule management.", - "mcpEnabledInfo": "MCP Tools is enabled. An \"MCP Tools\" page is available under System in the sidebar for configuration." + "mcpEnabledInfo": "MCP Tools is enabled. An \"MCP Tools\" page is available under System in the sidebar for configuration.", + "autoEndpoint": "Auto Endpoint", + "enableAutoEndpoint": "Enable Auto Endpoint Detection - automatically detect whether each model supports Chat Completions or Responses API", + "autoEndpointEnabledInfo": "Auto Endpoint Detection is enabled. For OpenAI-compatible providers, the system will automatically try the preferred protocol and fall back to the alternate on failure, caching successful results per model." }, "about": { "title": "About", diff --git a/frontend/src/i18n/locales/zh.ts b/frontend/src/i18n/locales/zh.ts index 8dd178e0c..bb6af12dc 100644 --- a/frontend/src/i18n/locales/zh.ts +++ b/frontend/src/i18n/locales/zh.ts @@ -736,7 +736,10 @@ export default { "enabled": "已启用", "disabled": "已禁用 - 点击启用", "guardrailsEnabledInfo": "Guardrails 已启用。侧边栏中提供了「Guardrails」页面用于规则管理。", - "mcpEnabledInfo": "MCP Tools 已启用。侧边栏 System 下方提供了「MCP Tools」页面进行配置。" + "mcpEnabledInfo": "MCP Tools 已启用。侧边栏 System 下方提供了「MCP Tools」页面进行配置。", + "autoEndpoint": "自动端点检测", + "enableAutoEndpoint": "启用自动端点检测 - 自动识别每个模型支持 Chat Completions 还是 Responses API", + "autoEndpointEnabledInfo": "自动端点检测已启用。对于 OpenAI 兼容的 Provider，系统将自动尝试首选协议，失败时回退到备选协议，并按模型缓存成功结果。" }, "about": { "title": "关于", diff --git a/internal/client/error.go b/internal/client/error.go index 279319aa3..3feba010a 100644 --- a/internal/client/error.go +++ b/internal/client/error.go @@ -1,6 +1,9 @@ package client -import "fmt" +import ( + "fmt" + "strings" +) // ErrModelsEndpointNotSupported is returned when the provider does not support the models endpoint type ErrModelsEndpointNotSupported struct { @@ -31,3 +34,26 @@ type ErrKimiNotSupported struct { func (e *ErrKimiNotSupported) Error() string { return fmt.Sprintf("Kimi Code does not support %s: %s", e.Operation, e.Reason) } + +// IsNonRetryableForProtocolSwitch reports whether err represents a condition +// where switching the OpenAI endpoint protocol (Chat ↔ Responses) would not +// help. Returns true for nil errors (nothing to retry), auth failures, +// rate-limiting, and content/model errors. +func IsNonRetryableForProtocolSwitch(err error) bool { + if err == nil { + return true + } + s := strings.ToLower(err.Error()) + + for _, kw := range []string{ + "401", "403", "unauthorized", "forbidden", + "429", "rate limit", "ratelimit", "1302", + "context_length", "content_policy", "content_filter", + "invalid_api_key", "model_not_found", "model not found", + } { + if strings.Contains(s, kw) { + return true + } + } + return false +} diff --git a/internal/client/error_test.go b/internal/client/error_test.go new file mode 100644 index 000000000..5f84ddfe0 --- /dev/null +++ b/internal/client/error_test.go @@ -0,0 +1,44 @@ +package client + +import ( + "errors" + "testing" +) + +func TestIsNonRetryableForProtocolSwitch(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {"nil error", nil, true}, + {"auth 401", errors.New("status 401 Unauthorized"), true}, + {"auth 403", errors.New("403 Forbidden"), true}, + {"rate limit 429", errors.New("status 429 Too Many Requests"), true}, + {"rate limit text", errors.New("rate limit exceeded"), true}, + {"rate limit 1302", errors.New("error code 1302"), true}, + {"context length", errors.New("context_length_exceeded"), true}, + {"content policy", errors.New("content_policy_violation"), true}, + {"invalid api key", errors.New("invalid_api_key"), true}, + {"model not found", errors.New("model_not_found"), true}, + {"model not found spaces", errors.New("model not found"), true}, + {"content filter", errors.New("content_filter triggered"), true}, + + // Retryable cases + {"404 not found", errors.New("status 404 Not Found"), false}, + {"500 internal", errors.New("status 500 Internal Server Error"), false}, + {"502 bad gateway", errors.New("502 Bad Gateway"), false}, + {"connection refused", errors.New("connection refused"), false}, + {"unknown error", errors.New("something went wrong"), false}, + {"empty error", errors.New(""), false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := IsNonRetryableForProtocolSwitch(tt.err) + if got != tt.want { + t.Errorf("IsNonRetryableForProtocolSwitch(%q) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} diff --git a/internal/probe/e2e.go b/internal/probe/e2e.go index 74b476087..9378b39fe 100644 --- a/internal/probe/e2e.go +++ b/internal/probe/e2e.go @@ -8,18 +8,28 @@ import ( "github.com/sirupsen/logrus" + "github.com/tingly-dev/tingly-box/ai" "github.com/tingly-dev/tingly-box/internal/client" "github.com/tingly-dev/tingly-box/internal/protocol" "github.com/tingly-dev/tingly-box/internal/server/config" "github.com/tingly-dev/tingly-box/internal/typ" ) +// EndpointCacheGetFn looks up a cached protocol for a provider+model pair. +// Returns the cached APIType and true on hit, or ("", false) on miss. +type EndpointCacheGetFn func(providerUUID, model string) (protocol.APIType, bool) + +// EndpointCacheSetFn writes a successful protocol result into the cache. +type EndpointCacheSetFn func(providerUUID, model string, target protocol.APIType) + // E2EService runs SDK-level end-to-end probes against a rule, a saved // provider, or an inline provider config. It is independent of *Server and // is wired in NewServer. type E2EService struct { - config *config.Config - clientPool *client.ClientPool + config *config.Config + clientPool *client.ClientPool + endpointGet EndpointCacheGetFn + endpointSet EndpointCacheSetFn } // NewE2EService constructs a E2EService. @@ -30,6 +40,21 @@ func NewE2EService(cfg *config.Config, pool *client.ClientPool) *E2EService { } } +// SetEndpointCache wires the endpoint auto-detection cache into the probe +// service. When set, ProbeProviderWithSDK uses the cache for providers with +// EndpointModeAuto to avoid unnecessary fallback retries. +func (e *E2EService) SetEndpointCache(get EndpointCacheGetFn, set EndpointCacheSetFn) { + e.endpointGet = get + e.endpointSet = set +} + +func (e *E2EService) autoEndpointEnabled() bool { + if e.config == nil { + return false + } + return e.config.GetScenarioFlag(typ.ScenarioGlobal, config.ExtensionAutoEndpoint) +} + // Probe performs a non-streaming probe against the target described by req. func (e *E2EService) Probe(ctx context.Context, req *E2ERequest) (*E2EData, error) { provider, model, probeHeaders, err := e.resolveTargetToProviderModel(ctx, req) @@ -166,7 +191,7 @@ func (e *E2EService) resolveProviderTarget(ctx context.Context, req *E2ERequest) apiStyle := provider.APIStyle probeHeaders := map[string]string{ "X-Tingly-Probe-Service": req.ProviderUUID + ":" + model, - "X-Tingly-Debug-Routing": "1", + "X-Tingly-Debug-Routing": "1", } logrus.Debugf("[probe-e2e] provider %s -> TB loopback %s (service pin=%s:%s)", provider.UUID, apiBase, req.ProviderUUID, model) @@ -300,9 +325,10 @@ func (e *E2EService) ProbeProviderWithSDK(ctx context.Context, provider *typ.Pro client.ApplyProbeHeadersToClient(oc) routing = client.ApplyRoutingCaptureToClient(oc) } - // Codex OAuth providers only speak the Responses API. if isCodexOAuth(provider) { result, err = probeOpenAIResponses(ctx, oc, model, message, mode) + } else if e.autoEndpointEnabled() && ai.IsAutoEndpointMode(provider.OpenAIEndpointMode) { + result, err = e.probeOpenAIAutoFallback(ctx, oc, provider, model, message, mode) } else { result, err = probeOpenAIChat(ctx, oc, model, message, mode) } @@ -367,6 +393,53 @@ func applyRoutingCapture(result *E2EData, cap *client.RoutingCapture) { } } +// probeOpenAIAutoFallback tries the preferred protocol first; on retryable +// failure it falls back to the alternate protocol. Results are cached via +// endpointGet/endpointSet when available. +func (e *E2EService) probeOpenAIAutoFallback( + ctx context.Context, oc client.OpenAIClientInterface, + provider *typ.Provider, model, message string, mode E2EMode, +) (*E2EData, error) { + firstProto, altProto := protocol.TypeOpenAIChat, protocol.TypeOpenAIResponses + if e.endpointGet != nil { + if cached, ok := e.endpointGet(provider.UUID, model); ok && cached == protocol.TypeOpenAIResponses { + firstProto, altProto = protocol.TypeOpenAIResponses, protocol.TypeOpenAIChat + } + } + + probeFn := func(proto protocol.APIType) (*E2EData, error) { + if proto == protocol.TypeOpenAIResponses { + return probeOpenAIResponses(ctx, oc, model, message, mode) + } + return probeOpenAIChat(ctx, oc, model, message, mode) + } + + result, err := probeFn(firstProto) + if err == nil && result != nil && result.Content != "" { + e.cacheEndpoint(provider.UUID, model, firstProto) + return result, nil + } + + if err != nil && client.IsNonRetryableForProtocolSwitch(err) { + return result, err + } + + logrus.Infof("[probe-auto] %s:%s %s probe failed, trying %s: %v", provider.Name, model, firstProto, altProto, err) + result, err = probeFn(altProto) + if err == nil && result != nil && result.Content != "" { + e.cacheEndpoint(provider.UUID, model, altProto) + return result, nil + } + + return result, err +} + +func (e *E2EService) cacheEndpoint(providerUUID, model string, proto protocol.APIType) { + if e.endpointSet != nil { + e.endpointSet(providerUUID, model, proto) + } +} + func (e *E2EService) probeProviderStream(ctx context.Context, provider *typ.Provider, model, message string, testMode E2EMode) (*E2EData, error) { return e.ProbeProviderWithSDK(ctx, provider, model, message, testMode) } diff --git a/internal/server/config/config.go b/internal/server/config/config.go index fe9462c2b..dd290f2b1 100644 --- a/internal/server/config/config.go +++ b/internal/server/config/config.go @@ -1815,6 +1815,11 @@ func (c *Config) SetScenarioFlag(scenario typ.RuleScenario, flagName string, val config.Extensions = make(map[string]interface{}) } config.Extensions[ExtensionMCP] = value + case ExtensionAutoEndpoint: + if config.Extensions == nil { + config.Extensions = make(map[string]interface{}) + } + config.Extensions[ExtensionAutoEndpoint] = value default: return fmt.Errorf("unknown flag name: %s", flagName) } diff --git a/internal/server/config/extension_keys.go b/internal/server/config/extension_keys.go index 4f03799ce..ffe277011 100644 --- a/internal/server/config/extension_keys.go +++ b/internal/server/config/extension_keys.go @@ -9,4 +9,5 @@ const ( ExtensionMCP = "mcp" ExtensionSkillUser = "skill_user" ExtensionSkillIDE = "skill_ide" + ExtensionAutoEndpoint = "auto_endpoint" ) diff --git a/internal/server/endpoint_auto.go b/internal/server/endpoint_auto.go new file mode 100644 index 000000000..51ba00966 --- /dev/null +++ b/internal/server/endpoint_auto.go @@ -0,0 +1,154 @@ +package server + +import ( + "net/http" + + "github.com/gin-gonic/gin" + "github.com/sirupsen/logrus" + "github.com/tingly-dev/tingly-box/internal/client" + "github.com/tingly-dev/tingly-box/internal/protocol" + "github.com/tingly-dev/tingly-box/internal/typ" +) + +// alternateOpenAIProtocol returns the other OpenAI protocol type. +func alternateOpenAIProtocol(current protocol.APIType) protocol.APIType { + if current == protocol.TypeOpenAIResponses { + return protocol.TypeOpenAIChat + } + return protocol.TypeOpenAIResponses +} + +// incomingToTarget maps IncomingAPIType to protocol.APIType. +func incomingToTarget(incoming IncomingAPIType) protocol.APIType { + if incoming == IncomingAPIResponses { + return protocol.TypeOpenAIResponses + } + return protocol.TypeOpenAIChat +} + +// extractLastGinError returns the most recent error recorded on the gin +// context via c.Error(). Returns nil when no errors exist. +func extractLastGinError(c *gin.Context) error { + errs := c.Errors + if len(errs) == 0 { + return nil + } + return errs[len(errs)-1].Err +} + +// clearGinErrors removes all errors from the gin context so that a +// fallback retry starts with a clean slate. +func clearGinErrors(c *gin.Context) { + c.Errors = c.Errors[:0] +} + +// overrideToTarget converts an EndpointOverride to a protocol.APIType. +func overrideToTarget(ov EndpointOverride) protocol.APIType { + if ov == OverrideResponses { + return protocol.TypeOpenAIResponses + } + return protocol.TypeOpenAIChat +} + +// scenarioPreferredProtocol returns the protocol an auto-mode provider +// should try first on a cache miss. Scenarios whose client ecosystem is +// natively Responses-based (Codex) start with Responses regardless of the +// incoming transport — providers serving such traffic overwhelmingly speak +// Responses, so leading with it saves a wasted first round trip. All other +// scenarios mirror the incoming API. +func scenarioPreferredProtocol(scenario typ.RuleScenario, incoming IncomingAPIType) protocol.APIType { + switch scenario.Base() { + case typ.ScenarioCodex: + return protocol.TypeOpenAIResponses + default: + return incomingToTarget(incoming) + } +} + +// resolveAutoTarget handles the auto-mode target resolution shared by both +// OpenAI Chat and Responses handlers. It checks override → cache → +// scenario-preferred default. Returns the resolved target and whether +// auto-fallback should be enabled. +func (s *Server) resolveAutoTarget( + flags typ.RuleFlags, provider *typ.Provider, model string, scenario typ.RuleScenario, incoming IncomingAPIType, +) (target protocol.APIType, autoFallback bool) { + if ov := ParseEndpointOverride(flags.OpenAIEndpointOverride); ov == OverrideChat || ov == OverrideResponses { + return overrideToTarget(ov), false + } + if cached, ok := s.endpointCache.Get(provider.UUID, model); ok { + return cached, false + } + return scenarioPreferredProtocol(scenario, incoming), true +} + +// autoDispatchFn is the callback for dispatchWithAutoFallback. +// It performs transform + dispatch for a given target protocol, using +// the provided gate. Returns the provider/model that served the final +// attempt (failover may have moved past the initially selected service) +// and ok=true if dispatch executed (even on error); ok=false means the +// transform itself failed. +type autoDispatchFn func(target protocol.APIType, gate *firstChunkGate) (served *typ.Provider, servedModel string, ok bool) + +// gateSucceeded reports whether the attempt behind the gate produced a +// success: either the stream committed its first chunk, or a buffered +// non-error status is waiting to flush. +func gateSucceeded(gate *firstChunkGate) bool { + return gate.Committed() || (gate.Status() > 0 && gate.Status() < http.StatusBadRequest) +} + +// dispatchWithAutoFallback wraps a dispatch attempt with protocol +// auto-detection. It tries the preferred target first; on retryable +// failure it falls back to the alternate protocol. Successful protocol +// choices are cached per provider+model, attributed to the service that +// actually served the request — under multi-service failover that may +// differ from the initially selected provider, and caching against the +// initial one would pin a protocol it never confirmed. +func (s *Server) dispatchWithAutoFallback( + c *gin.Context, + provider *typ.Provider, + model string, + preferredTarget protocol.APIType, + dispatch autoDispatchFn, +) { + realWriter := c.Writer + gate := newFirstChunkGate(realWriter) + c.Writer = gate + defer func() { + c.Writer = realWriter + gate.CommitIfBuffered() + }() + + // First attempt with preferred protocol + served, servedModel, ok := dispatch(preferredTarget, gate) + + if gateSucceeded(gate) { + if ok && served != nil { + s.endpointCache.Set(served.UUID, servedModel, preferredTarget) + } + return + } + + // Check if fallback is worthwhile + if !isRetryableStatus(gate.Status()) { + return + } + lastErr := extractLastGinError(c) + if client.IsNonRetryableForProtocolSwitch(lastErr) { + return + } + + // Fallback to alternate protocol + altTarget := alternateOpenAIProtocol(preferredTarget) + logrus.WithContext(c.Request.Context()).Infof( + "[auto-endpoint] %s:%s status=%d → fallback from %s to %s", + provider.UUID, model, gate.Status(), preferredTarget, altTarget, + ) + gate.Discard() + clearGinErrors(c) + + served, servedModel, ok = dispatch(altTarget, gate) + + if gateSucceeded(gate) && ok && served != nil { + s.endpointCache.Set(served.UUID, servedModel, altTarget) + } +} diff --git a/internal/server/endpoint_auto_test.go b/internal/server/endpoint_auto_test.go new file mode 100644 index 000000000..0b59b35fd --- /dev/null +++ b/internal/server/endpoint_auto_test.go @@ -0,0 +1,353 @@ +package server + +import ( + "fmt" + "net/http" + "net/http/httptest" + "testing" + + "github.com/gin-gonic/gin" + "github.com/tingly-dev/tingly-box/internal/protocol" + "github.com/tingly-dev/tingly-box/internal/typ" +) + +func TestAlternateOpenAIProtocol(t *testing.T) { + if got := alternateOpenAIProtocol(protocol.TypeOpenAIChat); got != protocol.TypeOpenAIResponses { + t.Errorf("alternate of chat = %v, want responses", got) + } + if got := alternateOpenAIProtocol(protocol.TypeOpenAIResponses); got != protocol.TypeOpenAIChat { + t.Errorf("alternate of responses = %v, want chat", got) + } +} + +func TestIncomingToTarget(t *testing.T) { + if got := incomingToTarget(IncomingAPIChat); got != protocol.TypeOpenAIChat { + t.Errorf("incoming chat → %v, want chat", got) + } + if got := incomingToTarget(IncomingAPIResponses); got != protocol.TypeOpenAIResponses { + t.Errorf("incoming responses → %v, want responses", got) + } +} + +func TestScenarioPreferredProtocol(t *testing.T) { + tests := []struct { + name string + scenario typ.RuleScenario + incoming IncomingAPIType + want protocol.APIType + }{ + {"codex prefers responses even on chat ingress", typ.ScenarioCodex, IncomingAPIChat, protocol.TypeOpenAIResponses}, + {"codex prefers responses on responses ingress", typ.ScenarioCodex, IncomingAPIResponses, protocol.TypeOpenAIResponses}, + {"codex profile suffix normalized", typ.RuleScenario("codex:p1"), IncomingAPIChat, protocol.TypeOpenAIResponses}, + {"openai mirrors chat ingress", typ.ScenarioOpenAI, IncomingAPIChat, protocol.TypeOpenAIChat}, + {"openai mirrors responses ingress", typ.ScenarioOpenAI, IncomingAPIResponses, protocol.TypeOpenAIResponses}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := scenarioPreferredProtocol(tt.scenario, tt.incoming); got != tt.want { + t.Errorf("scenarioPreferredProtocol(%q, %q) = %v, want %v", tt.scenario, tt.incoming, got, tt.want) + } + }) + } +} + +// TestDispatchWithAutoFallback_CacheAttributedToServingProvider covers the +// multi-service failover interaction: when the initial provider fails and a +// fallback provider serves the request, the protocol cache entry must be +// written for the serving provider, not the initial one — otherwise the +// initial provider gets pinned to a protocol it never confirmed. +func TestDispatchWithAutoFallback_CacheAttributedToServingProvider(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + + initial := &typ.Provider{UUID: "prov-initial"} + serving := &typ.Provider{UUID: "prov-serving"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + // Simulate a dispatch where failover moved past the initial provider: + // the gate commits (success) and the serving identity differs from initial. + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + gate.WriteHeader(200) + gate.CommitFirstChunk() + return serving, "served-model", true + } + + s.dispatchWithAutoFallback(c, initial, "req-model", protocol.TypeOpenAIChat, dispatch) + + if _, ok := s.endpointCache.Get(initial.UUID, "req-model"); ok { + t.Error("cache must not contain an entry for the initial provider") + } + got, ok := s.endpointCache.Get(serving.UUID, "served-model") + if !ok { + t.Fatal("cache must contain an entry for the serving provider") + } + if got != protocol.TypeOpenAIChat { + t.Errorf("cached protocol = %v, want chat", got) + } +} + +// TestDispatchWithAutoFallback_NoCacheOnTransformFailure ensures a failed +// transform (ok=false, served=nil) never writes a cache entry even if the +// gate state looks successful. +func TestDispatchWithAutoFallback_NoCacheOnTransformFailure(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + initial := &typ.Provider{UUID: "prov-initial"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + gate.WriteHeader(200) // transform error path writes a JSON error; simulate benign status + return nil, "", false + } + + s.dispatchWithAutoFallback(c, initial, "m", protocol.TypeOpenAIChat, dispatch) + + if _, ok := s.endpointCache.Get(initial.UUID, "m"); ok { + t.Error("cache must stay empty when the transform failed") + } +} + +// TestDispatchWithAutoFallback_FirstAttemptSucceeds verifies the happy path: +// preferred protocol works on the first try and gets cached. +func TestDispatchWithAutoFallback_FirstAttemptSucceeds(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + calls := 0 + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + calls++ + gate.WriteHeader(200) + gate.CommitFirstChunk() + return provider, "model-a", true + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIResponses, dispatch) + + if calls != 1 { + t.Errorf("dispatch called %d times, want 1", calls) + } + got, ok := s.endpointCache.Get(provider.UUID, "model-a") + if !ok || got != protocol.TypeOpenAIResponses { + t.Errorf("cache = (%v, %v), want (responses, true)", got, ok) + } +} + +// TestDispatchWithAutoFallback_FallbackSucceeds verifies: preferred fails +// with retryable status → alternate tried → alternate succeeds → alternate +// protocol cached. +func TestDispatchWithAutoFallback_FallbackSucceeds(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/responses", nil) + + var targets []protocol.APIType + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + targets = append(targets, target) + if target == protocol.TypeOpenAIResponses { + // In practice, upstream 404 is converted to 500 by SendStreamingError + gate.WriteHeader(http.StatusInternalServerError) + gate.Write([]byte(`{"error":"endpoint not found"}`)) + c.Error(fmt.Errorf("status 500: endpoint not found")) + return provider, "model-a", true + } + gate.WriteHeader(200) + gate.CommitFirstChunk() + return provider, "model-a", true + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIResponses, dispatch) + + if len(targets) != 2 { + t.Fatalf("dispatch called %d times, want 2", len(targets)) + } + if targets[0] != protocol.TypeOpenAIResponses { + t.Errorf("first attempt target = %v, want responses", targets[0]) + } + if targets[1] != protocol.TypeOpenAIChat { + t.Errorf("fallback target = %v, want chat", targets[1]) + } + got, ok := s.endpointCache.Get(provider.UUID, "model-a") + if !ok || got != protocol.TypeOpenAIChat { + t.Errorf("cache = (%v, %v), want (chat, true)", got, ok) + } +} + +// TestDispatchWithAutoFallback_BothFail verifies: both attempts fail → +// no cache entry written. +func TestDispatchWithAutoFallback_BothFail(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + calls := 0 + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + calls++ + gate.WriteHeader(http.StatusBadGateway) + gate.Write([]byte(`{"error":"upstream failed"}`)) + c.Error(fmt.Errorf("status 502: bad gateway")) + return provider, "model-a", true + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) + + if calls != 2 { + t.Errorf("dispatch called %d times, want 2 (initial + fallback)", calls) + } + if _, ok := s.endpointCache.Get(provider.UUID, "model-a"); ok { + t.Error("cache must stay empty when both attempts fail") + } +} + +// TestDispatchWithAutoFallback_NonRetryableSkipsFallback verifies: when +// the first attempt fails with a non-retryable error (e.g. 401), no +// fallback is attempted. +func TestDispatchWithAutoFallback_NonRetryableSkipsFallback(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + calls := 0 + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + calls++ + gate.WriteHeader(http.StatusUnauthorized) + gate.Write([]byte(`{"error":"unauthorized"}`)) + c.Error(fmt.Errorf("status 401: unauthorized")) + return provider, "model-a", true + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) + + if calls != 1 { + t.Errorf("dispatch called %d times, want 1 (no fallback for 401)", calls) + } + if _, ok := s.endpointCache.Get(provider.UUID, "model-a"); ok { + t.Error("cache must stay empty on non-retryable error") + } +} + +// TestDispatchWithAutoFallback_Status0NoRetry verifies: when the writer +// is never touched (status 0), no fallback is attempted — the handler +// ran to completion without producing output. +func TestDispatchWithAutoFallback_Status0NoRetry(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + calls := 0 + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + calls++ + // Intentionally do nothing — simulate a handler that returns without writing + return provider, "model-a", true + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) + + if calls != 1 { + t.Errorf("dispatch called %d times, want 1 (status 0 is non-retryable)", calls) + } +} + +// TestDispatchWithAutoFallback_GinErrorsClearedBetweenAttempts verifies +// that gin context errors from the first attempt are cleared before the +// fallback, so the fallback starts with a clean error slate. +func TestDispatchWithAutoFallback_GinErrorsClearedBetweenAttempts(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + if target == protocol.TypeOpenAIChat { + gate.WriteHeader(http.StatusInternalServerError) + gate.Write([]byte(`{"error":"endpoint not found"}`)) + c.Error(fmt.Errorf("status 500: endpoint not found")) + return provider, "model-a", true + } + // Fallback: verify errors were cleared + if len(c.Errors) != 0 { + t.Errorf("gin errors not cleared before fallback: %v", c.Errors) + } + gate.WriteHeader(200) + gate.CommitFirstChunk() + return provider, "model-a", true + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) +} + +// TestDispatchWithAutoFallback_NonRetryableErrorViaGinContext verifies +// that non-retryable classification uses the gin context error (not just +// status code). A 500 status with a "rate limit" error message should +// NOT trigger fallback. +func TestDispatchWithAutoFallback_NonRetryableErrorViaGinContext(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + calls := 0 + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + calls++ + gate.WriteHeader(http.StatusInternalServerError) + gate.Write([]byte(`{"error":"rate limit exceeded"}`)) + c.Error(fmt.Errorf("rate limit exceeded")) + return provider, "model-a", true + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) + + if calls != 1 { + t.Errorf("dispatch called %d times, want 1 (rate limit is non-retryable even with 500 status)", calls) + } +} + +// TestDispatchWithAutoFallback_BufferedSuccessNonStreaming verifies the +// non-streaming success path: gate is NOT committed (no CommitFirstChunk) +// but has a 200 status with body → gateSucceeded returns true → cache +// written. +func TestDispatchWithAutoFallback_BufferedSuccessNonStreaming(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + gate.WriteHeader(200) + gate.Write([]byte(`{"id":"resp-1"}`)) + // No CommitFirstChunk — simulates non-streaming response + return provider, "model-a", true + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) + + got, ok := s.endpointCache.Get(provider.UUID, "model-a") + if !ok || got != protocol.TypeOpenAIChat { + t.Errorf("cache = (%v, %v), want (chat, true) — buffered 200 should count as success", got, ok) + } +} diff --git a/internal/server/endpoint_cache.go b/internal/server/endpoint_cache.go new file mode 100644 index 000000000..c443b6471 --- /dev/null +++ b/internal/server/endpoint_cache.go @@ -0,0 +1,62 @@ +package server + +import ( + "sync" + "time" + + "github.com/tingly-dev/tingly-box/internal/protocol" +) + +const defaultEndpointCacheTTL = 24 * time.Hour + +type endpointCacheEntry struct { + target protocol.APIType + cachedAt time.Time +} + +type EndpointCache struct { + mu sync.RWMutex + store map[string]endpointCacheEntry + ttl time.Duration +} + +func NewEndpointCache(ttl time.Duration) *EndpointCache { + if ttl <= 0 { + ttl = defaultEndpointCacheTTL + } + return &EndpointCache{ + store: make(map[string]endpointCacheEntry), + ttl: ttl, + } +} + +func endpointCacheKey(providerUUID, model string) string { + return providerUUID + ":" + model +} + +func (c *EndpointCache) Get(providerUUID, model string) (protocol.APIType, bool) { + key := endpointCacheKey(providerUUID, model) + c.mu.RLock() + entry, ok := c.store[key] + c.mu.RUnlock() + if !ok { + return "", false + } + if time.Since(entry.cachedAt) > c.ttl { + c.mu.Lock() + delete(c.store, key) + c.mu.Unlock() + return "", false + } + return entry.target, true +} + +func (c *EndpointCache) Set(providerUUID, model string, target protocol.APIType) { + key := endpointCacheKey(providerUUID, model) + c.mu.Lock() + c.store[key] = endpointCacheEntry{ + target: target, + cachedAt: time.Now(), + } + c.mu.Unlock() +} diff --git a/internal/server/endpoint_cache_test.go b/internal/server/endpoint_cache_test.go new file mode 100644 index 000000000..603c83403 --- /dev/null +++ b/internal/server/endpoint_cache_test.go @@ -0,0 +1,52 @@ +package server + +import ( + "testing" + "time" + + "github.com/tingly-dev/tingly-box/internal/protocol" +) + +func TestEndpointCache_GetSet(t *testing.T) { + c := NewEndpointCache(time.Hour) + + // Miss + if _, ok := c.Get("p1", "m1"); ok { + t.Fatal("expected miss on empty cache") + } + + // Set and hit + c.Set("p1", "m1", protocol.TypeOpenAIChat) + got, ok := c.Get("p1", "m1") + if !ok || got != protocol.TypeOpenAIChat { + t.Fatalf("expected chat, got %v ok=%v", got, ok) + } + + // Different model is independent + if _, ok := c.Get("p1", "m2"); ok { + t.Fatal("expected miss for different model") + } +} + +func TestEndpointCache_TTLExpiry(t *testing.T) { + c := NewEndpointCache(10 * time.Millisecond) + + c.Set("p1", "m1", protocol.TypeOpenAIResponses) + time.Sleep(20 * time.Millisecond) + + if _, ok := c.Get("p1", "m1"); ok { + t.Fatal("expected miss after TTL expiry") + } +} + +func TestEndpointCache_Overwrite(t *testing.T) { + c := NewEndpointCache(time.Hour) + + c.Set("p1", "m1", protocol.TypeOpenAIChat) + c.Set("p1", "m1", protocol.TypeOpenAIResponses) + + got, ok := c.Get("p1", "m1") + if !ok || got != protocol.TypeOpenAIResponses { + t.Fatalf("expected responses after overwrite, got %v", got) + } +} diff --git a/internal/server/failover_dispatch.go b/internal/server/failover_dispatch.go index 5fc31b541..b8ddc2229 100644 --- a/internal/server/failover_dispatch.go +++ b/internal/server/failover_dispatch.go @@ -336,19 +336,42 @@ func (s *Server) dispatchWithPriorityFailover( initialModel string, attempt dispatchAttempt, ) { + s.dispatchWithPriorityFailoverGated(c, rule, initialProvider, initialModel, attempt, nil) +} + +// dispatchWithPriorityFailoverGated is the gated variant. When +// externalGate is non-nil the caller owns the gate lifecycle (install on +// c.Writer, defer restore+commit); the function reuses it instead of +// creating a new one. Pass nil for the original self-managed behavior. +// +// Returns the provider/model of the final attempt so callers that key +// state on the serving identity (e.g. the endpoint auto-detection cache) +// attribute it to the service that actually handled the request, not the +// initially selected one. +func (s *Server) dispatchWithPriorityFailoverGated( + c *gin.Context, + rule *typ.Rule, + initialProvider *typ.Provider, + initialModel string, + attempt dispatchAttempt, + externalGate *firstChunkGate, +) (*typ.Provider, string) { activeServices := rule.GetActiveServices() if len(activeServices) <= 1 { attempt(initialProvider, initialModel) - return + return initialProvider, initialModel } - realWriter := c.Writer - gate := newFirstChunkGate(realWriter) - c.Writer = gate - defer func() { - c.Writer = realWriter - gate.CommitIfBuffered() - }() + gate := externalGate + if gate == nil { + realWriter := c.Writer + gate = newFirstChunkGate(realWriter) + c.Writer = gate + defer func() { + c.Writer = realWriter + gate.CommitIfBuffered() + }() + } tried := map[string]bool{} provider := initialProvider @@ -392,7 +415,7 @@ func (s *Server) dispatchWithPriorityFailover( "provider": provider.Name, "model": model, }).Infof("[failover] succeeded on attempt %d with %s/%s", i+1, provider.UUID, model) - return + return provider, model } status := gate.Status() if !isRetryableStatus(status) { @@ -405,7 +428,7 @@ func (s *Server) dispatchWithPriorityFailover( "model": model, "status": status, }).Warnf("[failover] attempt %d returned status %d with %s/%s", i+1, status, provider.UUID, model) - return + return provider, model } // Pass "" so the candidate pool spans all API styles: each attempt @@ -420,7 +443,7 @@ func (s *Server) dispatchWithPriorityFailover( "status": status, "error": err.Error(), }).Warnf("[failover] load balancer failed selecting fallback after %d attempt(s) status=%d: %v", i+1, status, err) - return + return provider, model } if nextProvider == nil || nextService == nil { logrus.WithContext(c.Request.Context()).WithFields(logrus.Fields{ @@ -429,7 +452,7 @@ func (s *Server) dispatchWithPriorityFailover( "total_attempts": len(activeServices), "status": status, }).Warnf("[failover] giving up after %d attempt(s) status=%d (no more services)", i+1, status) - return + return provider, model } nextServiceID := loadbalance.FormatServiceID(nextProvider.UUID, nextService.Model) @@ -449,4 +472,5 @@ func (s *Server) dispatchWithPriorityFailover( provider = nextProvider model = nextService.Model } + return provider, model } diff --git a/internal/server/openai_chat.go b/internal/server/openai_chat.go index abdb59149..ae42480f8 100644 --- a/internal/server/openai_chat.go +++ b/internal/server/openai_chat.go @@ -6,6 +6,7 @@ import ( "net/http" "github.com/gin-gonic/gin" + "github.com/tingly-dev/tingly-box/ai" "github.com/tingly-dev/tingly-box/internal/constant" "github.com/tingly-dev/tingly-box/internal/loadbalance" "github.com/tingly-dev/tingly-box/internal/protocol" @@ -163,26 +164,56 @@ func (s *Server) OpenAIChatCompletion(c *gin.Context, req *protocol.OpenAIChatCo template = bs } + // ── Endpoint auto-detection (OpenAI providers in auto mode, flag on) ── + // resolveAutoTarget returns the preferred protocol and whether runtime + // fallback should wrap the dispatch. An override or cache hit yields a + // definite target (autoFallback=false) that is simply pinned; a cache + // miss yields the scenario-preferred protocol with fallback enabled. + var autoPreferred protocol.APIType + autoFallback := false + if s.autoEndpointEnabled() && provider.APIStyle == protocol.APIStyleOpenAI && ai.IsAutoEndpointMode(provider.OpenAIEndpointMode) { + autoPreferred, autoFallback = s.resolveAutoTarget(resolveRuleFlags(c, rule), provider, actualModel, scenarioType, IncomingAPIChat) + } + // ── Per-attempt pipeline (provider-dependent) ── - s.dispatchWithPriorityFailover(c, rule, provider, actualModel, - func(p *typ.Provider, retryModel string) { - areq := req - if multi { - cloned, err := cloneOpenAIChatRequest(template) - if err != nil { - s.failAttemptSetup(c, err) - return - } - areq = cloned + attempt := func(p *typ.Provider, retryModel string, tgt protocol.APIType) { + areq := req + if multi { + cloned, err := cloneOpenAIChatRequest(template) + if err != nil { + s.failAttemptSetup(c, err) + return } - s.runOpenAIChatAttempt(c, areq, responseModel, p, retryModel, rule, isStreaming, scenarioType, scenarioConfig) - }) + areq = cloned + } + s.runOpenAIChatAttempt(c, areq, responseModel, p, retryModel, rule, isStreaming, scenarioType, scenarioConfig, tgt) + } + + if autoFallback { + // Protocol fallback is the outer loop, provider failover the inner + // loop — both share the single gate owned by dispatchWithAutoFallback. + s.dispatchWithAutoFallback(c, provider, actualModel, autoPreferred, + func(t protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + served, servedModel := s.dispatchWithPriorityFailoverGated(c, rule, provider, actualModel, + func(p *typ.Provider, retryModel string) { attempt(p, retryModel, t) }, gate) + return served, servedModel, true + }) + return + } + + // Non-auto path (autoPreferred is "" → normal resolution; or a pinned + // override/cache target when auto mode resolved without fallback). + s.dispatchWithPriorityFailover(c, rule, provider, actualModel, + func(p *typ.Provider, retryModel string) { attempt(p, retryModel, autoPreferred) }) } // runOpenAIChatAttempt executes the provider-dependent half of an OpenAI chat // request for one failover attempt. Setup failures route through // failAttemptSetup so the orchestrator can advance to the next candidate. -func (s *Server) runOpenAIChatAttempt(c *gin.Context, req *protocol.OpenAIChatCompletionRequest, responseModel string, provider *typ.Provider, actualModel string, rule *typ.Rule, isStreaming bool, scenarioType typ.RuleScenario, scenarioConfig *typ.ScenarioConfig) { +// overrideTarget, when non-empty, forces the OpenAI upstream endpoint for this +// attempt instead of resolving it from the provider mode — used by endpoint +// auto-detection to pin a protocol hypothesis across a failover round. +func (s *Server) runOpenAIChatAttempt(c *gin.Context, req *protocol.OpenAIChatCompletionRequest, responseModel string, provider *typ.Provider, actualModel string, rule *typ.Rule, isStreaming bool, scenarioType typ.RuleScenario, scenarioConfig *typ.ScenarioConfig, overrideTarget protocol.APIType) { // Resolve dual endpoint: when the provider has an OpenAI-compatible // dual URL configured, route there natively to avoid a transform. provider = provider.ResolveStyle(protocol.APIStyleOpenAI) @@ -209,14 +240,19 @@ func (s *Server) runOpenAIChatAttempt(c *gin.Context, req *protocol.OpenAIChatCo case protocol.APIStyleGoogle: target = protocol.TypeGoogle case protocol.APIStyleOpenAI: - // Need flags for endpoint resolution, but we'll re-resolve with scenario after target is determined - tempFlags := resolveRuleFlags(c, rule) - resolvedTarget, routeErr := ResolveOpenAIEndpoint(provider, tempFlags, IncomingAPIChat) - if routeErr != nil { - s.failAttemptSetup(c, routeErr) - return + if overrideTarget != "" { + // Endpoint auto-detection pinned the protocol for this attempt. + target = overrideTarget + } else { + // Need flags for endpoint resolution, but we'll re-resolve with scenario after target is determined + tempFlags := resolveRuleFlags(c, rule) + resolvedTarget, routeErr := ResolveOpenAIEndpoint(provider, tempFlags, IncomingAPIChat) + if routeErr != nil { + s.failAttemptSetup(c, routeErr) + return + } + target = resolvedTarget } - target = resolvedTarget default: s.failAttemptSetup(c, fmt.Errorf("Unsupported API style: %s %s", provider.Name, apiStyle)) return diff --git a/internal/server/openai_responses.go b/internal/server/openai_responses.go index b2d3921c7..1b616fc91 100644 --- a/internal/server/openai_responses.go +++ b/internal/server/openai_responses.go @@ -9,6 +9,7 @@ import ( "github.com/openai/openai-go/v3/packages/param" "github.com/openai/openai-go/v3/responses" + "github.com/tingly-dev/tingly-box/ai" "github.com/tingly-dev/tingly-box/internal/constant" "github.com/tingly-dev/tingly-box/internal/loadbalance" "github.com/tingly-dev/tingly-box/internal/protocol" @@ -170,26 +171,49 @@ func (s *Server) ResponsesCreate(c *gin.Context, scenarioType typ.RuleScenario, // PreprocessInputData and vision proxy are not re-run). multi := len(rule.GetActiveServices()) > 1 + // ── Endpoint auto-detection (OpenAI providers in auto mode, flag on) ── + var autoPreferred protocol.APIType + autoFallback := false + if s.autoEndpointEnabled() && provider.APIStyle == protocol.APIStyleOpenAI && ai.IsAutoEndpointMode(provider.OpenAIEndpointMode) { + autoPreferred, autoFallback = s.resolveAutoTarget(resolveRuleFlags(c, rule), provider, actualModel, scenarioType, IncomingAPIResponses) + } + // ── Per-attempt pipeline (provider-dependent) ── - s.dispatchWithPriorityFailover(c, rule, provider, actualModel, - func(p *typ.Provider, retryModel string) { - areq := req - if multi { - clonedParams, err := cloneResponsesParams(req.ResponseNewParams) - if err != nil { - s.failAttemptSetup(c, err) - return - } - areq.ResponseNewParams = clonedParams + attempt := func(p *typ.Provider, retryModel string, tgt protocol.APIType) { + areq := req + if multi { + clonedParams, err := cloneResponsesParams(req.ResponseNewParams) + if err != nil { + s.failAttemptSetup(c, err) + return } - s.runOpenAIResponsesAttempt(c, areq, p, retryModel, rule, isStreaming, scenarioType, scenarioConfig) - }) + areq.ResponseNewParams = clonedParams + } + s.runOpenAIResponsesAttempt(c, areq, p, retryModel, rule, isStreaming, scenarioType, scenarioConfig, tgt) + } + + if autoFallback { + // Protocol fallback outer, provider failover inner — one shared gate. + s.dispatchWithAutoFallback(c, provider, actualModel, autoPreferred, + func(t protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + served, servedModel := s.dispatchWithPriorityFailoverGated(c, rule, provider, actualModel, + func(p *typ.Provider, retryModel string) { attempt(p, retryModel, t) }, gate) + return served, servedModel, true + }) + return + } + + s.dispatchWithPriorityFailover(c, rule, provider, actualModel, + func(p *typ.Provider, retryModel string) { attempt(p, retryModel, autoPreferred) }) } // runOpenAIResponsesAttempt executes the provider-dependent half of an OpenAI // Responses request for one failover attempt. Setup failures route through // failAttemptSetup so the orchestrator can advance to the next candidate. -func (s *Server) runOpenAIResponsesAttempt(c *gin.Context, req *protocol.ResponseCreateRequest, provider *typ.Provider, actualModel string, rule *typ.Rule, isStreaming bool, scenarioType typ.RuleScenario, scenarioConfig *typ.ScenarioConfig) { +// overrideTarget, when non-empty, forces the OpenAI upstream endpoint for this +// attempt instead of resolving it from the provider mode — used by endpoint +// auto-detection to pin a protocol hypothesis across a failover round. +func (s *Server) runOpenAIResponsesAttempt(c *gin.Context, req *protocol.ResponseCreateRequest, provider *typ.Provider, actualModel string, rule *typ.Rule, isStreaming bool, scenarioType typ.RuleScenario, scenarioConfig *typ.ScenarioConfig, overrideTarget protocol.APIType) { // Resolve dual endpoint: when the provider has an OpenAI-compatible // dual URL configured, route there natively to avoid a transform. provider = provider.ResolveStyle(protocol.APIStyleOpenAI) @@ -209,12 +233,17 @@ func (s *Server) runOpenAIResponsesAttempt(c *gin.Context, req *protocol.Respons s.failAttemptSetup(c, fmt.Errorf("Responses API does not support Google-style providers yet. Provider: %s", provider.Name)) return case protocol.APIStyleOpenAI: - resolvedTarget, routeErr := ResolveOpenAIEndpoint(provider, resolveRuleFlags(c, rule), IncomingAPIResponses) - if routeErr != nil { - s.failAttemptSetup(c, routeErr) - return + if overrideTarget != "" { + // Endpoint auto-detection pinned the protocol for this attempt. + target = overrideTarget + } else { + resolvedTarget, routeErr := ResolveOpenAIEndpoint(provider, resolveRuleFlags(c, rule), IncomingAPIResponses) + if routeErr != nil { + s.failAttemptSetup(c, routeErr) + return + } + target = resolvedTarget } - target = resolvedTarget default: s.failAttemptSetup(c, fmt.Errorf("Unsupported provider API style: %s", provider.APIStyle)) return diff --git a/internal/server/protocol_endpoint.go b/internal/server/protocol_endpoint.go index c4273f97d..2d69dc202 100644 --- a/internal/server/protocol_endpoint.go +++ b/internal/server/protocol_endpoint.go @@ -73,7 +73,11 @@ func ResolveOpenAIEndpoint(provider *typ.Provider, flags typ.RuleFlags, incoming switch mode { case ai.EndpointModeResponses: return protocol.TypeOpenAIResponses, nil - case ai.EndpointModeBoth: + case ai.EndpointModeBoth, ai.EndpointModeAuto: + // Both mirrors the incoming API. Auto also mirrors incoming here — + // this is the non-auto fallback path taken when runtime endpoint + // auto-detection is gated off; when enabled, resolveAutoTarget + + // dispatchWithAutoFallback handle Auto providers instead. if incoming == IncomingAPIResponses { return protocol.TypeOpenAIResponses, nil } diff --git a/internal/server/server.go b/internal/server/server.go index d9d947975..9fb0cb909 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -126,6 +126,9 @@ type Server struct { scenarioRecordSinks map[typ.RuleScenario]*obs.Sink scenarioRecordSinksMu sync.RWMutex + // endpoint cache for auto endpoint mode (per provider+model → protocol) + endpointCache *EndpointCache + // affinity store for smart routing session-model locking affinityStore *AffinityStore @@ -243,6 +246,7 @@ func NewServer(cfg *config.Config, opts ...ServerOption) *Server { server.jwtManager = jwtManager server.engine = gin.New() server.clientPool = client.NewClientPool() + server.endpointCache = NewEndpointCache(defaultEndpointCacheTTL) server.scenarioRecordSinks = make(map[typ.RuleScenario]*obs.Sink) historyStore := guardrailsutils.NewStore(200, GetGuardrailsHistoryPath(cfg.ConfigDir)) grRuntime := server.currentGuardrailsRuntime() @@ -379,6 +383,7 @@ func NewServer(cfg *config.Config, opts ...ServerOption) *Server { // E2E probe service handles /api/v2/probe end-to-end without touching *Server. server.probeE2EService = probe.NewE2EService(cfg, server.clientPool) + server.probeE2EService.SetEndpointCache(server.endpointCache.Get, server.endpointCache.Set) server.probeLightweight = probe.NewLightweightService(server.clientPool) // Initialize OTel meter setup for token tracking diff --git a/internal/server/server_flags.go b/internal/server/server_flags.go index 0f3dae3ee..997bf55d5 100644 --- a/internal/server/server_flags.go +++ b/internal/server/server_flags.go @@ -30,6 +30,14 @@ func (s *Server) mcpEnabled() bool { s.config.GetScenarioFlag(typ.ScenarioClaudeCode, config.ExtensionMCP) } +// autoEndpointEnabled checks if auto endpoint detection is enabled via scenario flag. +func (s *Server) autoEndpointEnabled() bool { + if s.config == nil { + return false + } + return s.config.GetScenarioFlag(typ.ScenarioGlobal, config.ExtensionAutoEndpoint) +} + func (s *Server) initGuardrailsRuntime() { runtime := s.currentGuardrailsRuntime() if (runtime != nil && runtime.PolicyEngine() != nil) || s.config == nil { From 907d13148ca0a129cc3c0e5adfea88db3d8a2821 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 03:33:16 +0000 Subject: [PATCH 2/3] simplify(endpoint-auto): dedupe dispatch wiring, drop dead ok flag - Extract autoDispatchOrFailover: the ~25-line "auto vs plain failover" decision was duplicated verbatim in openai_chat.go and openai_responses.go, differing only by the IncomingAPIType constant and inner attempt body. Both call sites now share one helper. - Drop the ok bool from autoDispatchFn. Every real call site returned ok=true unconditionally (dispatchWithPriorityFailoverGated never yields a nil provider), so served != nil alone was already the complete success signal; ok was a vestige of an earlier direct-transform design. No behavior change; verified via existing endpoint_auto_test.go suite. https://claude.ai/code/session_01N9JnUTVt41NcGxYyj3xewf --- internal/server/endpoint_auto.go | 54 ++++++++++++++++++++++---- internal/server/endpoint_auto_test.go | 48 +++++++++++------------ internal/server/openai_chat.go | 55 +++++++-------------------- internal/server/openai_responses.go | 48 +++++++---------------- 4 files changed, 98 insertions(+), 107 deletions(-) diff --git a/internal/server/endpoint_auto.go b/internal/server/endpoint_auto.go index 51ba00966..a443c5ec4 100644 --- a/internal/server/endpoint_auto.go +++ b/internal/server/endpoint_auto.go @@ -5,6 +5,7 @@ import ( "github.com/gin-gonic/gin" "github.com/sirupsen/logrus" + "github.com/tingly-dev/tingly-box/ai" "github.com/tingly-dev/tingly-box/internal/client" "github.com/tingly-dev/tingly-box/internal/protocol" "github.com/tingly-dev/tingly-box/internal/typ" @@ -84,10 +85,10 @@ func (s *Server) resolveAutoTarget( // autoDispatchFn is the callback for dispatchWithAutoFallback. // It performs transform + dispatch for a given target protocol, using // the provided gate. Returns the provider/model that served the final -// attempt (failover may have moved past the initially selected service) -// and ok=true if dispatch executed (even on error); ok=false means the -// transform itself failed. -type autoDispatchFn func(target protocol.APIType, gate *firstChunkGate) (served *typ.Provider, servedModel string, ok bool) +// attempt (failover may have moved past the initially selected service); +// served is nil when dispatch never got a servable provider (e.g. the +// transform itself failed before failover could run). +type autoDispatchFn func(target protocol.APIType, gate *firstChunkGate) (served *typ.Provider, servedModel string) // gateSucceeded reports whether the attempt behind the gate produced a // success: either the stream committed its first chunk, or a buffered @@ -119,10 +120,10 @@ func (s *Server) dispatchWithAutoFallback( }() // First attempt with preferred protocol - served, servedModel, ok := dispatch(preferredTarget, gate) + served, servedModel := dispatch(preferredTarget, gate) if gateSucceeded(gate) { - if ok && served != nil { + if served != nil { s.endpointCache.Set(served.UUID, servedModel, preferredTarget) } return @@ -146,9 +147,46 @@ func (s *Server) dispatchWithAutoFallback( gate.Discard() clearGinErrors(c) - served, servedModel, ok = dispatch(altTarget, gate) + served, servedModel = dispatch(altTarget, gate) - if gateSucceeded(gate) && ok && served != nil { + if gateSucceeded(gate) && served != nil { s.endpointCache.Set(served.UUID, servedModel, altTarget) } } + +// autoDispatchOrFailover decides whether endpoint auto-detection applies to +// this request and dispatches accordingly. With auto-detection, provider +// failover runs nested inside protocol fallback, both sharing one gate +// (owned by dispatchWithAutoFallback). Without it, this is a plain provider +// failover using the statically resolved target (which may be "" — the +// zero value protocol.APIType — meaning attempt must resolve it itself). +// +// attempt is the per-provider-attempt callback; its target argument is the +// protocol to use for that attempt. +func (s *Server) autoDispatchOrFailover( + c *gin.Context, + rule *typ.Rule, + provider *typ.Provider, + actualModel string, + scenarioType typ.RuleScenario, + incoming IncomingAPIType, + attempt func(p *typ.Provider, retryModel string, target protocol.APIType), +) { + var target protocol.APIType + autoFallback := false + if s.autoEndpointEnabled() && provider.APIStyle == protocol.APIStyleOpenAI && ai.IsAutoEndpointMode(provider.OpenAIEndpointMode) { + target, autoFallback = s.resolveAutoTarget(resolveRuleFlags(c, rule), provider, actualModel, scenarioType, incoming) + } + + if autoFallback { + s.dispatchWithAutoFallback(c, provider, actualModel, target, + func(t protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { + return s.dispatchWithPriorityFailoverGated(c, rule, provider, actualModel, + func(p *typ.Provider, retryModel string) { attempt(p, retryModel, t) }, gate) + }) + return + } + + s.dispatchWithPriorityFailover(c, rule, provider, actualModel, + func(p *typ.Provider, retryModel string) { attempt(p, retryModel, target) }) +} diff --git a/internal/server/endpoint_auto_test.go b/internal/server/endpoint_auto_test.go index 0b59b35fd..ed1841e57 100644 --- a/internal/server/endpoint_auto_test.go +++ b/internal/server/endpoint_auto_test.go @@ -68,10 +68,10 @@ func TestDispatchWithAutoFallback_CacheAttributedToServingProvider(t *testing.T) // Simulate a dispatch where failover moved past the initial provider: // the gate commits (success) and the serving identity differs from initial. - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { gate.WriteHeader(200) gate.CommitFirstChunk() - return serving, "served-model", true + return serving, "served-model" } s.dispatchWithAutoFallback(c, initial, "req-model", protocol.TypeOpenAIChat, dispatch) @@ -89,8 +89,8 @@ func TestDispatchWithAutoFallback_CacheAttributedToServingProvider(t *testing.T) } // TestDispatchWithAutoFallback_NoCacheOnTransformFailure ensures a failed -// transform (ok=false, served=nil) never writes a cache entry even if the -// gate state looks successful. +// transform (served=nil) never writes a cache entry even if the gate state +// looks successful. func TestDispatchWithAutoFallback_NoCacheOnTransformFailure(t *testing.T) { s := &Server{endpointCache: NewEndpointCache(0)} initial := &typ.Provider{UUID: "prov-initial"} @@ -99,9 +99,9 @@ func TestDispatchWithAutoFallback_NoCacheOnTransformFailure(t *testing.T) { c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { gate.WriteHeader(200) // transform error path writes a JSON error; simulate benign status - return nil, "", false + return nil, "" } s.dispatchWithAutoFallback(c, initial, "m", protocol.TypeOpenAIChat, dispatch) @@ -122,11 +122,11 @@ func TestDispatchWithAutoFallback_FirstAttemptSucceeds(t *testing.T) { c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) calls := 0 - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { calls++ gate.WriteHeader(200) gate.CommitFirstChunk() - return provider, "model-a", true + return provider, "model-a" } s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIResponses, dispatch) @@ -152,18 +152,18 @@ func TestDispatchWithAutoFallback_FallbackSucceeds(t *testing.T) { c.Request = httptest.NewRequest("POST", "/v1/responses", nil) var targets []protocol.APIType - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { targets = append(targets, target) if target == protocol.TypeOpenAIResponses { // In practice, upstream 404 is converted to 500 by SendStreamingError gate.WriteHeader(http.StatusInternalServerError) gate.Write([]byte(`{"error":"endpoint not found"}`)) c.Error(fmt.Errorf("status 500: endpoint not found")) - return provider, "model-a", true + return provider, "model-a" } gate.WriteHeader(200) gate.CommitFirstChunk() - return provider, "model-a", true + return provider, "model-a" } s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIResponses, dispatch) @@ -194,12 +194,12 @@ func TestDispatchWithAutoFallback_BothFail(t *testing.T) { c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) calls := 0 - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { calls++ gate.WriteHeader(http.StatusBadGateway) gate.Write([]byte(`{"error":"upstream failed"}`)) c.Error(fmt.Errorf("status 502: bad gateway")) - return provider, "model-a", true + return provider, "model-a" } s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) @@ -224,12 +224,12 @@ func TestDispatchWithAutoFallback_NonRetryableSkipsFallback(t *testing.T) { c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) calls := 0 - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { calls++ gate.WriteHeader(http.StatusUnauthorized) gate.Write([]byte(`{"error":"unauthorized"}`)) c.Error(fmt.Errorf("status 401: unauthorized")) - return provider, "model-a", true + return provider, "model-a" } s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) @@ -254,10 +254,10 @@ func TestDispatchWithAutoFallback_Status0NoRetry(t *testing.T) { c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) calls := 0 - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { calls++ // Intentionally do nothing — simulate a handler that returns without writing - return provider, "model-a", true + return provider, "model-a" } s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) @@ -278,12 +278,12 @@ func TestDispatchWithAutoFallback_GinErrorsClearedBetweenAttempts(t *testing.T) c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { if target == protocol.TypeOpenAIChat { gate.WriteHeader(http.StatusInternalServerError) gate.Write([]byte(`{"error":"endpoint not found"}`)) c.Error(fmt.Errorf("status 500: endpoint not found")) - return provider, "model-a", true + return provider, "model-a" } // Fallback: verify errors were cleared if len(c.Errors) != 0 { @@ -291,7 +291,7 @@ func TestDispatchWithAutoFallback_GinErrorsClearedBetweenAttempts(t *testing.T) } gate.WriteHeader(200) gate.CommitFirstChunk() - return provider, "model-a", true + return provider, "model-a" } s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) @@ -310,12 +310,12 @@ func TestDispatchWithAutoFallback_NonRetryableErrorViaGinContext(t *testing.T) { c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) calls := 0 - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { calls++ gate.WriteHeader(http.StatusInternalServerError) gate.Write([]byte(`{"error":"rate limit exceeded"}`)) c.Error(fmt.Errorf("rate limit exceeded")) - return provider, "model-a", true + return provider, "model-a" } s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) @@ -337,11 +337,11 @@ func TestDispatchWithAutoFallback_BufferedSuccessNonStreaming(t *testing.T) { c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { gate.WriteHeader(200) gate.Write([]byte(`{"id":"resp-1"}`)) // No CommitFirstChunk — simulates non-streaming response - return provider, "model-a", true + return provider, "model-a" } s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) diff --git a/internal/server/openai_chat.go b/internal/server/openai_chat.go index ae42480f8..4fecd6901 100644 --- a/internal/server/openai_chat.go +++ b/internal/server/openai_chat.go @@ -6,7 +6,6 @@ import ( "net/http" "github.com/gin-gonic/gin" - "github.com/tingly-dev/tingly-box/ai" "github.com/tingly-dev/tingly-box/internal/constant" "github.com/tingly-dev/tingly-box/internal/loadbalance" "github.com/tingly-dev/tingly-box/internal/protocol" @@ -164,47 +163,21 @@ func (s *Server) OpenAIChatCompletion(c *gin.Context, req *protocol.OpenAIChatCo template = bs } - // ── Endpoint auto-detection (OpenAI providers in auto mode, flag on) ── - // resolveAutoTarget returns the preferred protocol and whether runtime - // fallback should wrap the dispatch. An override or cache hit yields a - // definite target (autoFallback=false) that is simply pinned; a cache - // miss yields the scenario-preferred protocol with fallback enabled. - var autoPreferred protocol.APIType - autoFallback := false - if s.autoEndpointEnabled() && provider.APIStyle == protocol.APIStyleOpenAI && ai.IsAutoEndpointMode(provider.OpenAIEndpointMode) { - autoPreferred, autoFallback = s.resolveAutoTarget(resolveRuleFlags(c, rule), provider, actualModel, scenarioType, IncomingAPIChat) - } - - // ── Per-attempt pipeline (provider-dependent) ── - attempt := func(p *typ.Provider, retryModel string, tgt protocol.APIType) { - areq := req - if multi { - cloned, err := cloneOpenAIChatRequest(template) - if err != nil { - s.failAttemptSetup(c, err) - return + // ── Per-attempt pipeline (provider-dependent), routed through endpoint + // auto-detection when applicable ── + s.autoDispatchOrFailover(c, rule, provider, actualModel, scenarioType, IncomingAPIChat, + func(p *typ.Provider, retryModel string, tgt protocol.APIType) { + areq := req + if multi { + cloned, err := cloneOpenAIChatRequest(template) + if err != nil { + s.failAttemptSetup(c, err) + return + } + areq = cloned } - areq = cloned - } - s.runOpenAIChatAttempt(c, areq, responseModel, p, retryModel, rule, isStreaming, scenarioType, scenarioConfig, tgt) - } - - if autoFallback { - // Protocol fallback is the outer loop, provider failover the inner - // loop — both share the single gate owned by dispatchWithAutoFallback. - s.dispatchWithAutoFallback(c, provider, actualModel, autoPreferred, - func(t protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { - served, servedModel := s.dispatchWithPriorityFailoverGated(c, rule, provider, actualModel, - func(p *typ.Provider, retryModel string) { attempt(p, retryModel, t) }, gate) - return served, servedModel, true - }) - return - } - - // Non-auto path (autoPreferred is "" → normal resolution; or a pinned - // override/cache target when auto mode resolved without fallback). - s.dispatchWithPriorityFailover(c, rule, provider, actualModel, - func(p *typ.Provider, retryModel string) { attempt(p, retryModel, autoPreferred) }) + s.runOpenAIChatAttempt(c, areq, responseModel, p, retryModel, rule, isStreaming, scenarioType, scenarioConfig, tgt) + }) } // runOpenAIChatAttempt executes the provider-dependent half of an OpenAI chat diff --git a/internal/server/openai_responses.go b/internal/server/openai_responses.go index 1b616fc91..f97b6ceb0 100644 --- a/internal/server/openai_responses.go +++ b/internal/server/openai_responses.go @@ -9,7 +9,6 @@ import ( "github.com/openai/openai-go/v3/packages/param" "github.com/openai/openai-go/v3/responses" - "github.com/tingly-dev/tingly-box/ai" "github.com/tingly-dev/tingly-box/internal/constant" "github.com/tingly-dev/tingly-box/internal/loadbalance" "github.com/tingly-dev/tingly-box/internal/protocol" @@ -171,40 +170,21 @@ func (s *Server) ResponsesCreate(c *gin.Context, scenarioType typ.RuleScenario, // PreprocessInputData and vision proxy are not re-run). multi := len(rule.GetActiveServices()) > 1 - // ── Endpoint auto-detection (OpenAI providers in auto mode, flag on) ── - var autoPreferred protocol.APIType - autoFallback := false - if s.autoEndpointEnabled() && provider.APIStyle == protocol.APIStyleOpenAI && ai.IsAutoEndpointMode(provider.OpenAIEndpointMode) { - autoPreferred, autoFallback = s.resolveAutoTarget(resolveRuleFlags(c, rule), provider, actualModel, scenarioType, IncomingAPIResponses) - } - - // ── Per-attempt pipeline (provider-dependent) ── - attempt := func(p *typ.Provider, retryModel string, tgt protocol.APIType) { - areq := req - if multi { - clonedParams, err := cloneResponsesParams(req.ResponseNewParams) - if err != nil { - s.failAttemptSetup(c, err) - return + // ── Per-attempt pipeline (provider-dependent), routed through endpoint + // auto-detection when applicable ── + s.autoDispatchOrFailover(c, rule, provider, actualModel, scenarioType, IncomingAPIResponses, + func(p *typ.Provider, retryModel string, tgt protocol.APIType) { + areq := req + if multi { + clonedParams, err := cloneResponsesParams(req.ResponseNewParams) + if err != nil { + s.failAttemptSetup(c, err) + return + } + areq.ResponseNewParams = clonedParams } - areq.ResponseNewParams = clonedParams - } - s.runOpenAIResponsesAttempt(c, areq, p, retryModel, rule, isStreaming, scenarioType, scenarioConfig, tgt) - } - - if autoFallback { - // Protocol fallback outer, provider failover inner — one shared gate. - s.dispatchWithAutoFallback(c, provider, actualModel, autoPreferred, - func(t protocol.APIType, gate *firstChunkGate) (*typ.Provider, string, bool) { - served, servedModel := s.dispatchWithPriorityFailoverGated(c, rule, provider, actualModel, - func(p *typ.Provider, retryModel string) { attempt(p, retryModel, t) }, gate) - return served, servedModel, true - }) - return - } - - s.dispatchWithPriorityFailover(c, rule, provider, actualModel, - func(p *typ.Provider, retryModel string) { attempt(p, retryModel, autoPreferred) }) + s.runOpenAIResponsesAttempt(c, areq, p, retryModel, rule, isStreaming, scenarioType, scenarioConfig, tgt) + }) } // runOpenAIResponsesAttempt executes the provider-dependent half of an OpenAI From 33682e1047da1c2dabe4a7d5857d98fabab5726e Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 07:37:44 +0000 Subject: [PATCH 3/3] chore(endpoint): merge endpoint_auto/endpoint_cache into protocol_endpoint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit endpoint_auto.go, endpoint_cache.go and their tests were split out of the same feature (OpenAI endpoint resolution + auto-detection) that already lives in protocol_endpoint.go. Four files for one cohesive concern was unnecessary fragmentation — folded them back in, organized under section banners (static resolution / cache / runtime auto-detection dispatch). No code changes beyond the move; verified via full endpoint_auto_test.go suite now inlined in protocol_endpoint_test.go. https://claude.ai/code/session_01N9JnUTVt41NcGxYyj3xewf --- internal/server/endpoint_auto.go | 192 ----------- internal/server/endpoint_auto_test.go | 353 ------------------- internal/server/endpoint_cache.go | 62 ---- internal/server/endpoint_cache_test.go | 52 --- internal/server/protocol_endpoint.go | 247 ++++++++++++++ internal/server/protocol_endpoint_test.go | 398 ++++++++++++++++++++++ 6 files changed, 645 insertions(+), 659 deletions(-) delete mode 100644 internal/server/endpoint_auto.go delete mode 100644 internal/server/endpoint_auto_test.go delete mode 100644 internal/server/endpoint_cache.go delete mode 100644 internal/server/endpoint_cache_test.go diff --git a/internal/server/endpoint_auto.go b/internal/server/endpoint_auto.go deleted file mode 100644 index a443c5ec4..000000000 --- a/internal/server/endpoint_auto.go +++ /dev/null @@ -1,192 +0,0 @@ -package server - -import ( - "net/http" - - "github.com/gin-gonic/gin" - "github.com/sirupsen/logrus" - "github.com/tingly-dev/tingly-box/ai" - "github.com/tingly-dev/tingly-box/internal/client" - "github.com/tingly-dev/tingly-box/internal/protocol" - "github.com/tingly-dev/tingly-box/internal/typ" -) - -// alternateOpenAIProtocol returns the other OpenAI protocol type. -func alternateOpenAIProtocol(current protocol.APIType) protocol.APIType { - if current == protocol.TypeOpenAIResponses { - return protocol.TypeOpenAIChat - } - return protocol.TypeOpenAIResponses -} - -// incomingToTarget maps IncomingAPIType to protocol.APIType. -func incomingToTarget(incoming IncomingAPIType) protocol.APIType { - if incoming == IncomingAPIResponses { - return protocol.TypeOpenAIResponses - } - return protocol.TypeOpenAIChat -} - -// extractLastGinError returns the most recent error recorded on the gin -// context via c.Error(). Returns nil when no errors exist. -func extractLastGinError(c *gin.Context) error { - errs := c.Errors - if len(errs) == 0 { - return nil - } - return errs[len(errs)-1].Err -} - -// clearGinErrors removes all errors from the gin context so that a -// fallback retry starts with a clean slate. -func clearGinErrors(c *gin.Context) { - c.Errors = c.Errors[:0] -} - -// overrideToTarget converts an EndpointOverride to a protocol.APIType. -func overrideToTarget(ov EndpointOverride) protocol.APIType { - if ov == OverrideResponses { - return protocol.TypeOpenAIResponses - } - return protocol.TypeOpenAIChat -} - -// scenarioPreferredProtocol returns the protocol an auto-mode provider -// should try first on a cache miss. Scenarios whose client ecosystem is -// natively Responses-based (Codex) start with Responses regardless of the -// incoming transport — providers serving such traffic overwhelmingly speak -// Responses, so leading with it saves a wasted first round trip. All other -// scenarios mirror the incoming API. -func scenarioPreferredProtocol(scenario typ.RuleScenario, incoming IncomingAPIType) protocol.APIType { - switch scenario.Base() { - case typ.ScenarioCodex: - return protocol.TypeOpenAIResponses - default: - return incomingToTarget(incoming) - } -} - -// resolveAutoTarget handles the auto-mode target resolution shared by both -// OpenAI Chat and Responses handlers. It checks override → cache → -// scenario-preferred default. Returns the resolved target and whether -// auto-fallback should be enabled. -func (s *Server) resolveAutoTarget( - flags typ.RuleFlags, provider *typ.Provider, model string, scenario typ.RuleScenario, incoming IncomingAPIType, -) (target protocol.APIType, autoFallback bool) { - if ov := ParseEndpointOverride(flags.OpenAIEndpointOverride); ov == OverrideChat || ov == OverrideResponses { - return overrideToTarget(ov), false - } - if cached, ok := s.endpointCache.Get(provider.UUID, model); ok { - return cached, false - } - return scenarioPreferredProtocol(scenario, incoming), true -} - -// autoDispatchFn is the callback for dispatchWithAutoFallback. -// It performs transform + dispatch for a given target protocol, using -// the provided gate. Returns the provider/model that served the final -// attempt (failover may have moved past the initially selected service); -// served is nil when dispatch never got a servable provider (e.g. the -// transform itself failed before failover could run). -type autoDispatchFn func(target protocol.APIType, gate *firstChunkGate) (served *typ.Provider, servedModel string) - -// gateSucceeded reports whether the attempt behind the gate produced a -// success: either the stream committed its first chunk, or a buffered -// non-error status is waiting to flush. -func gateSucceeded(gate *firstChunkGate) bool { - return gate.Committed() || (gate.Status() > 0 && gate.Status() < http.StatusBadRequest) -} - -// dispatchWithAutoFallback wraps a dispatch attempt with protocol -// auto-detection. It tries the preferred target first; on retryable -// failure it falls back to the alternate protocol. Successful protocol -// choices are cached per provider+model, attributed to the service that -// actually served the request — under multi-service failover that may -// differ from the initially selected provider, and caching against the -// initial one would pin a protocol it never confirmed. -func (s *Server) dispatchWithAutoFallback( - c *gin.Context, - provider *typ.Provider, - model string, - preferredTarget protocol.APIType, - dispatch autoDispatchFn, -) { - realWriter := c.Writer - gate := newFirstChunkGate(realWriter) - c.Writer = gate - defer func() { - c.Writer = realWriter - gate.CommitIfBuffered() - }() - - // First attempt with preferred protocol - served, servedModel := dispatch(preferredTarget, gate) - - if gateSucceeded(gate) { - if served != nil { - s.endpointCache.Set(served.UUID, servedModel, preferredTarget) - } - return - } - - // Check if fallback is worthwhile - if !isRetryableStatus(gate.Status()) { - return - } - lastErr := extractLastGinError(c) - if client.IsNonRetryableForProtocolSwitch(lastErr) { - return - } - - // Fallback to alternate protocol - altTarget := alternateOpenAIProtocol(preferredTarget) - logrus.WithContext(c.Request.Context()).Infof( - "[auto-endpoint] %s:%s status=%d → fallback from %s to %s", - provider.UUID, model, gate.Status(), preferredTarget, altTarget, - ) - gate.Discard() - clearGinErrors(c) - - served, servedModel = dispatch(altTarget, gate) - - if gateSucceeded(gate) && served != nil { - s.endpointCache.Set(served.UUID, servedModel, altTarget) - } -} - -// autoDispatchOrFailover decides whether endpoint auto-detection applies to -// this request and dispatches accordingly. With auto-detection, provider -// failover runs nested inside protocol fallback, both sharing one gate -// (owned by dispatchWithAutoFallback). Without it, this is a plain provider -// failover using the statically resolved target (which may be "" — the -// zero value protocol.APIType — meaning attempt must resolve it itself). -// -// attempt is the per-provider-attempt callback; its target argument is the -// protocol to use for that attempt. -func (s *Server) autoDispatchOrFailover( - c *gin.Context, - rule *typ.Rule, - provider *typ.Provider, - actualModel string, - scenarioType typ.RuleScenario, - incoming IncomingAPIType, - attempt func(p *typ.Provider, retryModel string, target protocol.APIType), -) { - var target protocol.APIType - autoFallback := false - if s.autoEndpointEnabled() && provider.APIStyle == protocol.APIStyleOpenAI && ai.IsAutoEndpointMode(provider.OpenAIEndpointMode) { - target, autoFallback = s.resolveAutoTarget(resolveRuleFlags(c, rule), provider, actualModel, scenarioType, incoming) - } - - if autoFallback { - s.dispatchWithAutoFallback(c, provider, actualModel, target, - func(t protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { - return s.dispatchWithPriorityFailoverGated(c, rule, provider, actualModel, - func(p *typ.Provider, retryModel string) { attempt(p, retryModel, t) }, gate) - }) - return - } - - s.dispatchWithPriorityFailover(c, rule, provider, actualModel, - func(p *typ.Provider, retryModel string) { attempt(p, retryModel, target) }) -} diff --git a/internal/server/endpoint_auto_test.go b/internal/server/endpoint_auto_test.go deleted file mode 100644 index ed1841e57..000000000 --- a/internal/server/endpoint_auto_test.go +++ /dev/null @@ -1,353 +0,0 @@ -package server - -import ( - "fmt" - "net/http" - "net/http/httptest" - "testing" - - "github.com/gin-gonic/gin" - "github.com/tingly-dev/tingly-box/internal/protocol" - "github.com/tingly-dev/tingly-box/internal/typ" -) - -func TestAlternateOpenAIProtocol(t *testing.T) { - if got := alternateOpenAIProtocol(protocol.TypeOpenAIChat); got != protocol.TypeOpenAIResponses { - t.Errorf("alternate of chat = %v, want responses", got) - } - if got := alternateOpenAIProtocol(protocol.TypeOpenAIResponses); got != protocol.TypeOpenAIChat { - t.Errorf("alternate of responses = %v, want chat", got) - } -} - -func TestIncomingToTarget(t *testing.T) { - if got := incomingToTarget(IncomingAPIChat); got != protocol.TypeOpenAIChat { - t.Errorf("incoming chat → %v, want chat", got) - } - if got := incomingToTarget(IncomingAPIResponses); got != protocol.TypeOpenAIResponses { - t.Errorf("incoming responses → %v, want responses", got) - } -} - -func TestScenarioPreferredProtocol(t *testing.T) { - tests := []struct { - name string - scenario typ.RuleScenario - incoming IncomingAPIType - want protocol.APIType - }{ - {"codex prefers responses even on chat ingress", typ.ScenarioCodex, IncomingAPIChat, protocol.TypeOpenAIResponses}, - {"codex prefers responses on responses ingress", typ.ScenarioCodex, IncomingAPIResponses, protocol.TypeOpenAIResponses}, - {"codex profile suffix normalized", typ.RuleScenario("codex:p1"), IncomingAPIChat, protocol.TypeOpenAIResponses}, - {"openai mirrors chat ingress", typ.ScenarioOpenAI, IncomingAPIChat, protocol.TypeOpenAIChat}, - {"openai mirrors responses ingress", typ.ScenarioOpenAI, IncomingAPIResponses, protocol.TypeOpenAIResponses}, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - if got := scenarioPreferredProtocol(tt.scenario, tt.incoming); got != tt.want { - t.Errorf("scenarioPreferredProtocol(%q, %q) = %v, want %v", tt.scenario, tt.incoming, got, tt.want) - } - }) - } -} - -// TestDispatchWithAutoFallback_CacheAttributedToServingProvider covers the -// multi-service failover interaction: when the initial provider fails and a -// fallback provider serves the request, the protocol cache entry must be -// written for the serving provider, not the initial one — otherwise the -// initial provider gets pinned to a protocol it never confirmed. -func TestDispatchWithAutoFallback_CacheAttributedToServingProvider(t *testing.T) { - s := &Server{endpointCache: NewEndpointCache(0)} - - initial := &typ.Provider{UUID: "prov-initial"} - serving := &typ.Provider{UUID: "prov-serving"} - - rec := httptest.NewRecorder() - c, _ := gin.CreateTestContext(rec) - c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) - - // Simulate a dispatch where failover moved past the initial provider: - // the gate commits (success) and the serving identity differs from initial. - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { - gate.WriteHeader(200) - gate.CommitFirstChunk() - return serving, "served-model" - } - - s.dispatchWithAutoFallback(c, initial, "req-model", protocol.TypeOpenAIChat, dispatch) - - if _, ok := s.endpointCache.Get(initial.UUID, "req-model"); ok { - t.Error("cache must not contain an entry for the initial provider") - } - got, ok := s.endpointCache.Get(serving.UUID, "served-model") - if !ok { - t.Fatal("cache must contain an entry for the serving provider") - } - if got != protocol.TypeOpenAIChat { - t.Errorf("cached protocol = %v, want chat", got) - } -} - -// TestDispatchWithAutoFallback_NoCacheOnTransformFailure ensures a failed -// transform (served=nil) never writes a cache entry even if the gate state -// looks successful. -func TestDispatchWithAutoFallback_NoCacheOnTransformFailure(t *testing.T) { - s := &Server{endpointCache: NewEndpointCache(0)} - initial := &typ.Provider{UUID: "prov-initial"} - - rec := httptest.NewRecorder() - c, _ := gin.CreateTestContext(rec) - c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) - - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { - gate.WriteHeader(200) // transform error path writes a JSON error; simulate benign status - return nil, "" - } - - s.dispatchWithAutoFallback(c, initial, "m", protocol.TypeOpenAIChat, dispatch) - - if _, ok := s.endpointCache.Get(initial.UUID, "m"); ok { - t.Error("cache must stay empty when the transform failed") - } -} - -// TestDispatchWithAutoFallback_FirstAttemptSucceeds verifies the happy path: -// preferred protocol works on the first try and gets cached. -func TestDispatchWithAutoFallback_FirstAttemptSucceeds(t *testing.T) { - s := &Server{endpointCache: NewEndpointCache(0)} - provider := &typ.Provider{UUID: "prov-1"} - - rec := httptest.NewRecorder() - c, _ := gin.CreateTestContext(rec) - c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) - - calls := 0 - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { - calls++ - gate.WriteHeader(200) - gate.CommitFirstChunk() - return provider, "model-a" - } - - s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIResponses, dispatch) - - if calls != 1 { - t.Errorf("dispatch called %d times, want 1", calls) - } - got, ok := s.endpointCache.Get(provider.UUID, "model-a") - if !ok || got != protocol.TypeOpenAIResponses { - t.Errorf("cache = (%v, %v), want (responses, true)", got, ok) - } -} - -// TestDispatchWithAutoFallback_FallbackSucceeds verifies: preferred fails -// with retryable status → alternate tried → alternate succeeds → alternate -// protocol cached. -func TestDispatchWithAutoFallback_FallbackSucceeds(t *testing.T) { - s := &Server{endpointCache: NewEndpointCache(0)} - provider := &typ.Provider{UUID: "prov-1"} - - rec := httptest.NewRecorder() - c, _ := gin.CreateTestContext(rec) - c.Request = httptest.NewRequest("POST", "/v1/responses", nil) - - var targets []protocol.APIType - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { - targets = append(targets, target) - if target == protocol.TypeOpenAIResponses { - // In practice, upstream 404 is converted to 500 by SendStreamingError - gate.WriteHeader(http.StatusInternalServerError) - gate.Write([]byte(`{"error":"endpoint not found"}`)) - c.Error(fmt.Errorf("status 500: endpoint not found")) - return provider, "model-a" - } - gate.WriteHeader(200) - gate.CommitFirstChunk() - return provider, "model-a" - } - - s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIResponses, dispatch) - - if len(targets) != 2 { - t.Fatalf("dispatch called %d times, want 2", len(targets)) - } - if targets[0] != protocol.TypeOpenAIResponses { - t.Errorf("first attempt target = %v, want responses", targets[0]) - } - if targets[1] != protocol.TypeOpenAIChat { - t.Errorf("fallback target = %v, want chat", targets[1]) - } - got, ok := s.endpointCache.Get(provider.UUID, "model-a") - if !ok || got != protocol.TypeOpenAIChat { - t.Errorf("cache = (%v, %v), want (chat, true)", got, ok) - } -} - -// TestDispatchWithAutoFallback_BothFail verifies: both attempts fail → -// no cache entry written. -func TestDispatchWithAutoFallback_BothFail(t *testing.T) { - s := &Server{endpointCache: NewEndpointCache(0)} - provider := &typ.Provider{UUID: "prov-1"} - - rec := httptest.NewRecorder() - c, _ := gin.CreateTestContext(rec) - c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) - - calls := 0 - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { - calls++ - gate.WriteHeader(http.StatusBadGateway) - gate.Write([]byte(`{"error":"upstream failed"}`)) - c.Error(fmt.Errorf("status 502: bad gateway")) - return provider, "model-a" - } - - s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) - - if calls != 2 { - t.Errorf("dispatch called %d times, want 2 (initial + fallback)", calls) - } - if _, ok := s.endpointCache.Get(provider.UUID, "model-a"); ok { - t.Error("cache must stay empty when both attempts fail") - } -} - -// TestDispatchWithAutoFallback_NonRetryableSkipsFallback verifies: when -// the first attempt fails with a non-retryable error (e.g. 401), no -// fallback is attempted. -func TestDispatchWithAutoFallback_NonRetryableSkipsFallback(t *testing.T) { - s := &Server{endpointCache: NewEndpointCache(0)} - provider := &typ.Provider{UUID: "prov-1"} - - rec := httptest.NewRecorder() - c, _ := gin.CreateTestContext(rec) - c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) - - calls := 0 - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { - calls++ - gate.WriteHeader(http.StatusUnauthorized) - gate.Write([]byte(`{"error":"unauthorized"}`)) - c.Error(fmt.Errorf("status 401: unauthorized")) - return provider, "model-a" - } - - s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) - - if calls != 1 { - t.Errorf("dispatch called %d times, want 1 (no fallback for 401)", calls) - } - if _, ok := s.endpointCache.Get(provider.UUID, "model-a"); ok { - t.Error("cache must stay empty on non-retryable error") - } -} - -// TestDispatchWithAutoFallback_Status0NoRetry verifies: when the writer -// is never touched (status 0), no fallback is attempted — the handler -// ran to completion without producing output. -func TestDispatchWithAutoFallback_Status0NoRetry(t *testing.T) { - s := &Server{endpointCache: NewEndpointCache(0)} - provider := &typ.Provider{UUID: "prov-1"} - - rec := httptest.NewRecorder() - c, _ := gin.CreateTestContext(rec) - c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) - - calls := 0 - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { - calls++ - // Intentionally do nothing — simulate a handler that returns without writing - return provider, "model-a" - } - - s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) - - if calls != 1 { - t.Errorf("dispatch called %d times, want 1 (status 0 is non-retryable)", calls) - } -} - -// TestDispatchWithAutoFallback_GinErrorsClearedBetweenAttempts verifies -// that gin context errors from the first attempt are cleared before the -// fallback, so the fallback starts with a clean error slate. -func TestDispatchWithAutoFallback_GinErrorsClearedBetweenAttempts(t *testing.T) { - s := &Server{endpointCache: NewEndpointCache(0)} - provider := &typ.Provider{UUID: "prov-1"} - - rec := httptest.NewRecorder() - c, _ := gin.CreateTestContext(rec) - c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) - - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { - if target == protocol.TypeOpenAIChat { - gate.WriteHeader(http.StatusInternalServerError) - gate.Write([]byte(`{"error":"endpoint not found"}`)) - c.Error(fmt.Errorf("status 500: endpoint not found")) - return provider, "model-a" - } - // Fallback: verify errors were cleared - if len(c.Errors) != 0 { - t.Errorf("gin errors not cleared before fallback: %v", c.Errors) - } - gate.WriteHeader(200) - gate.CommitFirstChunk() - return provider, "model-a" - } - - s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) -} - -// TestDispatchWithAutoFallback_NonRetryableErrorViaGinContext verifies -// that non-retryable classification uses the gin context error (not just -// status code). A 500 status with a "rate limit" error message should -// NOT trigger fallback. -func TestDispatchWithAutoFallback_NonRetryableErrorViaGinContext(t *testing.T) { - s := &Server{endpointCache: NewEndpointCache(0)} - provider := &typ.Provider{UUID: "prov-1"} - - rec := httptest.NewRecorder() - c, _ := gin.CreateTestContext(rec) - c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) - - calls := 0 - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { - calls++ - gate.WriteHeader(http.StatusInternalServerError) - gate.Write([]byte(`{"error":"rate limit exceeded"}`)) - c.Error(fmt.Errorf("rate limit exceeded")) - return provider, "model-a" - } - - s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) - - if calls != 1 { - t.Errorf("dispatch called %d times, want 1 (rate limit is non-retryable even with 500 status)", calls) - } -} - -// TestDispatchWithAutoFallback_BufferedSuccessNonStreaming verifies the -// non-streaming success path: gate is NOT committed (no CommitFirstChunk) -// but has a 200 status with body → gateSucceeded returns true → cache -// written. -func TestDispatchWithAutoFallback_BufferedSuccessNonStreaming(t *testing.T) { - s := &Server{endpointCache: NewEndpointCache(0)} - provider := &typ.Provider{UUID: "prov-1"} - - rec := httptest.NewRecorder() - c, _ := gin.CreateTestContext(rec) - c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) - - dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { - gate.WriteHeader(200) - gate.Write([]byte(`{"id":"resp-1"}`)) - // No CommitFirstChunk — simulates non-streaming response - return provider, "model-a" - } - - s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) - - got, ok := s.endpointCache.Get(provider.UUID, "model-a") - if !ok || got != protocol.TypeOpenAIChat { - t.Errorf("cache = (%v, %v), want (chat, true) — buffered 200 should count as success", got, ok) - } -} diff --git a/internal/server/endpoint_cache.go b/internal/server/endpoint_cache.go deleted file mode 100644 index c443b6471..000000000 --- a/internal/server/endpoint_cache.go +++ /dev/null @@ -1,62 +0,0 @@ -package server - -import ( - "sync" - "time" - - "github.com/tingly-dev/tingly-box/internal/protocol" -) - -const defaultEndpointCacheTTL = 24 * time.Hour - -type endpointCacheEntry struct { - target protocol.APIType - cachedAt time.Time -} - -type EndpointCache struct { - mu sync.RWMutex - store map[string]endpointCacheEntry - ttl time.Duration -} - -func NewEndpointCache(ttl time.Duration) *EndpointCache { - if ttl <= 0 { - ttl = defaultEndpointCacheTTL - } - return &EndpointCache{ - store: make(map[string]endpointCacheEntry), - ttl: ttl, - } -} - -func endpointCacheKey(providerUUID, model string) string { - return providerUUID + ":" + model -} - -func (c *EndpointCache) Get(providerUUID, model string) (protocol.APIType, bool) { - key := endpointCacheKey(providerUUID, model) - c.mu.RLock() - entry, ok := c.store[key] - c.mu.RUnlock() - if !ok { - return "", false - } - if time.Since(entry.cachedAt) > c.ttl { - c.mu.Lock() - delete(c.store, key) - c.mu.Unlock() - return "", false - } - return entry.target, true -} - -func (c *EndpointCache) Set(providerUUID, model string, target protocol.APIType) { - key := endpointCacheKey(providerUUID, model) - c.mu.Lock() - c.store[key] = endpointCacheEntry{ - target: target, - cachedAt: time.Now(), - } - c.mu.Unlock() -} diff --git a/internal/server/endpoint_cache_test.go b/internal/server/endpoint_cache_test.go deleted file mode 100644 index 603c83403..000000000 --- a/internal/server/endpoint_cache_test.go +++ /dev/null @@ -1,52 +0,0 @@ -package server - -import ( - "testing" - "time" - - "github.com/tingly-dev/tingly-box/internal/protocol" -) - -func TestEndpointCache_GetSet(t *testing.T) { - c := NewEndpointCache(time.Hour) - - // Miss - if _, ok := c.Get("p1", "m1"); ok { - t.Fatal("expected miss on empty cache") - } - - // Set and hit - c.Set("p1", "m1", protocol.TypeOpenAIChat) - got, ok := c.Get("p1", "m1") - if !ok || got != protocol.TypeOpenAIChat { - t.Fatalf("expected chat, got %v ok=%v", got, ok) - } - - // Different model is independent - if _, ok := c.Get("p1", "m2"); ok { - t.Fatal("expected miss for different model") - } -} - -func TestEndpointCache_TTLExpiry(t *testing.T) { - c := NewEndpointCache(10 * time.Millisecond) - - c.Set("p1", "m1", protocol.TypeOpenAIResponses) - time.Sleep(20 * time.Millisecond) - - if _, ok := c.Get("p1", "m1"); ok { - t.Fatal("expected miss after TTL expiry") - } -} - -func TestEndpointCache_Overwrite(t *testing.T) { - c := NewEndpointCache(time.Hour) - - c.Set("p1", "m1", protocol.TypeOpenAIChat) - c.Set("p1", "m1", protocol.TypeOpenAIResponses) - - got, ok := c.Get("p1", "m1") - if !ok || got != protocol.TypeOpenAIResponses { - t.Fatalf("expected responses after overwrite, got %v", got) - } -} diff --git a/internal/server/protocol_endpoint.go b/internal/server/protocol_endpoint.go index 2d69dc202..ebe804333 100644 --- a/internal/server/protocol_endpoint.go +++ b/internal/server/protocol_endpoint.go @@ -2,9 +2,14 @@ package server import ( "fmt" + "net/http" + "sync" + "time" + "github.com/gin-gonic/gin" "github.com/sirupsen/logrus" "github.com/tingly-dev/tingly-box/ai" + "github.com/tingly-dev/tingly-box/internal/client" "github.com/tingly-dev/tingly-box/internal/protocol" "github.com/tingly-dev/tingly-box/internal/typ" ) @@ -124,3 +129,245 @@ func logModeOverrideIgnored(provider *typ.Provider, requestedOverride string) { } logrus.Warnf("rule openai_endpoint_override=%s ignored: provider %s declares mode=%s", requestedOverride, provider.UUID, mode) } + +// ───────────────────────────────────────────────────────────────────────── +// Endpoint cache — per provider+model auto-detection results +// ───────────────────────────────────────────────────────────────────────── + +const defaultEndpointCacheTTL = 24 * time.Hour + +type endpointCacheEntry struct { + target protocol.APIType + cachedAt time.Time +} + +type EndpointCache struct { + mu sync.RWMutex + store map[string]endpointCacheEntry + ttl time.Duration +} + +func NewEndpointCache(ttl time.Duration) *EndpointCache { + if ttl <= 0 { + ttl = defaultEndpointCacheTTL + } + return &EndpointCache{ + store: make(map[string]endpointCacheEntry), + ttl: ttl, + } +} + +func endpointCacheKey(providerUUID, model string) string { + return providerUUID + ":" + model +} + +func (c *EndpointCache) Get(providerUUID, model string) (protocol.APIType, bool) { + key := endpointCacheKey(providerUUID, model) + c.mu.RLock() + entry, ok := c.store[key] + c.mu.RUnlock() + if !ok { + return "", false + } + if time.Since(entry.cachedAt) > c.ttl { + c.mu.Lock() + delete(c.store, key) + c.mu.Unlock() + return "", false + } + return entry.target, true +} + +func (c *EndpointCache) Set(providerUUID, model string, target protocol.APIType) { + key := endpointCacheKey(providerUUID, model) + c.mu.Lock() + c.store[key] = endpointCacheEntry{ + target: target, + cachedAt: time.Now(), + } + c.mu.Unlock() +} + +// ───────────────────────────────────────────────────────────────────────── +// Runtime auto-detection — protocol fallback + failover orchestration +// ───────────────────────────────────────────────────────────────────────── + +// alternateOpenAIProtocol returns the other OpenAI protocol type. +func alternateOpenAIProtocol(current protocol.APIType) protocol.APIType { + if current == protocol.TypeOpenAIResponses { + return protocol.TypeOpenAIChat + } + return protocol.TypeOpenAIResponses +} + +// incomingToTarget maps IncomingAPIType to protocol.APIType. +func incomingToTarget(incoming IncomingAPIType) protocol.APIType { + if incoming == IncomingAPIResponses { + return protocol.TypeOpenAIResponses + } + return protocol.TypeOpenAIChat +} + +// extractLastGinError returns the most recent error recorded on the gin +// context via c.Error(). Returns nil when no errors exist. +func extractLastGinError(c *gin.Context) error { + errs := c.Errors + if len(errs) == 0 { + return nil + } + return errs[len(errs)-1].Err +} + +// clearGinErrors removes all errors from the gin context so that a +// fallback retry starts with a clean slate. +func clearGinErrors(c *gin.Context) { + c.Errors = c.Errors[:0] +} + +// overrideToTarget converts an EndpointOverride to a protocol.APIType. +func overrideToTarget(ov EndpointOverride) protocol.APIType { + if ov == OverrideResponses { + return protocol.TypeOpenAIResponses + } + return protocol.TypeOpenAIChat +} + +// scenarioPreferredProtocol returns the protocol an auto-mode provider +// should try first on a cache miss. Scenarios whose client ecosystem is +// natively Responses-based (Codex) start with Responses regardless of the +// incoming transport — providers serving such traffic overwhelmingly speak +// Responses, so leading with it saves a wasted first round trip. All other +// scenarios mirror the incoming API. +func scenarioPreferredProtocol(scenario typ.RuleScenario, incoming IncomingAPIType) protocol.APIType { + switch scenario.Base() { + case typ.ScenarioCodex: + return protocol.TypeOpenAIResponses + default: + return incomingToTarget(incoming) + } +} + +// resolveAutoTarget handles the auto-mode target resolution shared by both +// OpenAI Chat and Responses handlers. It checks override → cache → +// scenario-preferred default. Returns the resolved target and whether +// auto-fallback should be enabled. +func (s *Server) resolveAutoTarget( + flags typ.RuleFlags, provider *typ.Provider, model string, scenario typ.RuleScenario, incoming IncomingAPIType, +) (target protocol.APIType, autoFallback bool) { + if ov := ParseEndpointOverride(flags.OpenAIEndpointOverride); ov == OverrideChat || ov == OverrideResponses { + return overrideToTarget(ov), false + } + if cached, ok := s.endpointCache.Get(provider.UUID, model); ok { + return cached, false + } + return scenarioPreferredProtocol(scenario, incoming), true +} + +// autoDispatchFn is the callback for dispatchWithAutoFallback. +// It performs transform + dispatch for a given target protocol, using +// the provided gate. Returns the provider/model that served the final +// attempt (failover may have moved past the initially selected service); +// served is nil when dispatch never got a servable provider (e.g. the +// transform itself failed before failover could run). +type autoDispatchFn func(target protocol.APIType, gate *firstChunkGate) (served *typ.Provider, servedModel string) + +// gateSucceeded reports whether the attempt behind the gate produced a +// success: either the stream committed its first chunk, or a buffered +// non-error status is waiting to flush. +func gateSucceeded(gate *firstChunkGate) bool { + return gate.Committed() || (gate.Status() > 0 && gate.Status() < http.StatusBadRequest) +} + +// dispatchWithAutoFallback wraps a dispatch attempt with protocol +// auto-detection. It tries the preferred target first; on retryable +// failure it falls back to the alternate protocol. Successful protocol +// choices are cached per provider+model, attributed to the service that +// actually served the request — under multi-service failover that may +// differ from the initially selected provider, and caching against the +// initial one would pin a protocol it never confirmed. +func (s *Server) dispatchWithAutoFallback( + c *gin.Context, + provider *typ.Provider, + model string, + preferredTarget protocol.APIType, + dispatch autoDispatchFn, +) { + realWriter := c.Writer + gate := newFirstChunkGate(realWriter) + c.Writer = gate + defer func() { + c.Writer = realWriter + gate.CommitIfBuffered() + }() + + // First attempt with preferred protocol + served, servedModel := dispatch(preferredTarget, gate) + + if gateSucceeded(gate) { + if served != nil { + s.endpointCache.Set(served.UUID, servedModel, preferredTarget) + } + return + } + + // Check if fallback is worthwhile + if !isRetryableStatus(gate.Status()) { + return + } + lastErr := extractLastGinError(c) + if client.IsNonRetryableForProtocolSwitch(lastErr) { + return + } + + // Fallback to alternate protocol + altTarget := alternateOpenAIProtocol(preferredTarget) + logrus.WithContext(c.Request.Context()).Infof( + "[auto-endpoint] %s:%s status=%d → fallback from %s to %s", + provider.UUID, model, gate.Status(), preferredTarget, altTarget, + ) + gate.Discard() + clearGinErrors(c) + + served, servedModel = dispatch(altTarget, gate) + + if gateSucceeded(gate) && served != nil { + s.endpointCache.Set(served.UUID, servedModel, altTarget) + } +} + +// autoDispatchOrFailover decides whether endpoint auto-detection applies to +// this request and dispatches accordingly. With auto-detection, provider +// failover runs nested inside protocol fallback, both sharing one gate +// (owned by dispatchWithAutoFallback). Without it, this is a plain provider +// failover using the statically resolved target (which may be "" — the +// zero value protocol.APIType — meaning attempt must resolve it itself). +// +// attempt is the per-provider-attempt callback; its target argument is the +// protocol to use for that attempt. +func (s *Server) autoDispatchOrFailover( + c *gin.Context, + rule *typ.Rule, + provider *typ.Provider, + actualModel string, + scenarioType typ.RuleScenario, + incoming IncomingAPIType, + attempt func(p *typ.Provider, retryModel string, target protocol.APIType), +) { + var target protocol.APIType + autoFallback := false + if s.autoEndpointEnabled() && provider.APIStyle == protocol.APIStyleOpenAI && ai.IsAutoEndpointMode(provider.OpenAIEndpointMode) { + target, autoFallback = s.resolveAutoTarget(resolveRuleFlags(c, rule), provider, actualModel, scenarioType, incoming) + } + + if autoFallback { + s.dispatchWithAutoFallback(c, provider, actualModel, target, + func(t protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { + return s.dispatchWithPriorityFailoverGated(c, rule, provider, actualModel, + func(p *typ.Provider, retryModel string) { attempt(p, retryModel, t) }, gate) + }) + return + } + + s.dispatchWithPriorityFailover(c, rule, provider, actualModel, + func(p *typ.Provider, retryModel string) { attempt(p, retryModel, target) }) +} diff --git a/internal/server/protocol_endpoint_test.go b/internal/server/protocol_endpoint_test.go index 947554686..65573038b 100644 --- a/internal/server/protocol_endpoint_test.go +++ b/internal/server/protocol_endpoint_test.go @@ -1,8 +1,13 @@ package server import ( + "fmt" + "net/http" + "net/http/httptest" "testing" + "time" + "github.com/gin-gonic/gin" "github.com/tingly-dev/tingly-box/ai" "github.com/tingly-dev/tingly-box/internal/protocol" "github.com/tingly-dev/tingly-box/internal/typ" @@ -180,3 +185,396 @@ func TestIsCodexProvider(t *testing.T) { t.Error("Codex-issuer OAuth provider should be Codex") } } + +// ───────────────────────────────────────────────────────────────────────── +// Endpoint cache +// ───────────────────────────────────────────────────────────────────────── + +func TestEndpointCache_GetSet(t *testing.T) { + c := NewEndpointCache(time.Hour) + + // Miss + if _, ok := c.Get("p1", "m1"); ok { + t.Fatal("expected miss on empty cache") + } + + // Set and hit + c.Set("p1", "m1", protocol.TypeOpenAIChat) + got, ok := c.Get("p1", "m1") + if !ok || got != protocol.TypeOpenAIChat { + t.Fatalf("expected chat, got %v ok=%v", got, ok) + } + + // Different model is independent + if _, ok := c.Get("p1", "m2"); ok { + t.Fatal("expected miss for different model") + } +} + +func TestEndpointCache_TTLExpiry(t *testing.T) { + c := NewEndpointCache(10 * time.Millisecond) + + c.Set("p1", "m1", protocol.TypeOpenAIResponses) + time.Sleep(20 * time.Millisecond) + + if _, ok := c.Get("p1", "m1"); ok { + t.Fatal("expected miss after TTL expiry") + } +} + +func TestEndpointCache_Overwrite(t *testing.T) { + c := NewEndpointCache(time.Hour) + + c.Set("p1", "m1", protocol.TypeOpenAIChat) + c.Set("p1", "m1", protocol.TypeOpenAIResponses) + + got, ok := c.Get("p1", "m1") + if !ok || got != protocol.TypeOpenAIResponses { + t.Fatalf("expected responses after overwrite, got %v", got) + } +} + +// ───────────────────────────────────────────────────────────────────────── +// Runtime auto-detection +// ───────────────────────────────────────────────────────────────────────── + +func TestAlternateOpenAIProtocol(t *testing.T) { + if got := alternateOpenAIProtocol(protocol.TypeOpenAIChat); got != protocol.TypeOpenAIResponses { + t.Errorf("alternate of chat = %v, want responses", got) + } + if got := alternateOpenAIProtocol(protocol.TypeOpenAIResponses); got != protocol.TypeOpenAIChat { + t.Errorf("alternate of responses = %v, want chat", got) + } +} + +func TestIncomingToTarget(t *testing.T) { + if got := incomingToTarget(IncomingAPIChat); got != protocol.TypeOpenAIChat { + t.Errorf("incoming chat → %v, want chat", got) + } + if got := incomingToTarget(IncomingAPIResponses); got != protocol.TypeOpenAIResponses { + t.Errorf("incoming responses → %v, want responses", got) + } +} + +func TestScenarioPreferredProtocol(t *testing.T) { + tests := []struct { + name string + scenario typ.RuleScenario + incoming IncomingAPIType + want protocol.APIType + }{ + {"codex prefers responses even on chat ingress", typ.ScenarioCodex, IncomingAPIChat, protocol.TypeOpenAIResponses}, + {"codex prefers responses on responses ingress", typ.ScenarioCodex, IncomingAPIResponses, protocol.TypeOpenAIResponses}, + {"codex profile suffix normalized", typ.RuleScenario("codex:p1"), IncomingAPIChat, protocol.TypeOpenAIResponses}, + {"openai mirrors chat ingress", typ.ScenarioOpenAI, IncomingAPIChat, protocol.TypeOpenAIChat}, + {"openai mirrors responses ingress", typ.ScenarioOpenAI, IncomingAPIResponses, protocol.TypeOpenAIResponses}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := scenarioPreferredProtocol(tt.scenario, tt.incoming); got != tt.want { + t.Errorf("scenarioPreferredProtocol(%q, %q) = %v, want %v", tt.scenario, tt.incoming, got, tt.want) + } + }) + } +} + +// TestDispatchWithAutoFallback_CacheAttributedToServingProvider covers the +// multi-service failover interaction: when the initial provider fails and a +// fallback provider serves the request, the protocol cache entry must be +// written for the serving provider, not the initial one — otherwise the +// initial provider gets pinned to a protocol it never confirmed. +func TestDispatchWithAutoFallback_CacheAttributedToServingProvider(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + + initial := &typ.Provider{UUID: "prov-initial"} + serving := &typ.Provider{UUID: "prov-serving"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + // Simulate a dispatch where failover moved past the initial provider: + // the gate commits (success) and the serving identity differs from initial. + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { + gate.WriteHeader(200) + gate.CommitFirstChunk() + return serving, "served-model" + } + + s.dispatchWithAutoFallback(c, initial, "req-model", protocol.TypeOpenAIChat, dispatch) + + if _, ok := s.endpointCache.Get(initial.UUID, "req-model"); ok { + t.Error("cache must not contain an entry for the initial provider") + } + got, ok := s.endpointCache.Get(serving.UUID, "served-model") + if !ok { + t.Fatal("cache must contain an entry for the serving provider") + } + if got != protocol.TypeOpenAIChat { + t.Errorf("cached protocol = %v, want chat", got) + } +} + +// TestDispatchWithAutoFallback_NoCacheOnTransformFailure ensures a failed +// transform (served=nil) never writes a cache entry even if the gate state +// looks successful. +func TestDispatchWithAutoFallback_NoCacheOnTransformFailure(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + initial := &typ.Provider{UUID: "prov-initial"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { + gate.WriteHeader(200) // transform error path writes a JSON error; simulate benign status + return nil, "" + } + + s.dispatchWithAutoFallback(c, initial, "m", protocol.TypeOpenAIChat, dispatch) + + if _, ok := s.endpointCache.Get(initial.UUID, "m"); ok { + t.Error("cache must stay empty when the transform failed") + } +} + +// TestDispatchWithAutoFallback_FirstAttemptSucceeds verifies the happy path: +// preferred protocol works on the first try and gets cached. +func TestDispatchWithAutoFallback_FirstAttemptSucceeds(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + calls := 0 + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { + calls++ + gate.WriteHeader(200) + gate.CommitFirstChunk() + return provider, "model-a" + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIResponses, dispatch) + + if calls != 1 { + t.Errorf("dispatch called %d times, want 1", calls) + } + got, ok := s.endpointCache.Get(provider.UUID, "model-a") + if !ok || got != protocol.TypeOpenAIResponses { + t.Errorf("cache = (%v, %v), want (responses, true)", got, ok) + } +} + +// TestDispatchWithAutoFallback_FallbackSucceeds verifies: preferred fails +// with retryable status → alternate tried → alternate succeeds → alternate +// protocol cached. +func TestDispatchWithAutoFallback_FallbackSucceeds(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/responses", nil) + + var targets []protocol.APIType + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { + targets = append(targets, target) + if target == protocol.TypeOpenAIResponses { + // In practice, upstream 404 is converted to 500 by SendStreamingError + gate.WriteHeader(http.StatusInternalServerError) + gate.Write([]byte(`{"error":"endpoint not found"}`)) + c.Error(fmt.Errorf("status 500: endpoint not found")) + return provider, "model-a" + } + gate.WriteHeader(200) + gate.CommitFirstChunk() + return provider, "model-a" + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIResponses, dispatch) + + if len(targets) != 2 { + t.Fatalf("dispatch called %d times, want 2", len(targets)) + } + if targets[0] != protocol.TypeOpenAIResponses { + t.Errorf("first attempt target = %v, want responses", targets[0]) + } + if targets[1] != protocol.TypeOpenAIChat { + t.Errorf("fallback target = %v, want chat", targets[1]) + } + got, ok := s.endpointCache.Get(provider.UUID, "model-a") + if !ok || got != protocol.TypeOpenAIChat { + t.Errorf("cache = (%v, %v), want (chat, true)", got, ok) + } +} + +// TestDispatchWithAutoFallback_BothFail verifies: both attempts fail → +// no cache entry written. +func TestDispatchWithAutoFallback_BothFail(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + calls := 0 + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { + calls++ + gate.WriteHeader(http.StatusBadGateway) + gate.Write([]byte(`{"error":"upstream failed"}`)) + c.Error(fmt.Errorf("status 502: bad gateway")) + return provider, "model-a" + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) + + if calls != 2 { + t.Errorf("dispatch called %d times, want 2 (initial + fallback)", calls) + } + if _, ok := s.endpointCache.Get(provider.UUID, "model-a"); ok { + t.Error("cache must stay empty when both attempts fail") + } +} + +// TestDispatchWithAutoFallback_NonRetryableSkipsFallback verifies: when +// the first attempt fails with a non-retryable error (e.g. 401), no +// fallback is attempted. +func TestDispatchWithAutoFallback_NonRetryableSkipsFallback(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + calls := 0 + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { + calls++ + gate.WriteHeader(http.StatusUnauthorized) + gate.Write([]byte(`{"error":"unauthorized"}`)) + c.Error(fmt.Errorf("status 401: unauthorized")) + return provider, "model-a" + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) + + if calls != 1 { + t.Errorf("dispatch called %d times, want 1 (no fallback for 401)", calls) + } + if _, ok := s.endpointCache.Get(provider.UUID, "model-a"); ok { + t.Error("cache must stay empty on non-retryable error") + } +} + +// TestDispatchWithAutoFallback_Status0NoRetry verifies: when the writer +// is never touched (status 0), no fallback is attempted — the handler +// ran to completion without producing output. +func TestDispatchWithAutoFallback_Status0NoRetry(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + calls := 0 + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { + calls++ + // Intentionally do nothing — simulate a handler that returns without writing + return provider, "model-a" + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) + + if calls != 1 { + t.Errorf("dispatch called %d times, want 1 (status 0 is non-retryable)", calls) + } +} + +// TestDispatchWithAutoFallback_GinErrorsClearedBetweenAttempts verifies +// that gin context errors from the first attempt are cleared before the +// fallback, so the fallback starts with a clean error slate. +func TestDispatchWithAutoFallback_GinErrorsClearedBetweenAttempts(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { + if target == protocol.TypeOpenAIChat { + gate.WriteHeader(http.StatusInternalServerError) + gate.Write([]byte(`{"error":"endpoint not found"}`)) + c.Error(fmt.Errorf("status 500: endpoint not found")) + return provider, "model-a" + } + // Fallback: verify errors were cleared + if len(c.Errors) != 0 { + t.Errorf("gin errors not cleared before fallback: %v", c.Errors) + } + gate.WriteHeader(200) + gate.CommitFirstChunk() + return provider, "model-a" + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) +} + +// TestDispatchWithAutoFallback_NonRetryableErrorViaGinContext verifies +// that non-retryable classification uses the gin context error (not just +// status code). A 500 status with a "rate limit" error message should +// NOT trigger fallback. +func TestDispatchWithAutoFallback_NonRetryableErrorViaGinContext(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + calls := 0 + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { + calls++ + gate.WriteHeader(http.StatusInternalServerError) + gate.Write([]byte(`{"error":"rate limit exceeded"}`)) + c.Error(fmt.Errorf("rate limit exceeded")) + return provider, "model-a" + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) + + if calls != 1 { + t.Errorf("dispatch called %d times, want 1 (rate limit is non-retryable even with 500 status)", calls) + } +} + +// TestDispatchWithAutoFallback_BufferedSuccessNonStreaming verifies the +// non-streaming success path: gate is NOT committed (no CommitFirstChunk) +// but has a 200 status with body → gateSucceeded returns true → cache +// written. +func TestDispatchWithAutoFallback_BufferedSuccessNonStreaming(t *testing.T) { + s := &Server{endpointCache: NewEndpointCache(0)} + provider := &typ.Provider{UUID: "prov-1"} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest("POST", "/v1/chat/completions", nil) + + dispatch := func(target protocol.APIType, gate *firstChunkGate) (*typ.Provider, string) { + gate.WriteHeader(200) + gate.Write([]byte(`{"id":"resp-1"}`)) + // No CommitFirstChunk — simulates non-streaming response + return provider, "model-a" + } + + s.dispatchWithAutoFallback(c, provider, "model-a", protocol.TypeOpenAIChat, dispatch) + + got, ok := s.endpointCache.Get(provider.UUID, "model-a") + if !ok || got != protocol.TypeOpenAIChat { + t.Errorf("cache = (%v, %v), want (chat, true) — buffered 200 should count as success", got, ok) + } +}