From 54352f3da7c791efa6d5a84820784d824f7ff627 Mon Sep 17 00:00:00 2001 From: Luxferre Date: Sat, 5 Sep 2026 16:39:35 +0300 Subject: [PATCH] feat: add non-ZeroGPU endpoints, automatic failover, and endpoint cooldown --- README.md | 44 +++++---- main.go | 260 ++++++++++++++++++++++++++++++++++++++++++------- qflash_test.go | 186 ++++++++++++++++++++++++++++++++++- 3 files changed, 433 insertions(+), 57 deletions(-) diff --git a/README.md b/README.md index 2f20ccc..38cbbc0 100644 --- a/README.md +++ b/README.md @@ -4,15 +4,17 @@ Qflash is a standalone, single-binary gateway that exposes Qwen3.8 model spaces on Hugging Face through a standard OpenAI-compatible API. It automatically detects and supports multiple upstream protocols: -- Gradio `/respond` endpoints (such as `https://microhero-qwen3-8-27b-uncensored-chat.hf.space`, running live on ZeroGPU) -- Direct OpenAI `/v1/chat/completions` endpoints (such as `https://apathy-exe-qwen3-8-flash-next.hf.space`, running `llama-server`) -- Gradio `/chat_response` endpoints (such as `https://halvo78-qwen3-8-flash-next-playground.hf.space`) +- gradio `/respond` endpoints (such as `https://microhero-qwen3-8-27b-uncensored-chat.hf.space`, running live on ZeroGPU) +- direct OpenAI `/v1/chat/completions` endpoints (such as `https://wanyamaelis-qwen3-8-27b.hf.space` and `https://apathy-exe-qwen3-8-flash-next.hf.space`, running `llama-server` on CPU) +- gradio `/chat_response` endpoints (such as `https://halvo78-qwen3-8-flash-next-playground.hf.space`) It translates standard `/v1/chat/completions` and `/v1/models` requests and Server-Sent Events (SSE) stream protocols into upstream formats, allowing any standard OpenAI-compatible client, agent, or IDE to interface with Qwen3.8 models without modification. ## Features - openai-compatible chat completions (streaming and non-streaming) +- automatic upstream failover across ZeroGPU and non-ZeroGPU endpoints +- smart endpoint cooldown (5 minutes on quota exhaustion, 30 seconds on network errors) - automatic upstream endpoint detection (`respond`, `openai`, `chat_response`) - deep reasoning extraction with thinking trace passthrough (`` tags and blockquotes mapped to `reasoning_content`) - stateful streaming tool call interception (`StreamToolCallFilter`) with zero XML or JSON leakage into `delta.content` @@ -28,20 +30,21 @@ It translates standard `/v1/chat/completions` and `/v1/models` requests and Serv | Space | Model | Hardware | Protocol | Notes | |---|---|---|---|---| -| `MicroHERO/qwen3.8-27b-uncensored-chat` *(default)* | Qwen3.8-27B Uncensored | ZeroGPU (A10G) | Gradio `/respond` | Fast live inference (~2-5s), vision projector, uncensored build | -| `apathy-exe/Qwen3.8-Flash-Next` | Qwen3.8-Flash-Next (~177B) | CPU (OpenMP/AVX512) | Native OpenAI `/v1` | Full 177B model, single execution slot | -| `apathy-exe/Qwen3.8-27B` | Qwen3.8-27B (MTP) | CPU (OpenMP/AVX512) | Native OpenAI `/v1` | 27B model with speculative decoding, single execution slot | +| `MicroHERO/qwen3.8-27b-uncensored-chat` *(primary)* | Qwen3.8-27B Uncensored | ZeroGPU (A10G) | Gradio `/respond` | Fast live GPU inference (~2s), uncensored weights | +| `Wanyamaelis/Qwen3.8-27B` *(fallback)* | Qwen3.8-27B (MTP) | CPU basic (8 vCPU) | Native OpenAI `/v1` | Non-ZeroGPU, no quota limits, fast speculative decoding (~5s) | +| `apathy-exe/Qwen3.8-27B` | Qwen3.8-27B (MTP) | CPU (OpenMP/AVX512) | Native OpenAI `/v1` | Non-ZeroGPU, no quota limits, speculative decoding | +| `apathy-exe/Qwen3.8-Flash-Next` | Qwen3.8-Flash-Next (~177B) | CPU (OpenMP/AVX512) | Native OpenAI `/v1` | Non-ZeroGPU, no quota limits, full 177B Flash-Next model | | `Halvo78/qwen3-8-flash-next-playground` | Qwen3.8-Flash-Next | CPU basic | Gradio `/chat_response` | Sandbox client; requires BYOK API key/base URL | -### Hugging Face ZeroGPU quota note +### Auto-failover and ZeroGPU quota handling -Spaces running on ZeroGPU provide free compute. Anonymous requests share a small pool per IP address. To avoid quota limits, set a free Hugging Face token: +Spaces running on ZeroGPU provide free A10G compute, but anonymous requests share a small pool per IP address. When ZeroGPU runs limit is reached, upstream returns a quota error (`429` or `ZeroGPU runs limit`). -```bash -export HF_TOKEN="hf_..." -``` +Qflash handles this seamlessly: -Or pass it via the `-hf-token` flag, or via the `Authorization: Bearer hf_...` header in your client requests. Personal access tokens can be created for free at `https://huggingface.co/settings/tokens`. +- with auto-failover enabled (default), when a ZeroGPU space hits its runs limit, it is placed on a 5-minute cooldown and the gateway automatically fails over to the next configured endpoint (e.g. `wanyamaelis-qwen3-8-27b`, which runs on CPU with zero quota limits) +- transient network errors trigger a shorter 30-second cooldown before retrying +- you can also provide a free Hugging Face personal access token (`https://huggingface.co/settings/tokens`) via the `HF_TOKEN` environment variable, the `-hf-token` CLI flag, or the `Authorization: Bearer hf_...` header to authenticate ZeroGPU requests directly ## Installation @@ -77,7 +80,7 @@ The gateway advertises the following models under `/v1/models`: | Model ID | Target model | Description | |---|---|---| -| `Qwen/Qwen3.8-27B-Uncensored` | `Qwen/Qwen3.8-27B-Uncensored` | Default primary live model (ZeroGPU) | +| `Qwen/Qwen3.8-27B-Uncensored` | `Qwen/Qwen3.8-27B-Uncensored` | Default primary live model | | `Qwen/Qwen3.8-Flash-Next` | `Qwen/Qwen3.8-Flash-Next` | Flash-Next model identifier | | `qwen3.8-27b-uncensored` | `Qwen/Qwen3.8-27B-Uncensored` | Standard lowercase alias | | `qwen3.8-flash-next` | `Qwen/Qwen3.8-Flash-Next` | Flash-Next lowercase alias | @@ -89,18 +92,19 @@ Any unlisted custom model name requested by the client is passed through directl ## Usage -Run the gateway: +Run the gateway with default auto-failover endpoints: ```bash qflash ``` -By default, this listens on `http://127.0.0.1:8080` and connects to `https://microhero-qwen3-8-27b-uncensored-chat.hf.space`. +By default, this listens on `http://127.0.0.1:8080` with failover configured across `MicroHERO` (ZeroGPU), `wanyamaelis` (CPU non-ZeroGPU), and `apathy-exe` endpoints. Available flags: -- `-port` — TCP port to listen on (default `8080`) -- `-endpoint` — root URL of upstream space (default `https://microhero-qwen3-8-27b-uncensored-chat.hf.space`, `QFLASH_ENDPOINT` env) +- `-port` — tcp port to listen on (default `8080`) +- `-endpoints` / `-endpoint` — comma-separated upstream space or OpenAI URLs (default list of 4 endpoints, `QFLASH_ENDPOINTS` / `QFLASH_ENDPOINT` env) +- `-failover` / `-auto-failover` — enable automatic failover across endpoints on quota exhaustion or error (default `true`, `QFLASH_FAILOVER` env) - `-mode` — upstream protocol mode: `auto`, `respond`, `chat_response`, `openai` (default `auto`, `QFLASH_MODE` env) - `-model` — exposed model name (default `Qwen/Qwen3.8-27B-Uncensored`, `QFLASH_MODEL` env) - `-thinking` / `-enable-thinking` — enable chain-of-thought reasoning by default (default `true`) @@ -139,11 +143,11 @@ curl http://localhost:8080/v1/chat/completions \ -d '{"model":"qwen","messages":[{"role":"user","content":"Hello!"}],"reasoning_effort":"none"}' ``` -Using an alternative upstream space: +Using custom upstream spaces or a single endpoint: ```bash -# Point to apathy-exe CPU llama-server -qflash -endpoint https://apathy-exe-qwen3-8-flash-next.hf.space -mode openai +# Point to a single CPU llama-server without failover +qflash -endpoint https://wanyamaelis-qwen3-8-27b.hf.space -failover=false # Point to Halvo78 playground with your own API credentials qflash -endpoint https://halvo78-qwen3-8-flash-next-playground.hf.space -api-key "sk-..." -base-url "https://api.openai.com/v1" diff --git a/main.go b/main.go index 196c0bd..c4cfa6a 100644 --- a/main.go +++ b/main.go @@ -15,9 +15,11 @@ import ( "log" "net" "net/http" + "net/http/httptest" "os" "strconv" "strings" + "sync" "time" ) @@ -29,6 +31,12 @@ var ( ConfiguredBaseURL string ConfiguredModel string EnableThinkingDefault = true + DefaultEndpoints = []string{ + "https://microhero-qwen3-8-27b-uncensored-chat.hf.space", + "https://wanyamaelis-qwen3-8-27b.hf.space", + "https://apathy-exe-qwen3-8-27b.hf.space", + "https://apathy-exe-qwen3-8-flash-next.hf.space", + } ) // --------------------------------------------------------------------------- @@ -1195,19 +1203,52 @@ func parseAssistantText(dataJSON string) (string, bool) { return "", false } +type EndpointNode struct { + URL string + Mode string + CooldownUntil time.Time + FailureCount int +} + type QwenService struct { - endpoint string + endpoints []*EndpointNode modelName string mode string token string apiKey string baseURL string enableThinking bool + autoFailover bool + mu sync.Mutex client *http.Client } -func NewQwenService(endpoint, modelName, mode, token, apiKey, baseURL, socksProxy string, enableThinking bool) *QwenService { - cleanEndpoint := strings.TrimRight(endpoint, "/") +func parseEndpointList(rawList []string) []*EndpointNode { + var nodes []*EndpointNode + seen := make(map[string]bool) + for _, item := range rawList { + parts := strings.Split(item, ",") + for _, p := range parts { + clean := strings.TrimRight(strings.TrimSpace(p), "/") + if clean != "" && !seen[clean] { + seen[clean] = true + nodes = append(nodes, &EndpointNode{ + URL: clean, + }) + } + } + } + if len(nodes) == 0 { + for _, ep := range DefaultEndpoints { + nodes = append(nodes, &EndpointNode{ + URL: ep, + }) + } + } + return nodes +} + +func NewQwenService(endpoints []string, modelName, mode, token, apiKey, baseURL, socksProxy string, enableThinking, autoFailover bool) *QwenService { if modelName == "" { modelName = "Qwen/Qwen3.8-27B-Uncensored" } @@ -1227,14 +1268,17 @@ func NewQwenService(endpoint, modelName, mode, token, apiKey, baseURL, socksProx } } + nodes := parseEndpointList(endpoints) + return &QwenService{ - endpoint: cleanEndpoint, + endpoints: nodes, modelName: modelName, mode: mode, token: token, apiKey: apiKey, baseURL: baseURL, enableThinking: enableThinking, + autoFailover: autoFailover, client: &http.Client{Transport: transport, Timeout: 300 * time.Second}, } } @@ -1274,24 +1318,24 @@ func (s *QwenService) ListModels() []ModelItem { return models } -func (s *QwenService) detectEndpointMode() string { - if s.mode != "" && s.mode != "auto" { - return s.mode +func (s *QwenService) detectEndpointModeFor(epURL, defaultMode string) string { + if defaultMode != "" && defaultMode != "auto" { + return defaultMode } - ep := strings.ToLower(s.endpoint) + ep := strings.ToLower(epURL) if strings.Contains(ep, "microhero") || strings.HasSuffix(ep, "/respond") { return "respond" } if strings.Contains(ep, "halvo78") || strings.HasSuffix(ep, "/chat_response") { return "chat_response" } - if strings.Contains(ep, "apathy-exe") || strings.HasSuffix(ep, "/v1") { + if strings.Contains(ep, "wanyamaelis") || strings.Contains(ep, "apathy-exe") || strings.HasSuffix(ep, "/v1") { return "openai" } // Probe /gradio_api/info - infoURL := s.endpoint + "/gradio_api/info" + infoURL := epURL + "/gradio_api/info" req, err := http.NewRequest("GET", infoURL, nil) if err == nil { req.Header.Set("User-Agent", DefaultUserAgent) @@ -1316,28 +1360,144 @@ func (s *QwenService) detectEndpointMode() string { } } - return "respond" + return "openai" +} + +func (s *QwenService) getEligibleEndpoints() []*EndpointNode { + s.mu.Lock() + defer s.mu.Unlock() + + now := time.Now() + var ready []*EndpointNode + var cooling []*EndpointNode + + for _, node := range s.endpoints { + if now.After(node.CooldownUntil) { + ready = append(ready, node) + } else { + cooling = append(cooling, node) + } + } + + if len(ready) > 0 { + return ready + } + + // If all are cooling down, return all so we still attempt + return cooling +} + +func (s *QwenService) markEndpointFailure(node *EndpointNode, err error) { + s.mu.Lock() + defer s.mu.Unlock() + + node.FailureCount++ + errStr := strings.ToLower(err.Error()) + // If it's a quota or rate-limit error, cool down for 5 minutes + if strings.Contains(errStr, "zerogpu") || strings.Contains(errStr, "quota") || strings.Contains(errStr, "429") { + node.CooldownUntil = time.Now().Add(5 * time.Minute) + log.Printf("Endpoint %s hit quota/rate limit, cooling down until %s", node.URL, node.CooldownUntil.Format("15:04:05")) + } else { + // For other transient errors, cool down for 30 seconds + node.CooldownUntil = time.Now().Add(30 * time.Second) + log.Printf("Endpoint %s failed (%v), cooling down for 30s", node.URL, err) + } +} + +func (s *QwenService) markEndpointSuccess(node *EndpointNode) { + s.mu.Lock() + defer s.mu.Unlock() + node.FailureCount = 0 + node.CooldownUntil = time.Time{} } func (s *QwenService) Chat(w http.ResponseWriter, r *http.Request, req ChatCompletionRequest) error { resolvedModel := EffectiveModelID(req.Model, s.modelName) maxTokens := ResolveMaxTokens(req) - mode := s.detectEndpointMode() - switch mode { - case "openai": - return s.chatDirectOpenAI(w, r, req, resolvedModel, maxTokens) - case "chat_response": - return s.chatGradioChatResponse(w, r, req, resolvedModel, maxTokens) - case "respond": - fallthrough - default: - return s.chatGradioRespond(w, r, req, resolvedModel, maxTokens) + endpoints := s.getEligibleEndpoints() + if len(endpoints) == 0 { + return fmt.Errorf("no upstream endpoints configured") } + + var lastErr error + for i, node := range endpoints { + mode := node.Mode + if mode == "" || mode == "auto" { + mode = s.detectEndpointModeFor(node.URL, s.mode) + node.Mode = mode + } + + if !req.Stream { + rec := httptest.NewRecorder() + var err error + switch mode { + case "openai": + err = s.chatDirectOpenAI(node.URL, rec, r, req, resolvedModel, maxTokens) + case "chat_response": + err = s.chatGradioChatResponse(node.URL, rec, r, req, resolvedModel, maxTokens) + case "respond": + fallthrough + default: + err = s.chatGradioRespond(node.URL, rec, r, req, resolvedModel, maxTokens) + } + + if err == nil && rec.Code == http.StatusOK { + s.markEndpointSuccess(node) + for k, vv := range rec.Header() { + for _, v := range vv { + w.Header().Add(k, v) + } + } + w.WriteHeader(rec.Code) + w.Write(rec.Body.Bytes()) + return nil + } + + if err == nil && rec.Code != http.StatusOK { + err = fmt.Errorf("HTTP %d: %s", rec.Code, rec.Body.String()) + } + + lastErr = err + s.markEndpointFailure(node, err) + if !s.autoFailover || i == len(endpoints)-1 { + break + } + log.Printf("Endpoint %s failed (%v), failing over to next endpoint...", node.URL, err) + continue + } + + // Streaming request + var err error + switch mode { + case "openai": + err = s.chatDirectOpenAI(node.URL, w, r, req, resolvedModel, maxTokens) + case "chat_response": + err = s.chatGradioChatResponse(node.URL, w, r, req, resolvedModel, maxTokens) + case "respond": + fallthrough + default: + err = s.chatGradioRespond(node.URL, w, r, req, resolvedModel, maxTokens) + } + + if err == nil { + s.markEndpointSuccess(node) + return nil + } + + lastErr = err + s.markEndpointFailure(node, err) + if !s.autoFailover || i == len(endpoints)-1 { + break + } + log.Printf("Endpoint %s streaming failed (%v), failing over to next endpoint...", node.URL, err) + } + + return fmt.Errorf("all endpoints failed (last error: %v)", lastErr) } -func (s *QwenService) chatDirectOpenAI(w http.ResponseWriter, r *http.Request, req ChatCompletionRequest, resolvedModel string, maxTokens int) error { - targetURL := s.endpoint +func (s *QwenService) chatDirectOpenAI(endpointURL string, w http.ResponseWriter, r *http.Request, req ChatCompletionRequest, resolvedModel string, maxTokens int) error { + targetURL := endpointURL if !strings.HasSuffix(targetURL, "/v1/chat/completions") && !strings.HasSuffix(targetURL, "/chat/completions") { if strings.HasSuffix(targetURL, "/v1") { targetURL += "/chat/completions" @@ -1375,6 +1535,11 @@ func (s *QwenService) chatDirectOpenAI(w http.ResponseWriter, r *http.Request, r } defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + bodyBytes, _ := io.ReadAll(resp.Body) + return fmt.Errorf("upstream HTTP %d: %s", resp.StatusCode, string(bodyBytes)) + } + for k, vv := range resp.Header { for _, v := range vv { w.Header().Add(k, v) @@ -1404,7 +1569,7 @@ func (s *QwenService) chatDirectOpenAI(w http.ResponseWriter, r *http.Request, r return err } -func (s *QwenService) chatGradioRespond(w http.ResponseWriter, r *http.Request, req ChatCompletionRequest, resolvedModel string, maxTokens int) error { +func (s *QwenService) chatGradioRespond(endpointURL string, w http.ResponseWriter, r *http.Request, req ChatCompletionRequest, resolvedModel string, maxTokens int) error { var promptText string toolsPrompt := FormatToolsPrompt(req.Tools) @@ -1496,7 +1661,7 @@ func (s *QwenService) chatGradioRespond(w http.ResponseWriter, r *http.Request, return fmt.Errorf("failed to encode request: %w", err) } - callURL := s.endpoint + "/gradio_api/call/respond" + callURL := endpointURL + "/gradio_api/call/respond" makeCallReq := func() (*http.Request, error) { reqObj, err := http.NewRequest("POST", callURL, bytes.NewBuffer(jsonPayload)) if err != nil { @@ -1521,7 +1686,7 @@ func (s *QwenService) chatGradioRespond(w http.ResponseWriter, r *http.Request, return fmt.Errorf("failed to parse Gradio event ID") } - streamURL := fmt.Sprintf("%s/gradio_api/call/respond/%s", s.endpoint, joinRes.EventID) + streamURL := fmt.Sprintf("%s/gradio_api/call/respond/%s", endpointURL, joinRes.EventID) makeStreamReq := func() (*http.Request, error) { reqObj, err := http.NewRequest("GET", streamURL, nil) if err != nil { @@ -1624,7 +1789,7 @@ func (s *QwenService) chatGradioRespond(w http.ResponseWriter, r *http.Request, return nil } -func (s *QwenService) chatGradioChatResponse(w http.ResponseWriter, r *http.Request, req ChatCompletionRequest, resolvedModel string, maxTokens int) error { +func (s *QwenService) chatGradioChatResponse(endpointURL string, w http.ResponseWriter, r *http.Request, req ChatCompletionRequest, resolvedModel string, maxTokens int) error { var systemPromptStr string var historyArray []map[string]interface{} var messageStr string @@ -1791,7 +1956,7 @@ func (s *QwenService) chatGradioChatResponse(w http.ResponseWriter, r *http.Requ return fmt.Errorf("failed to encode request: %w", err) } - callURL := s.endpoint + "/gradio_api/call/chat_response" + callURL := endpointURL + "/gradio_api/call/chat_response" makeCallReq := func() (*http.Request, error) { reqObj, err := http.NewRequest("POST", callURL, bytes.NewBuffer(jsonPayload)) if err != nil { @@ -1816,7 +1981,7 @@ func (s *QwenService) chatGradioChatResponse(w http.ResponseWriter, r *http.Requ return fmt.Errorf("failed to parse Gradio event ID") } - streamURL := fmt.Sprintf("%s/gradio_api/call/chat_response/%s", s.endpoint, joinRes.EventID) + streamURL := fmt.Sprintf("%s/gradio_api/call/chat_response/%s", endpointURL, joinRes.EventID) makeStreamReq := func() (*http.Request, error) { reqObj, err := http.NewRequest("GET", streamURL, nil) if err != nil { @@ -1977,11 +2142,18 @@ func (s *QwenService) chatGradioChatResponse(w http.ResponseWriter, r *http.Requ func main() { port := flag.Int("port", 8080, "Port to listen on") - defaultEndpoint := "https://microhero-qwen3-8-27b-uncensored-chat.hf.space" - if envEP := os.Getenv("QFLASH_ENDPOINT"); envEP != "" { - defaultEndpoint = envEP + defaultEndpointsStr := strings.Join(DefaultEndpoints, ",") + if envEP := os.Getenv("QFLASH_ENDPOINTS"); envEP != "" { + defaultEndpointsStr = envEP + } else if envEP := os.Getenv("QFLASH_ENDPOINT"); envEP != "" { + defaultEndpointsStr = envEP } - endpoint := flag.String("endpoint", defaultEndpoint, "Upstream Hugging Face Space or OpenAI URL") + endpointsFlag := flag.String("endpoints", defaultEndpointsStr, "Comma-separated upstream Hugging Face Spaces or OpenAI URLs") + flag.StringVar(endpointsFlag, "endpoint", defaultEndpointsStr, "Alias for -endpoints") + + autoFailover := flag.Bool("failover", true, "Enable automatic failover across endpoints on error or quota limit") + flag.BoolVar(autoFailover, "auto-failover", true, "Alias for -failover") + defaultModelVal := "Qwen/Qwen3.8-27B-Uncensored" if envModel := os.Getenv("QFLASH_MODEL"); envModel != "" { defaultModelVal = envModel @@ -2005,6 +2177,11 @@ func main() { if envMode := os.Getenv("QFLASH_MODE"); envMode != "" && *mode == "auto" { *mode = envMode } + if envFailover := os.Getenv("QFLASH_FAILOVER"); envFailover != "" { + if v, err := strconv.ParseBool(envFailover); err == nil { + *autoFailover = v + } + } if *userAgent != "" { ConfiguredUserAgent = *userAgent @@ -2031,7 +2208,17 @@ func main() { } } - svc := NewQwenService(*endpoint, *defaultModel, *mode, *hfToken, *apiKey, *baseURL, proxyURL, *thinking) + var rawEndpoints []string + if *endpointsFlag != "" { + for _, ep := range strings.Split(*endpointsFlag, ",") { + ep = strings.TrimSpace(ep) + if ep != "" { + rawEndpoints = append(rawEndpoints, ep) + } + } + } + + svc := NewQwenService(rawEndpoints, *defaultModel, *mode, *hfToken, *apiKey, *baseURL, proxyURL, *thinking, *autoFailover) mux := http.NewServeMux() @@ -2098,7 +2285,10 @@ func main() { }) addr := fmt.Sprintf(":%d", *port) - log.Printf("Starting qflash gateway on %s -> %s", addr, *endpoint) + log.Printf("Starting qflash gateway on %s with %d endpoints (auto-failover: %v)", addr, len(svc.endpoints), *autoFailover) + for _, ep := range svc.endpoints { + log.Printf(" - Upstream endpoint: %s", ep.URL) + } if proxyURL != "" { log.Printf("Routing through SOCKS5 proxy: %s", proxyURL) } diff --git a/qflash_test.go b/qflash_test.go index 7f93708..ec5c340 100644 --- a/qflash_test.go +++ b/qflash_test.go @@ -232,7 +232,7 @@ func TestQwenServiceChatMock(t *testing.T) { })) defer mockServer.Close() - svc := NewQwenService(mockServer.URL, "Qwen/Qwen3.8-Flash-Next", "chat_response", "", "", "", "", true) + svc := NewQwenService([]string{mockServer.URL}, "Qwen/Qwen3.8-Flash-Next", "chat_response", "", "", "", "", true, true) // 1. Test Non-streaming completion rec := httptest.NewRecorder() @@ -362,7 +362,7 @@ func TestQwenServiceChatRespondMock(t *testing.T) { })) defer mockServer.Close() - svc := NewQwenService(mockServer.URL, "Qwen/Qwen3.8-27B-Uncensored", "respond", "", "", "", "", true) + svc := NewQwenService([]string{mockServer.URL}, "Qwen/Qwen3.8-27B-Uncensored", "respond", "", "", "", "", true, true) // Non-streaming test rec := httptest.NewRecorder() @@ -445,3 +445,185 @@ func TestQwenServiceChatRespondMock(t *testing.T) { } } +func TestQwenServiceDirectOpenAIMock(t *testing.T) { + mockServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/chat/completions" && r.Method == http.MethodPost { + var req ChatCompletionRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + if !req.Stream { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(ChatCompletionResponse{ + ID: "chatcmpl-openai-mock", + Object: "chat.completion", + Created: 1234567890, + Model: req.Model, + Choices: []ChatCompletionResponseChoice{ + { + Index: 0, + Message: ChatMessage{ + Role: "assistant", + Content: "Hello from native OpenAI upstream!", + }, + FinishReason: "stop", + }, + }, + }) + return + } + + // Stream + w.Header().Set("Content-Type", "text/event-stream") + flusher, _ := w.(http.Flusher) + chunk := `data: {"id":"chatcmpl-openai-mock","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"Hello from "}}]} + +data: {"id":"chatcmpl-openai-mock","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"native stream!"}}]} + +data: [DONE] + +` + w.Write([]byte(chunk)) + if flusher != nil { + flusher.Flush() + } + return + } + + http.NotFound(w, r) + })) + defer mockServer.Close() + + svc := NewQwenService([]string{mockServer.URL}, "Qwen/Qwen3.8-27B-Uncensored", "openai", "", "", "", "", true, true) + + // Non-streaming + rec := httptest.NewRecorder() + req := ChatCompletionRequest{ + Model: "qwen-openai", + Messages: []ChatMessage{ + {Role: "user", Content: "Hello"}, + }, + Stream: false, + } + if err := svc.Chat(rec, nil, req); err != nil { + t.Fatalf("unexpected error in direct openai non-streaming: %v", err) + } + if rec.Code != http.StatusOK { + t.Fatalf("expected HTTP 200, got %d", rec.Code) + } + var resp ChatCompletionResponse + if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil { + t.Fatalf("failed to decode response: %v", err) + } + if resp.Choices[0].Message.Content != "Hello from native OpenAI upstream!" { + t.Fatalf("unexpected content: %v", resp.Choices[0].Message.Content) + } + + // Streaming + recStream := httptest.NewRecorder() + reqStream := ChatCompletionRequest{ + Model: "qwen-openai", + Messages: []ChatMessage{ + {Role: "user", Content: "Hello"}, + }, + Stream: true, + } + if err := svc.Chat(recStream, nil, reqStream); err != nil { + t.Fatalf("unexpected error in direct openai streaming: %v", err) + } + if !strings.Contains(recStream.Body.String(), "native stream!") { + t.Fatalf("expected streamed content, got %q", recStream.Body.String()) + } +} + +func TestQwenServiceAutoFailover(t *testing.T) { + // Server 1 simulates ZeroGPU quota exceeded (returns HTTP 429) + server1Hits := 0 + server1 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + server1Hits++ + w.WriteHeader(http.StatusTooManyRequests) + w.Write([]byte(`{"error": "ZeroGPU runs limit reached: quota exceeded"}`)) + })) + defer server1.Close() + + // Server 2 is healthy and serves completions + server2Hits := 0 + server2 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + server2Hits++ + if r.URL.Path == "/v1/chat/completions" { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(ChatCompletionResponse{ + ID: "chatcmpl-failover-mock", + Object: "chat.completion", + Created: 1234567890, + Choices: []ChatCompletionResponseChoice{ + { + Index: 0, + Message: ChatMessage{ + Role: "assistant", + Content: "Response from fallback server!", + }, + FinishReason: "stop", + }, + }, + }) + return + } + http.NotFound(w, r) + })) + defer server2.Close() + + endpoints := []string{server1.URL, server2.URL} + svc := NewQwenService(endpoints, "Qwen/Qwen3.8-27B", "openai", "", "", "", "", true, true) + + rec := httptest.NewRecorder() + req := ChatCompletionRequest{ + Model: "qwen3.8-27b", + Messages: []ChatMessage{ + {Role: "user", Content: "Failover test"}, + }, + Stream: false, + } + + err := svc.Chat(rec, nil, req) + if err != nil { + t.Fatalf("expected failover to succeed, got error: %v", err) + } + + if rec.Code != http.StatusOK { + t.Fatalf("expected HTTP 200 after failover, got %d", rec.Code) + } + + var resp ChatCompletionResponse + if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil { + t.Fatalf("failed to decode failover response: %v", err) + } + + if resp.Choices[0].Message.Content != "Response from fallback server!" { + t.Fatalf("unexpected content: %v", resp.Choices[0].Message.Content) + } + + if server1Hits != 1 { + t.Fatalf("expected server1 to be attempted once, got %d", server1Hits) + } + if server2Hits != 1 { + t.Fatalf("expected server2 to be called once on failover, got %d", server2Hits) + } + + // Second request should skip server 1 because it entered cooldown + rec2 := httptest.NewRecorder() + err2 := svc.Chat(rec2, nil, req) + if err2 != nil { + t.Fatalf("expected second request to succeed directly on server2: %v", err2) + } + if server1Hits != 1 { + t.Fatalf("expected server1 to be skipped during cooldown, but was hit %d times", server1Hits) + } + if server2Hits != 2 { + t.Fatalf("expected server2 to receive the second request, got hits: %d", server2Hits) + } +} + +