commit d06c025c249f4711c61f77722e394940fba5c84f Author: Luxferre Date: Sat Sep 5 15:42:07 2026 +0300 feat: initial implementation of qflash gateway diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ab52275 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +bin/ +*.exe diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..663ac5b --- /dev/null +++ b/Makefile @@ -0,0 +1,18 @@ +# Makefile for qflash (Qwen3.8-Flash-Next OpenAI Proxy Gateway) +# Created by Luxferre in 2026, released into the public domain. + +qflash: + go build -trimpath -ldflags="-s -w" -o bin/qflash . + +all: qflash + +test: + go test -v ./... + +clean: + rm -rf bin/ + +run: qflash + ./bin/qflash + +.PHONY: all clean qflash run test diff --git a/README.md b/README.md new file mode 100644 index 0000000..f934c81 --- /dev/null +++ b/README.md @@ -0,0 +1,161 @@ +# qflash: OpenAI Proxy Gateway for Qwen3.8-Flash-Next + +Standalone, performant, zero-dependency Go OpenAI proxy gateway for the **Qwen3.8-Flash-Next** Hugging Face Gradio space (`https://halvo78-qwen3-8-flash-next-playground.hf.space`). + +Created by Luxferre in 2026, released into the public domain. + +--- + +## Features + +- **Zero External Dependencies**: Built entirely with Go standard library packages (`net/http`, `encoding/json`, `bufio`, etc.). +- **OpenAI-Compatible API**: Implements standard `/v1/chat/completions` (streaming & non-streaming) and `/v1/models`. +- **Real-Time Token Streaming**: Streams SSE chunks with incremental token delivery directly to clients. +- **Deep Reasoning Separation**: + - Automatically isolates thinking traces from both standard `...` tags and the playground's blockquote thinking blocks (`> 💭 **Thinking Process...**`). + - Emits pure thought traces to `delta.reasoning_content` (streaming) and `message.reasoning_content` (non-streaming). + - Keeps `delta.content` and `message.content` clean. +- **Stateful Streaming Tool Call Interception**: + - Injects tool schemas into system instructions. + - Intercepts `` blocks in real time via `StreamToolCallFilter` without leaking raw XML or JSON into `delta.content`. + - Emits structured `delta.tool_calls` chunks and sets `finish_reason: "tool_calls"`. +- **Reasoning Effort Control**: Respects standard `reasoning_effort: "none"` to switch dynamically into high-speed Instruct Mode. +- **Zero-Dependency SOCKS5 Proxy Client**: + - RFC 1928 and RFC 1929 compliant client with domain resolution (`socks5h://`), IPv4, IPv6, and authentication. + - Wireable via `-socks` CLI flag or `ALL_PROXY` / `SOCKS5_PROXY` environment variables. +- **Bring Your Own Key (BYOK) Pass-through**: + - Passes client API keys or custom base URLs directly to upstream inference engines when provided. + +--- + +## Architecture & Model Aliases + +The gateway serves the following models under `/v1/models`: + +| Model ID | Target Model | Description | +|---|---|---| +| `Qwen/Qwen3.8-Flash-Next` | `Qwen/Qwen3.8-Flash-Next` | Primary playground model (125B MoE, 6B activated) | +| `qwen3.8-flash-next` | `Qwen/Qwen3.8-Flash-Next` | Standard lowercase alias | +| `qwen-flash-next` | `Qwen/Qwen3.8-Flash-Next` | Shorthand alias | +| `qwen-flash` | `Qwen/Qwen3.8-Flash-Next` | Quick convenience alias | + +Any unlisted custom model name requested by the client is passed through directly. + +--- + +## Build Instructions + +Build binary with Go: + +```bash +make qflash +``` + +Or run test suite: + +```bash +make test +``` + +The resulting binary will be placed at `bin/qflash`. + +--- + +## Configuration Flags & Environment Variables + +| Flag | Shorthand | Environment Variable | Default | Description | +|---|---|---|---|---| +| `-port` | | `PORT` | `8080` | Port to bind the HTTP server | +| `-endpoint` | | | `https://halvo78-qwen3-8-flash-next-playground.hf.space` | Upstream Gradio space base URL | +| `-model` | | | `Qwen/Qwen3.8-Flash-Next` | Default model ID | +| `-thinking` | `-enable-thinking` | | `true` | Enable chain-of-thought reasoning by default | +| `-hf-token` | | `HF_TOKEN` | `""` | Hugging Face user access token | +| `-api-key` | | `OPENAI_API_KEY` / `QWEN_API_KEY` | `""` | Upstream inference engine API key | +| `-base-url` | | `OPENAI_BASE_URL` / `QWEN_BASE_URL` | `""` | Upstream inference engine base URL | +| `-user-agent` | `-ua` | `USER_AGENT` | Firefox 153 on Linux | Custom User-Agent header | +| `-socks` | `-proxy`, `-socks5` | `ALL_PROXY`, `SOCKS5_PROXY` | `""` | SOCKS5 proxy URL (`socks5://127.0.0.1:1080`) | + +--- + +## Usage Examples + +### 1. Launch Gateway + +```bash +./bin/qflash -port 8080 +``` + +### 2. List Models + +```bash +curl http://127.0.0.1:8080/v1/models +``` + +### 3. Non-Streaming Chat Completion + +```bash +curl -X POST http://127.0.0.1:8080/v1/chat/completions \ + -H "Content-Type: application/json" \ + -d '{ + "model": "qwen3.8-flash-next", + "messages": [ + {"role": "user", "content": "Explain QSA micro-blocks in one sentence."} + ], + "stream": false + }' +``` + +### 4. Streaming Chat Completion (Real-Time SSE) + +```bash +curl -N -X POST http://127.0.0.1:8080/v1/chat/completions \ + -H "Content-Type: application/json" \ + -d '{ + "model": "qwen-flash", + "messages": [ + {"role": "user", "content": "Write a quick Python countdown loop."} + ], + "stream": true + }' +``` + +### 5. Instruct Mode (Disable Thinking) + +```bash +curl -X POST http://127.0.0.1:8080/v1/chat/completions \ + -H "Content-Type: application/json" \ + -d '{ + "model": "qwen-flash", + "messages": [ + {"role": "user", "content": "Hello!"} + ], + "reasoning_effort": "none" + }' +``` + +### 6. Python OpenAI SDK Integration + +```python +from openai import OpenAI + +client = OpenAI( + base_url="http://127.0.0.1:8080/v1", + api_key="sk-dummy" +) + +stream = client.chat.completions.create( + model="qwen3.8-flash-next", + messages=[ + {"role": "user", "content": "Prove that the sum of the first n odd numbers is n^2."} + ], + stream=True +) + +for chunk in stream: + delta = chunk.choices[0].delta + if hasattr(delta, "reasoning_content") and delta.reasoning_content: + print(f"[THINK] {delta.reasoning_content}", end="", flush=True) + if delta.content: + print(delta.content, end="", flush=True) +print() +``` diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..62b2fdc --- /dev/null +++ b/go.mod @@ -0,0 +1,3 @@ +module qflash + +go 1.22 diff --git a/main.go b/main.go new file mode 100644 index 0000000..6eb95d7 --- /dev/null +++ b/main.go @@ -0,0 +1,1710 @@ +// qflash: Standalone OpenAI-compatible gateway for Qwen3.8-Flash-Next Gradio space +// Created by Luxferre in 2026, released into the public domain + +package main + +import ( + "bufio" + "bytes" + "context" + "crypto/rand" + "encoding/json" + "flag" + "fmt" + "io" + "log" + "net" + "net/http" + "os" + "strconv" + "strings" + "time" +) + +var ( + DefaultUserAgent = "Mozilla/5.0 (X11; Linux x86_64; rv:153.0) Gecko/20100101 Firefox/153.0" + ConfiguredUserAgent string + ConfiguredToken string + ConfiguredAPIKey string + ConfiguredBaseURL string + ConfiguredModel string + EnableThinkingDefault = true +) + +// --------------------------------------------------------------------------- +// OpenAI API Data Structures +// --------------------------------------------------------------------------- + +type ModelItem struct { + ID string `json:"id"` + Object string `json:"object"` + Created int64 `json:"created"` + OwnedBy string `json:"owned_by"` +} + +type ModelsResponse struct { + Object string `json:"object"` + Data []ModelItem `json:"data"` +} + +type ToolCallFunction struct { + Name string `json:"name"` + Arguments string `json:"arguments"` +} + +type ToolCall struct { + Index *int `json:"index,omitempty"` + ID string `json:"id,omitempty"` + Type string `json:"type,omitempty"` + Function ToolCallFunction `json:"function"` +} + +type Tool struct { + Type string `json:"type"` + Function interface{} `json:"function"` +} + +type ChatMessage struct { + Role string `json:"role"` + Content interface{} `json:"content"` + ReasoningContent string `json:"reasoning_content,omitempty"` + ToolCalls []ToolCall `json:"tool_calls,omitempty"` + ToolCallID string `json:"tool_call_id,omitempty"` + Name string `json:"name,omitempty"` +} + +func (m *ChatMessage) GetContentString() string { + if m.Content == nil { + return "" + } + if str, ok := m.Content.(string); ok { + return str + } + if parts, ok := m.Content.([]interface{}); ok { + var sb strings.Builder + for _, p := range parts { + if str, ok := p.(string); ok { + sb.WriteString(str) + } else if itemMap, ok := p.(map[string]interface{}); ok { + if textVal, ok := itemMap["text"].(string); ok { + sb.WriteString(textVal) + } + } + } + return sb.String() + } + b, err := json.Marshal(m.Content) + if err == nil { + return string(b) + } + return fmt.Sprintf("%v", m.Content) +} + +type ChatCompletionRequest struct { + Model string `json:"model"` + Messages []ChatMessage `json:"messages"` + Tools []Tool `json:"tools,omitempty"` + ToolChoice interface{} `json:"tool_choice,omitempty"` + Stream bool `json:"stream"` + MaxTokens int `json:"max_tokens"` + MaxCompletionTokens int `json:"max_completion_tokens"` + Temperature *float64 `json:"temperature,omitempty"` + TopP *float64 `json:"top_p,omitempty"` + ReasoningEffort string `json:"reasoning_effort,omitempty"` +} + +type ChatCompletionResponseChoice struct { + Index int `json:"index"` + Message ChatMessage `json:"message"` + FinishReason string `json:"finish_reason"` +} + +type Usage struct { + PromptTokens int `json:"prompt_tokens"` + CompletionTokens int `json:"completion_tokens"` + TotalTokens int `json:"total_tokens"` +} + +type ChatCompletionResponse struct { + ID string `json:"id"` + Object string `json:"object"` + Created int64 `json:"created"` + Model string `json:"model"` + Choices []ChatCompletionResponseChoice `json:"choices"` + Usage Usage `json:"usage"` +} + +type StreamDelta struct { + Role string `json:"role,omitempty"` + Content string `json:"content,omitempty"` + ReasoningContent string `json:"reasoning_content,omitempty"` + ToolCalls []ToolCall `json:"tool_calls,omitempty"` +} + +type StreamChoice struct { + Index int `json:"index"` + Delta StreamDelta `json:"delta"` + FinishReason *string `json:"finish_reason,omitempty"` +} + +type StreamResponse struct { + ID string `json:"id"` + Object string `json:"object"` + Created int64 `json:"created"` + Model string `json:"model"` + Choices []StreamChoice `json:"choices"` +} + +type GradioJoinResponse struct { + EventID string `json:"event_id"` +} + +// --------------------------------------------------------------------------- +// Zero-Dependency SOCKS5 Proxy Client (RFC 1928 / RFC 1929) +// --------------------------------------------------------------------------- + +type SOCKS5Config struct { + Address string + Username string + Password string +} + +func ParseSOCKS5URL(proxyURL string) (*SOCKS5Config, error) { + cleanURL := strings.TrimSpace(proxyURL) + if cleanURL == "" { + return nil, nil + } + if strings.HasPrefix(cleanURL, "socks5://") { + cleanURL = strings.TrimPrefix(cleanURL, "socks5://") + } else if strings.HasPrefix(cleanURL, "socks5h://") { + cleanURL = strings.TrimPrefix(cleanURL, "socks5h://") + } + + cfg := &SOCKS5Config{} + if atIdx := strings.LastIndex(cleanURL, "@"); atIdx != -1 { + userPass := cleanURL[:atIdx] + cfg.Address = cleanURL[atIdx+1:] + if colonIdx := strings.Index(userPass, ":"); colonIdx != -1 { + cfg.Username = userPass[:colonIdx] + cfg.Password = userPass[colonIdx+1:] + } else { + cfg.Username = userPass + } + } else { + cfg.Address = cleanURL + } + + if !strings.Contains(cfg.Address, ":") { + cfg.Address = cfg.Address + ":1080" + } + return cfg, nil +} + +func DialSOCKS5(ctx context.Context, proxyURL, targetAddr string) (net.Conn, error) { + cfg, err := ParseSOCKS5URL(proxyURL) + if err != nil { + return nil, fmt.Errorf("invalid socks5 proxy configuration: %w", err) + } + if cfg == nil { + var d net.Dialer + return d.DialContext(ctx, "tcp", targetAddr) + } + + var d net.Dialer + conn, err := d.DialContext(ctx, "tcp", cfg.Address) + if err != nil { + return nil, fmt.Errorf("failed to connect to socks5 proxy at %s: %w", cfg.Address, err) + } + + deadline, ok := ctx.Deadline() + if !ok { + deadline = time.Now().Add(30 * time.Second) + } + conn.SetDeadline(deadline) + defer conn.SetDeadline(time.Time{}) + + // 1. Negotiation Greeting (RFC 1928) + var greeting []byte + if cfg.Username != "" { + greeting = []byte{0x05, 0x02, 0x00, 0x02} + } else { + greeting = []byte{0x05, 0x01, 0x00} + } + + if _, err := conn.Write(greeting); err != nil { + conn.Close() + return nil, fmt.Errorf("failed to write socks5 greeting: %w", err) + } + + resp := make([]byte, 2) + if _, err := io.ReadFull(conn, resp); err != nil { + conn.Close() + return nil, fmt.Errorf("failed to read socks5 greeting response: %w", err) + } + + if resp[0] != 0x05 { + conn.Close() + return nil, fmt.Errorf("unsupported socks version: 0x%02x", resp[0]) + } + + // 2. Authentication if required (RFC 1929) + if resp[1] == 0x02 { + if cfg.Username == "" { + conn.Close() + return nil, fmt.Errorf("socks5 proxy requires authentication, but no credentials provided") + } + uLen := byte(len(cfg.Username)) + pLen := byte(len(cfg.Password)) + authReq := []byte{0x01, uLen} + authReq = append(authReq, []byte(cfg.Username)...) + authReq = append(authReq, pLen) + authReq = append(authReq, []byte(cfg.Password)...) + + if _, err := conn.Write(authReq); err != nil { + conn.Close() + return nil, fmt.Errorf("failed to send socks5 authentication: %w", err) + } + + authResp := make([]byte, 2) + if _, err := io.ReadFull(conn, authResp); err != nil { + conn.Close() + return nil, fmt.Errorf("failed to read socks5 auth response: %w", err) + } + if authResp[1] != 0x00 { + conn.Close() + return nil, fmt.Errorf("socks5 authentication failed with status 0x%02x", authResp[1]) + } + } else if resp[1] != 0x00 { + conn.Close() + return nil, fmt.Errorf("socks5 proxy rejected authentication methods: 0x%02x", resp[1]) + } + + // 3. Connection Request (CONNECT command) + host, portStr, err := net.SplitHostPort(targetAddr) + if err != nil { + conn.Close() + return nil, fmt.Errorf("invalid target address %s: %w", targetAddr, err) + } + + port, err := strconv.Atoi(portStr) + if err != nil || port < 1 || port > 65535 { + conn.Close() + return nil, fmt.Errorf("invalid port in target address %s", targetAddr) + } + + reqBuf := []byte{0x05, 0x01, 0x00} + ip := net.ParseIP(host) + if ip4 := ip.To4(); ip4 != nil { + reqBuf = append(reqBuf, 0x01) + reqBuf = append(reqBuf, ip4...) + } else if ip6 := ip.To16(); ip6 != nil { + reqBuf = append(reqBuf, 0x04) + reqBuf = append(reqBuf, ip6...) + } else { + reqBuf = append(reqBuf, 0x03, byte(len(host))) + reqBuf = append(reqBuf, []byte(host)...) + } + reqBuf = append(reqBuf, byte(port>>8), byte(port&0xFF)) + + if _, err := conn.Write(reqBuf); err != nil { + conn.Close() + return nil, fmt.Errorf("failed to send socks5 connect request: %w", err) + } + + // 4. Connection Response + respHdr := make([]byte, 4) + if _, err := io.ReadFull(conn, respHdr); err != nil { + conn.Close() + return nil, fmt.Errorf("failed to read socks5 connect response: %w", err) + } + + if respHdr[1] != 0x00 { + conn.Close() + return nil, fmt.Errorf("socks5 connect failed with reply code 0x%02x", respHdr[1]) + } + + switch respHdr[3] { + case 0x01: + bnd := make([]byte, 6) + io.ReadFull(conn, bnd) + case 0x03: + lenBuf := make([]byte, 1) + io.ReadFull(conn, lenBuf) + bnd := make([]byte, int(lenBuf[0])+2) + io.ReadFull(conn, bnd) + case 0x04: + bnd := make([]byte, 18) + io.ReadFull(conn, bnd) + } + + return conn, nil +} + +// --------------------------------------------------------------------------- +// Helper Utilities +// --------------------------------------------------------------------------- + +func GenerateUUID() string { + var b [16]byte + _, err := rand.Read(b[:]) + if err != nil { + return "00000000-0000-4000-8000-000000000000" + } + b[6] = (b[6] & 0x0f) | 0x40 + b[8] = (b[8] & 0x3f) | 0x80 + return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:]) +} + +func FibonacciDelay(attempt int) time.Duration { + if attempt <= 0 { + return 1 * time.Second + } + a, b := 1, 1 + for i := 1; i < attempt; i++ { + a, b = b, a+b + } + return time.Duration(a) * time.Second +} + +func DoWithFibonacciRetry(client *http.Client, makeReq func() (*http.Request, error), maxRetries int) (*http.Response, error) { + var lastErr error + for attempt := 1; attempt <= maxRetries; attempt++ { + req, err := makeReq() + if err != nil { + return nil, err + } + resp, err := client.Do(req) + if err == nil && resp.StatusCode == http.StatusOK { + return resp, nil + } + + if resp != nil { + respBody, _ := io.ReadAll(resp.Body) + resp.Body.Close() + lastErr = fmt.Errorf("HTTP status %d: %s", resp.StatusCode, string(respBody)) + } else { + lastErr = err + } + + if attempt < maxRetries { + delay := FibonacciDelay(attempt) + time.Sleep(delay) + } + } + return nil, fmt.Errorf("request failed after %d retries: %v", maxRetries, lastErr) +} + +func EnableCORS(w http.ResponseWriter) { + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("Access-Control-Allow-Methods", "GET, POST, OPTIONS") + w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization, api-key, X-User-Agent, X-HF-Token, X-Base-URL, X-Model-ID") +} + +func ResolveMaxTokens(req ChatCompletionRequest) int { + mt := req.MaxTokens + if mt == 0 && req.MaxCompletionTokens > 0 { + mt = req.MaxCompletionTokens + } + if mt <= 0 { + mt = 8192 + } + if mt > 32768 { + mt = 32768 + } + return mt +} + +func EffectiveUserAgent(r *http.Request) string { + if r != nil { + if c := r.Header.Get("X-User-Agent"); c != "" { + return c + } + } + if ConfiguredUserAgent != "" { + return ConfiguredUserAgent + } + return DefaultUserAgent +} + +func EffectiveHFToken(r *http.Request) string { + if r != nil { + if hf := r.Header.Get("X-HF-Token"); hf != "" { + return hf + } + if auth := r.Header.Get("Authorization"); auth != "" { + if strings.HasPrefix(strings.ToLower(auth), "bearer ") { + tok := strings.TrimSpace(auth[7:]) + if strings.HasPrefix(tok, "hf_") { + return tok + } + } + } + } + if ConfiguredToken != "" { + return ConfiguredToken + } + if envTok := os.Getenv("HF_TOKEN"); envTok != "" { + return envTok + } + return "" +} + +func EffectiveUpstreamKey(r *http.Request) string { + if r != nil { + if auth := r.Header.Get("Authorization"); auth != "" { + if strings.HasPrefix(strings.ToLower(auth), "bearer ") { + tok := strings.TrimSpace(auth[7:]) + if tok != "" && tok != "-" && !strings.HasPrefix(tok, "hf_") && tok != "sk-dummy" { + return tok + } + } + } + if key := r.Header.Get("api-key"); key != "" { + return key + } + } + if ConfiguredAPIKey != "" { + return ConfiguredAPIKey + } + if envKey := os.Getenv("OPENAI_API_KEY"); envKey != "" { + return envKey + } + if envKey := os.Getenv("QWEN_API_KEY"); envKey != "" { + return envKey + } + return "" +} + +func EffectiveUpstreamBaseURL(r *http.Request) string { + if r != nil { + if bu := r.Header.Get("X-Base-URL"); bu != "" { + return bu + } + } + if ConfiguredBaseURL != "" { + return ConfiguredBaseURL + } + if envBU := os.Getenv("OPENAI_BASE_URL"); envBU != "" { + return envBU + } + if envBU := os.Getenv("QWEN_BASE_URL"); envBU != "" { + return envBU + } + return "" +} + +func EffectiveModelID(reqModel string, defaultModel string) string { + clean := strings.TrimSpace(reqModel) + if clean == "" { + return defaultModel + } + switch strings.ToLower(clean) { + case "qwen/qwen3.8-flash-next", "qwen3.8-flash-next", "qwen-flash-next", "qwen-flash", "qwen3.8-flash", "qwen": + return "Qwen/Qwen3.8-Flash-Next" + default: + return clean + } +} + +// --------------------------------------------------------------------------- +// Response Framing +// --------------------------------------------------------------------------- + +type FinalOutput struct { + Content interface{} + ReasoningContent string + ToolCalls []ToolCall + FinishReason string +} + +func WriteCompletionResponse(w http.ResponseWriter, completionID string, created int64, model string, out FinalOutput) { + finish := out.FinishReason + if finish == "" { + finish = "stop" + } + resp := ChatCompletionResponse{ + ID: completionID, + Object: "chat.completion", + Created: created, + Model: model, + Choices: []ChatCompletionResponseChoice{ + { + Index: 0, + Message: ChatMessage{ + Role: "assistant", + Content: out.Content, + ReasoningContent: out.ReasoningContent, + ToolCalls: out.ToolCalls, + }, + FinishReason: finish, + }, + }, + Usage: Usage{ + PromptTokens: 0, + CompletionTokens: 0, + TotalTokens: 0, + }, + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(resp) +} + +type Streamer struct { + w http.ResponseWriter + flusher http.Flusher + id string + created int64 + model string +} + +func NewStreamer(w http.ResponseWriter, flusher http.Flusher, id string, created int64, model string) *Streamer { + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + return &Streamer{w: w, flusher: flusher, id: id, created: created, model: model} +} + +func (s *Streamer) Role() { + sendStreamDelta(s.w, s.flusher, s.id, s.created, s.model, StreamDelta{Role: "assistant"}) +} + +func (s *Streamer) Reasoning(text string) { + sendStreamDelta(s.w, s.flusher, s.id, s.created, s.model, StreamDelta{ReasoningContent: text}) +} + +func (s *Streamer) Content(text string) { + sendStreamDelta(s.w, s.flusher, s.id, s.created, s.model, StreamDelta{Content: text}) +} + +func (s *Streamer) ToolCallDelta(tc ToolCall) { + sendStreamDelta(s.w, s.flusher, s.id, s.created, s.model, StreamDelta{ToolCalls: []ToolCall{tc}}) +} + +func (s *Streamer) Finish(reason string) { + sendStreamChunk(s.w, s.flusher, s.id, s.created, s.model, StreamDelta{}, &reason) +} + +func (s *Streamer) Done() { + fmt.Fprintf(s.w, "data: [DONE]\n\n") + if s.flusher != nil { + s.flusher.Flush() + } +} + +func sendStreamDelta(w http.ResponseWriter, flusher http.Flusher, completionID string, createdTime int64, modelName string, delta StreamDelta) { + sendStreamChunk(w, flusher, completionID, createdTime, modelName, delta, nil) +} + +func sendStreamChunk(w http.ResponseWriter, flusher http.Flusher, completionID string, createdTime int64, modelName string, delta StreamDelta, finishReason *string) { + chunk := StreamResponse{ + ID: completionID, + Object: "chat.completion.chunk", + Created: createdTime, + Model: modelName, + Choices: []StreamChoice{ + { + Index: 0, + Delta: delta, + FinishReason: finishReason, + }, + }, + } + b, _ := json.Marshal(chunk) + fmt.Fprintf(w, "data: %s\n\n", b) + if flusher != nil { + flusher.Flush() + } +} + +// --------------------------------------------------------------------------- +// Tool Calling and Reasoning Extraction +// --------------------------------------------------------------------------- + +func cleanJSONBlock(input string) string { + s := strings.TrimSpace(input) + if strings.HasPrefix(s, "```") { + lines := strings.Split(s, "\n") + if len(lines) >= 2 { + if strings.HasPrefix(lines[len(lines)-1], "```") { + lines = lines[1 : len(lines)-1] + } else { + lines = lines[1:] + } + s = strings.TrimSpace(strings.Join(lines, "\n")) + } + } + return s +} + +func sanitizeJSONValue(v interface{}) interface{} { + switch val := v.(type) { + case string: + return strings.TrimSpace(val) + case map[string]interface{}: + cleanMap := make(map[string]interface{}) + for k, childV := range val { + cleanKey := strings.TrimSpace(k) + cleanMap[cleanKey] = sanitizeJSONValue(childV) + } + return cleanMap + case []interface{}: + cleanSlice := make([]interface{}, len(val)) + for i, childV := range val { + cleanSlice[i] = sanitizeJSONValue(childV) + } + return cleanSlice + default: + return v + } +} + +func parseSingleToolCall(jsonStr string) (ToolCall, bool) { + cleaned := cleanJSONBlock(jsonStr) + var raw map[string]interface{} + if err := json.Unmarshal([]byte(cleaned), &raw); err == nil { + sanitizedRaw, ok := sanitizeJSONValue(raw).(map[string]interface{}) + if !ok { + sanitizedRaw = raw + } + + for _, wrapperKey := range []string{"function", "function_call", "tool_call"} { + if fnObj, ok := sanitizedRaw[wrapperKey].(map[string]interface{}); ok { + if nameVal, ok := fnObj["name"].(string); ok && nameVal != "" { + argsStr := "{}" + var argsVal interface{} + if a, hasA := fnObj["arguments"]; hasA { + argsVal = a + } else if p, hasP := fnObj["parameters"]; hasP { + argsVal = p + } else if args, hasArgs := fnObj["args"]; hasArgs { + argsVal = args + } + if argsVal != nil { + if s, isStr := argsVal.(string); isStr { + var innerObj interface{} + if json.Unmarshal([]byte(s), &innerObj) == nil { + b, _ := json.Marshal(sanitizeJSONValue(innerObj)) + argsStr = string(b) + } else { + argsStr = strings.TrimSpace(s) + } + } else { + b, _ := json.Marshal(sanitizeJSONValue(argsVal)) + argsStr = string(b) + } + } + return ToolCall{ + ID: "call_" + GenerateUUID()[:8], + Type: "function", + Function: ToolCallFunction{ + Name: nameVal, + Arguments: argsStr, + }, + }, true + } + } + } + + nameVal := "" + for _, key := range []string{"name", "function", "action", "call"} { + if n, ok := sanitizedRaw[key].(string); ok && n != "" { + nameVal = n + break + } + } + + if nameVal != "" { + argsStr := "{}" + var argsVal interface{} + for _, key := range []string{"arguments", "parameters", "args", "input"} { + if a, ok := sanitizedRaw[key]; ok { + argsVal = a + break + } + } + if argsVal != nil { + if s, isStr := argsVal.(string); isStr { + var innerObj interface{} + if json.Unmarshal([]byte(s), &innerObj) == nil { + b, _ := json.Marshal(sanitizeJSONValue(innerObj)) + argsStr = string(b) + } else { + argsStr = strings.TrimSpace(s) + } + } else { + b, _ := json.Marshal(sanitizeJSONValue(argsVal)) + argsStr = string(b) + } + } else { + argsMap := make(map[string]interface{}) + for k, v := range sanitizedRaw { + if k != "name" && k != "function" && k != "type" && k != "action" && k != "call" { + argsMap[k] = v + } + } + if len(argsMap) > 0 { + b, _ := json.Marshal(sanitizeJSONValue(argsMap)) + argsStr = string(b) + } + } + return ToolCall{ + ID: "call_" + GenerateUUID()[:8], + Type: "function", + Function: ToolCallFunction{ + Name: nameVal, + Arguments: argsStr, + }, + }, true + } + } + + return ToolCall{}, false +} + +func parseXMLToolCall(block string) (ToolCall, bool) { + inner := strings.TrimSpace(block) + if strings.HasPrefix(inner, "") { + inner = strings.TrimPrefix(inner, "") + } + if strings.HasSuffix(inner, "") { + inner = strings.TrimSuffix(inner, "") + } + inner = cleanJSONBlock(inner) + + if tc, ok := parseSingleToolCall(inner); ok { + return tc, true + } + + var fnName string + if strings.Contains(inner, "") && strings.Contains(inner, "") { + nStart := strings.Index(inner, "") + len("") + nEnd := strings.Index(inner, "") + if nStart < nEnd { + fnName = strings.TrimSpace(inner[nStart:nEnd]) + } + } else if strings.Contains(inner, "") && strings.Contains(inner, "") { + nStart := strings.Index(inner, "") + len("") + nEnd := strings.Index(inner, "") + if nStart < nEnd { + fnName = strings.TrimSpace(inner[nStart:nEnd]) + } + } else if strings.Contains(inner, "") && strings.Contains(inner, "") { + nStart := strings.Index(inner, "") + len("") + nEnd := strings.Index(inner, "") + if nStart < nEnd { + fnName = strings.TrimSpace(inner[nStart:nEnd]) + } + } + + var argsStr string + if strings.Contains(inner, "") && strings.Contains(inner, "") { + aStart := strings.Index(inner, "") + len("") + aEnd := strings.Index(inner, "") + if aStart < aEnd { + argsStr = strings.TrimSpace(inner[aStart:aEnd]) + } + } else if strings.Contains(inner, "") && strings.Contains(inner, "") { + pStart := strings.Index(inner, "") + len("") + pEnd := strings.Index(inner, "") + if pStart < pEnd { + argsStr = strings.TrimSpace(inner[pStart:pEnd]) + } + } + + if fnName != "" { + if argsStr == "" { + argsStr = "{}" + } + return ToolCall{ + ID: "call_" + GenerateUUID()[:8], + Type: "function", + Function: ToolCallFunction{ + Name: fnName, + Arguments: argsStr, + }, + }, true + } + + return ToolCall{}, false +} + +func ExtractToolCallBlocks(content string) (blocks []string, remaining string) { + s := content + remaining = content + + for strings.Contains(s, "") { + sIdx := strings.Index(s, "") + rest := s[sIdx+len(""):] + + relNextSIdx := strings.Index(rest, "") + var nextSIdx int + if relNextSIdx != -1 { + nextSIdx = sIdx + len("") + relNextSIdx + } else { + nextSIdx = -1 + } + + relEIdx := strings.Index(rest, "") + var eIdx int + if relEIdx != -1 { + eIdx = sIdx + len("") + relEIdx + } else { + eIdx = -1 + } + + var blockText string + var blockEndPos int + + if eIdx != -1 && (nextSIdx == -1 || eIdx < nextSIdx) { + blockEndPos = eIdx + len("") + blockText = s[sIdx:blockEndPos] + s = s[blockEndPos:] + } else if nextSIdx != -1 { + blockEndPos = nextSIdx + blockText = s[sIdx:blockEndPos] + s = s[blockEndPos:] + } else { + blockText = s[sIdx:] + s = "" + } + + blocks = append(blocks, blockText) + } + + for strings.Contains(remaining, "") { + st := strings.Index(remaining, "") + rest := remaining[st+len(""):] + + relNext := strings.Index(rest, "") + var nextSt int + if relNext != -1 { + nextSt = st + len("") + relNext + } else { + nextSt = -1 + } + + relEn := strings.Index(rest, "") + var en int + if relEn != -1 { + en = st + len("") + relEn + } else { + en = -1 + } + + if en != -1 && (nextSt == -1 || en < nextSt) { + remaining = strings.TrimSpace(remaining[:st] + remaining[en+len(""):]) + } else if nextSt != -1 { + remaining = strings.TrimSpace(remaining[:st] + remaining[nextSt:]) + } else { + remaining = strings.TrimSpace(remaining[:st]) + } + } + + return blocks, remaining +} + +func DetectToolCalls(content string) ([]ToolCall, string, bool) { + blocks, remaining := ExtractToolCallBlocks(content) + var calls []ToolCall + + for _, block := range blocks { + if toolCall, ok := parseXMLToolCall(block); ok { + calls = append(calls, toolCall) + } + } + + if len(calls) > 0 { + return calls, remaining, true + } + + if tc, ok := parseSingleToolCall(strings.TrimSpace(content)); ok { + return []ToolCall{tc}, "", true + } + + return nil, content, false +} + +// CleanGradioThought strips Markdown blockquote indicators ('> ') and header lines +// from the playground's reasoning trace. +func CleanGradioThought(raw string) string { + lines := strings.Split(raw, "\n") + var cleaned []string + skipHeader := true + + for _, line := range lines { + trimmed := strings.TrimSpace(line) + if skipHeader { + low := strings.ToLower(trimmed) + if strings.Contains(low, "thinking process") || trimmed == ">" || trimmed == "" { + continue + } + skipHeader = false + } + + if strings.HasPrefix(line, "> ") { + cleaned = append(cleaned, strings.TrimPrefix(line, "> ")) + } else if line == ">" { + cleaned = append(cleaned, "") + } else if strings.HasPrefix(line, ">") { + cleaned = append(cleaned, strings.TrimPrefix(line, ">")) + } else { + cleaned = append(cleaned, line) + } + } + + return strings.TrimSpace(strings.Join(cleaned, "\n")) +} + +// SeparateReasoningAndContent separates thinking/reasoning traces from actual response content. +// Handles both standard ... tags and the HF Gradio playground blockquote format. +func SeparateReasoningAndContent(text string) (string, string) { + // 1. Standard tags + if strings.Contains(text, "") { + sIdx := strings.Index(text, "") + if eIdx := strings.Index(text, ""); eIdx != -1 && eIdx > sIdx { + reasoning := text[sIdx+len("") : eIdx] + content := text[:sIdx] + text[eIdx+len(""):] + return strings.TrimSpace(reasoning), strings.TrimSpace(content) + } else { + reasoning := text[sIdx+len(""):] + content := text[:sIdx] + return strings.TrimSpace(reasoning), strings.TrimSpace(content) + } + } + + // 2. HF Gradio Playground blockquote format: "> ... Thinking Process" + low := strings.ToLower(text) + if strings.Contains(low, "thinking process") && (strings.HasPrefix(strings.TrimSpace(text), ">") || strings.Contains(text, "\n>")) { + // Check if completion divider has arrived: "\n\n---\n\n" + if divIdx := strings.Index(text, "\n\n---\n\n"); divIdx != -1 { + rawThought := text[:divIdx] + content := text[divIdx+len("\n\n---\n\n"):] + return CleanGradioThought(rawThought), content + } + // Check if intermediate streaming divider is present: "\n\n---\n*Generating response...*" + if divIdx := strings.Index(text, "\n\n---\n*Generating response...*"); divIdx != -1 { + rawThought := text[:divIdx] + return CleanGradioThought(rawThought), "" + } + // If still in thought generation phase + if strings.HasPrefix(strings.TrimSpace(text), ">") { + return CleanGradioThought(text), "" + } + } + + return "", text +} + +func FormatToolsPrompt(tools []Tool) string { + if len(tools) == 0 { + return "" + } + b, err := json.MarshalIndent(tools, "", " ") + if err != nil { + return "" + } + var sb strings.Builder + sb.WriteString("\n\n[Available Tools]\nYou have access to the following tools:\n```json\n") + sb.WriteString(string(b)) + sb.WriteString("\n```\n") + sb.WriteString("If you choose to invoke one or more tools, respond ONLY with the tool invocation formatted as:\n") + sb.WriteString("\n{\"name\": \"function_name\", \"arguments\": {\"param\": \"value\"}}\n\n") + sb.WriteString("Do not add conversational preamble around the tool call when invoking a tool.\n") + return sb.String() +} + +// StreamToolCallFilter is a stateful streaming filter that intercepts tags +// in real-time, preventing control tags from leaking to delta.content while emitting delta.tool_calls. +type StreamToolCallFilter struct { + inToolCall bool + toolCallBuf string + buf string + toolIndex int + emittedCall bool +} + +func (f *StreamToolCallFilter) Feed(chunk string, onContent func(string), onToolCall func(ToolCall)) { + f.buf += chunk + for len(f.buf) > 0 { + if !f.inToolCall { + idx := strings.Index(f.buf, "") + if idx != -1 { + if idx > 0 { + onContent(f.buf[:idx]) + } + f.inToolCall = true + f.toolCallBuf = "" + f.buf = f.buf[idx+len(""):] + } else { + matchLen := 0 + tag := "" + for i := 1; i < len(tag) && i <= len(f.buf); i++ { + if strings.HasSuffix(f.buf, tag[:i]) { + matchLen = i + } + } + if matchLen > 0 { + onContent(f.buf[:len(f.buf)-matchLen]) + f.buf = f.buf[len(f.buf)-matchLen:] + break + } else { + onContent(f.buf) + f.buf = "" + break + } + } + } else { + idx := strings.Index(f.buf, "") + if idx != -1 { + f.toolCallBuf += f.buf[:idx] + f.inToolCall = false + f.buf = f.buf[idx+len(""):] + + if tc, ok := parseSingleToolCall(f.toolCallBuf); ok { + idxCopy := f.toolIndex + tc.Index = &idxCopy + f.toolIndex++ + f.emittedCall = true + onToolCall(tc) + } else if tc2, ok2 := parseXMLToolCall("" + f.toolCallBuf + ""); ok2 { + idxCopy := f.toolIndex + tc2.Index = &idxCopy + f.toolIndex++ + f.emittedCall = true + onToolCall(tc2) + } else { + onContent("" + f.toolCallBuf + "") + } + f.toolCallBuf = "" + } else { + matchLen := 0 + tag := "" + for i := 1; i < len(tag) && i <= len(f.buf); i++ { + if strings.HasSuffix(f.buf, tag[:i]) { + matchLen = i + } + } + if matchLen > 0 { + f.toolCallBuf += f.buf[:len(f.buf)-matchLen] + f.buf = f.buf[len(f.buf)-matchLen:] + break + } else { + f.toolCallBuf += f.buf + f.buf = "" + break + } + } + } + } +} + +func (f *StreamToolCallFilter) Flush(onContent func(string), onToolCall func(ToolCall)) { + if f.inToolCall && len(f.toolCallBuf) > 0 { + if tc, ok := parseSingleToolCall(f.toolCallBuf); ok { + idxCopy := f.toolIndex + tc.Index = &idxCopy + f.emittedCall = true + onToolCall(tc) + } else if tc2, ok2 := parseXMLToolCall("" + f.toolCallBuf + ""); ok2 { + idxCopy := f.toolIndex + tc2.Index = &idxCopy + f.emittedCall = true + onToolCall(tc2) + } else { + onContent("" + f.toolCallBuf) + } + f.toolCallBuf = "" + } + if len(f.buf) > 0 { + onContent(f.buf) + f.buf = "" + } +} + +func (f *StreamToolCallFilter) HasEmittedCalls() bool { + return f.emittedCall +} + +// --------------------------------------------------------------------------- +// Qwen3.8-Flash Service & Gradio Stream Parsing +// --------------------------------------------------------------------------- + +// parseAssistantText extracts the latest assistant message text from the Gradio output array. +func parseAssistantText(dataJSON string) (string, bool) { + var raw []interface{} + if err := json.Unmarshal([]byte(dataJSON), &raw); err != nil || len(raw) == 0 { + return "", false + } + + // The first element is the chatbot message list + msgList, ok := raw[0].([]interface{}) + if !ok || len(msgList) == 0 { + return "", false + } + + // Find the last assistant message + for i := len(msgList) - 1; i >= 0; i-- { + msgMap, ok := msgList[i].(map[string]interface{}) + if !ok { + continue + } + role, _ := msgMap["role"].(string) + if role != "assistant" { + continue + } + + contentVal := msgMap["content"] + if contentStr, ok := contentVal.(string); ok { + return contentStr, true + } + if contentSlice, ok := contentVal.([]interface{}); ok { + var sb strings.Builder + for _, item := range contentSlice { + if s, ok := item.(string); ok { + sb.WriteString(s) + } else if m, ok := item.(map[string]interface{}); ok { + if textVal, ok := m["text"].(string); ok { + sb.WriteString(textVal) + } + } + } + return sb.String(), true + } + } + + return "", false +} + +type QwenService struct { + endpoint string + modelName string + token string + apiKey string + baseURL string + enableThinking bool + client *http.Client +} + +func NewQwenService(endpoint, modelName, token, apiKey, baseURL, socksProxy string, enableThinking bool) *QwenService { + cleanEndpoint := strings.TrimRight(endpoint, "/") + if modelName == "" { + modelName = "Qwen/Qwen3.8-Flash-Next" + } + + transport := &http.Transport{ + MaxIdleConns: 100, + IdleConnTimeout: 90 * time.Second, + TLSHandshakeTimeout: 10 * time.Second, + } + + if socksProxy != "" { + transport.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { + return DialSOCKS5(ctx, socksProxy, addr) + } + } + + return &QwenService{ + endpoint: cleanEndpoint, + modelName: modelName, + token: token, + apiKey: apiKey, + baseURL: baseURL, + enableThinking: enableThinking, + client: &http.Client{Transport: transport, Timeout: 300 * time.Second}, + } +} + +func (s *QwenService) ListModels() []ModelItem { + now := time.Now().Unix() + primaryID := s.modelName + if primaryID == "" { + primaryID = "Qwen/Qwen3.8-Flash-Next" + } + + models := []ModelItem{ + {ID: primaryID, Object: "model", Created: now, OwnedBy: "qwen"}, + {ID: "qwen3.8-flash-next", Object: "model", Created: now, OwnedBy: "qwen"}, + {ID: "qwen-flash-next", Object: "model", Created: now, OwnedBy: "qwen"}, + {ID: "qwen-flash", Object: "model", Created: now, OwnedBy: "qwen"}, + } + + return models +} + +func (s *QwenService) Chat(w http.ResponseWriter, r *http.Request, req ChatCompletionRequest) error { + resolvedModel := EffectiveModelID(req.Model, s.modelName) + maxTokens := ResolveMaxTokens(req) + + var systemPromptStr string + var historyArray []map[string]interface{} + var messageStr string + + var nonSystemMsgs []ChatMessage + for _, msg := range req.Messages { + cStr := msg.GetContentString() + if msg.Role == "system" && systemPromptStr == "" { + systemPromptStr = cStr + } else { + nonSystemMsgs = append(nonSystemMsgs, msg) + } + } + + toolsPrompt := FormatToolsPrompt(req.Tools) + + if len(nonSystemMsgs) > 0 { + for i := 0; i < len(nonSystemMsgs)-1; i++ { + m := nonSystemMsgs[i] + cStr := m.GetContentString() + itemRole := m.Role + + switch m.Role { + case "assistant": + contentBlocks := []interface{}{} + if cStr != "" { + contentBlocks = append(contentBlocks, map[string]interface{}{"text": cStr, "type": "text"}) + } + item := map[string]interface{}{ + "role": "assistant", + "metadata": nil, + "content": contentBlocks, + "options": nil, + } + historyArray = append(historyArray, item) + case "tool", "function": + toolName := m.Name + if toolName == "" { + toolName = m.ToolCallID + } + formatted := fmt.Sprintf("\n%s\n", toolName, cStr) + item := map[string]interface{}{ + "role": "user", + "metadata": nil, + "content": []interface{}{map[string]interface{}{"text": formatted, "type": "text"}}, + "options": nil, + } + historyArray = append(historyArray, item) + default: + item := map[string]interface{}{ + "role": itemRole, + "metadata": nil, + "content": []interface{}{map[string]interface{}{"text": cStr, "type": "text"}}, + "options": nil, + } + historyArray = append(historyArray, item) + } + } + + lastMsg := nonSystemMsgs[len(nonSystemMsgs)-1] + lastContent := lastMsg.GetContentString() + if lastMsg.Role == "tool" || lastMsg.Role == "function" { + toolName := lastMsg.Name + if toolName == "" { + toolName = lastMsg.ToolCallID + } + messageStr = fmt.Sprintf("\n%s\n", toolName, lastContent) + } else { + messageStr = lastContent + } + } + + if toolsPrompt != "" { + if systemPromptStr != "" { + systemPromptStr = systemPromptStr + "\n" + toolsPrompt + } else { + systemPromptStr = strings.TrimSpace(toolsPrompt) + } + } + + // Determine thinking mode: + // Can be controlled by reasoning_effort ("none" disables thinking) or service default + enableThinking := s.enableThinking + if req.ReasoningEffort != "" { + if strings.EqualFold(req.ReasoningEffort, "none") { + enableThinking = false + } else { + enableThinking = true + } + } + + tempVal := 1.0 + if req.Temperature != nil { + tempVal = *req.Temperature + } else if !enableThinking { + tempVal = 0.7 + } + + topPVal := 0.95 + if req.TopP != nil { + topPVal = *req.TopP + } else if !enableThinking { + topPVal = 0.80 + } + + topKVal := 20 + presenceVal := 0.0 + if !enableThinking { + presenceVal = 1.5 + } + + customAPIKey := EffectiveUpstreamKey(r) + if customAPIKey == "" { + customAPIKey = s.apiKey + } + + customBaseURL := EffectiveUpstreamBaseURL(r) + if customBaseURL == "" { + customBaseURL = s.baseURL + } + + effUA := EffectiveUserAgent(r) + effHFToken := EffectiveHFToken(r) + if effHFToken == "" { + effHFToken = s.token + } + + // Gradio parameter list for /chat_response: + // 0: message (dict: text, files) + // 1: history (list of Message objects) + // 2: enable_thinking (bool) + // 3: preserve_thinking (bool) + // 4: temperature (float) + // 5: top_p (float) + // 6: top_k (float) + // 7: presence_penalty (float) + // 8: max_tokens (float) + // 9: system_prompt (str) + // 10: custom_base_url (str) + // 11: custom_api_key (str) + // 12: custom_model_id (str) + gradioData := []interface{}{ + map[string]interface{}{ + "text": messageStr, + "files": []interface{}{}, + }, + historyArray, + enableThinking, + false, // preserve_thinking + tempVal, + topPVal, + topKVal, + presenceVal, + maxTokens, + systemPromptStr, + customBaseURL, + customAPIKey, + resolvedModel, + } + + gradioPayload := map[string]interface{}{"data": gradioData} + jsonPayload, err := json.Marshal(gradioPayload) + if err != nil { + return fmt.Errorf("failed to encode request: %w", err) + } + + callURL := s.endpoint + "/gradio_api/call/chat_response" + makeCallReq := func() (*http.Request, error) { + reqObj, err := http.NewRequest("POST", callURL, bytes.NewBuffer(jsonPayload)) + if err != nil { + return nil, err + } + reqObj.Header.Set("Content-Type", "application/json") + reqObj.Header.Set("User-Agent", effUA) + if effHFToken != "" { + reqObj.Header.Set("Authorization", "Bearer "+effHFToken) + } + return reqObj, nil + } + + resp, err := DoWithFibonacciRetry(s.client, makeCallReq, 5) + if err != nil { + return fmt.Errorf("upstream join error: %w", err) + } + defer resp.Body.Close() + + var joinRes GradioJoinResponse + if err := json.NewDecoder(resp.Body).Decode(&joinRes); err != nil || joinRes.EventID == "" { + return fmt.Errorf("failed to parse Gradio event ID") + } + + streamURL := fmt.Sprintf("%s/gradio_api/call/chat_response/%s", s.endpoint, joinRes.EventID) + makeStreamReq := func() (*http.Request, error) { + reqObj, err := http.NewRequest("GET", streamURL, nil) + if err != nil { + return nil, err + } + reqObj.Header.Set("Accept", "text/event-stream") + reqObj.Header.Set("User-Agent", effUA) + if effHFToken != "" { + reqObj.Header.Set("Authorization", "Bearer "+effHFToken) + } + return reqObj, nil + } + + streamResp, err := DoWithFibonacciRetry(s.client, makeStreamReq, 5) + if err != nil { + return fmt.Errorf("upstream stream error: %w", err) + } + defer streamResp.Body.Close() + + completionID := "chatcmpl-" + GenerateUUID() + createdTime := time.Now().Unix() + + // Non-streaming completion + if !req.Stream { + reader := bufio.NewReader(streamResp.Body) + var currentEvent string + var finalRawText string + + for { + line, err := reader.ReadString('\n') + if err != nil { + break + } + line = strings.TrimSpace(line) + if line == "" { + continue + } + + if strings.HasPrefix(line, "event: ") { + currentEvent = strings.TrimPrefix(line, "event: ") + continue + } + + if strings.HasPrefix(line, "data: ") { + dataJSON := strings.TrimPrefix(line, "data: ") + if currentEvent == "error" { + return fmt.Errorf("gradio upstream error: %s", dataJSON) + } + if text, ok := parseAssistantText(dataJSON); ok { + finalRawText = text + } + } + } + + cleanedReasoning, cleanedContent := SeparateReasoningAndContent(finalRawText) + toolCalls, remContent, hasToolCalls := DetectToolCalls(cleanedContent) + + finishReason := "stop" + var msgContent interface{} = cleanedContent + + if hasToolCalls && len(toolCalls) > 0 { + finishReason = "tool_calls" + if remContent == "" { + msgContent = nil + } else { + msgContent = remContent + } + } + + WriteCompletionResponse(w, completionID, createdTime, resolvedModel, FinalOutput{ + Content: msgContent, + ReasoningContent: cleanedReasoning, + ToolCalls: toolCalls, + FinishReason: finishReason, + }) + return nil + } + + // Streaming completion + flusher, _ := w.(http.Flusher) + streamer := NewStreamer(w, flusher, completionID, createdTime, resolvedModel) + streamer.Role() + + reader := bufio.NewReader(streamResp.Body) + var currentEvent string + var emittedReasoning string + var emittedContent string + + toolFilter := &StreamToolCallFilter{} + + onContentChunk := func(text string) { + if text != "" { + streamer.Content(text) + } + } + + onToolCallChunk := func(tc ToolCall) { + streamer.ToolCallDelta(tc) + } + + for { + line, err := reader.ReadString('\n') + if err != nil { + break + } + line = strings.TrimSpace(line) + if line == "" { + continue + } + + if strings.HasPrefix(line, "event: ") { + currentEvent = strings.TrimPrefix(line, "event: ") + continue + } + + if strings.HasPrefix(line, "data: ") { + dataJSON := strings.TrimPrefix(line, "data: ") + if currentEvent == "error" { + break + } + + if fullAssistantText, ok := parseAssistantText(dataJSON); ok { + currentReasoning, currentContent := SeparateReasoningAndContent(fullAssistantText) + + // Stream reasoning tokens incrementally + if len(currentReasoning) > len(emittedReasoning) { + rDelta := currentReasoning[len(emittedReasoning):] + emittedReasoning = currentReasoning + streamer.Reasoning(rDelta) + } + + // Stream content tokens incrementally through tool filter + if len(currentContent) > len(emittedContent) { + cDelta := currentContent[len(emittedContent):] + emittedContent = currentContent + toolFilter.Feed(cDelta, onContentChunk, onToolCallChunk) + } + } + } + } + + // Flush remaining buffer in tool filter + toolFilter.Flush(onContentChunk, onToolCallChunk) + + finishReason := "stop" + if toolFilter.HasEmittedCalls() { + finishReason = "tool_calls" + } + streamer.Finish(finishReason) + streamer.Done() + + return nil +} + +// --------------------------------------------------------------------------- +// Main Server & Handlers +// --------------------------------------------------------------------------- + +func main() { + port := flag.Int("port", 8080, "Port to listen on") + endpoint := flag.String("endpoint", "https://halvo78-qwen3-8-flash-next-playground.hf.space", "Upstream HuggingFace Space URL") + defaultModel := flag.String("model", "Qwen/Qwen3.8-Flash-Next", "Default model ID") + thinking := flag.Bool("thinking", true, "Enable thinking/reasoning mode by default") + flag.BoolVar(thinking, "enable-thinking", true, "Alias for -thinking") + hfToken := flag.String("hf-token", "", "Optional HuggingFace Token for private/gated spaces") + apiKey := flag.String("api-key", "", "Optional upstream API key for BYOK inference") + baseURL := flag.String("base-url", "", "Optional upstream base URL for BYOK inference") + userAgent := flag.String("user-agent", "", "Custom User-Agent header") + flag.StringVar(userAgent, "ua", "", "Alias for -user-agent") + socksProxy := flag.String("socks", "", "SOCKS5 proxy URL (e.g. socks5://127.0.0.1:1080)") + flag.StringVar(socksProxy, "proxy", "", "Alias for -socks") + flag.StringVar(socksProxy, "socks5", "", "Alias for -socks") + + flag.Parse() + + if *userAgent != "" { + ConfiguredUserAgent = *userAgent + } + if *hfToken != "" { + ConfiguredToken = *hfToken + } + if *apiKey != "" { + ConfiguredAPIKey = *apiKey + } + if *baseURL != "" { + ConfiguredBaseURL = *baseURL + } + ConfiguredModel = *defaultModel + EnableThinkingDefault = *thinking + + proxyURL := *socksProxy + if proxyURL == "" { + for _, envKey := range []string{"ALL_PROXY", "all_proxy", "SOCKS5_PROXY", "socks5_proxy", "SOCKS_PROXY", "socks_proxy"} { + if v := os.Getenv(envKey); v != "" { + proxyURL = v + break + } + } + } + + svc := NewQwenService(*endpoint, *defaultModel, *hfToken, *apiKey, *baseURL, proxyURL, *thinking) + + mux := http.NewServeMux() + + handleModels := func(w http.ResponseWriter, r *http.Request) { + EnableCORS(w) + if r.Method == http.MethodOptions { + w.WriteHeader(http.StatusNoContent) + return + } + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + models := svc.ListModels() + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(ModelsResponse{ + Object: "list", + Data: models, + }) + } + + handleChatCompletions := func(w http.ResponseWriter, r *http.Request) { + EnableCORS(w) + if r.Method == http.MethodOptions { + w.WriteHeader(http.StatusNoContent) + return + } + if r.Method != http.MethodPost { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + var req ChatCompletionRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, fmt.Sprintf("Invalid JSON request: %v", err), http.StatusBadRequest) + return + } + + if len(req.Messages) == 0 { + http.Error(w, "messages array must not be empty", http.StatusBadRequest) + return + } + + if err := svc.Chat(w, r, req); err != nil { + log.Printf("Chat completion error: %v", err) + http.Error(w, fmt.Sprintf("Upstream gateway error: %v", err), http.StatusBadGateway) + return + } + } + + mux.HandleFunc("/models", handleModels) + mux.HandleFunc("/v1/models", handleModels) + mux.HandleFunc("/chat/completions", handleChatCompletions) + mux.HandleFunc("/v1/chat/completions", handleChatCompletions) + + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + EnableCORS(w) + if r.URL.Path == "/" || r.URL.Path == "/healthz" { + w.Header().Set("Content-Type", "application/json") + fmt.Fprintf(w, "{\"status\":\"ok\",\"service\":\"qflash\",\"model\":\"%s\"}\n", *defaultModel) + return + } + http.NotFound(w, r) + }) + + addr := fmt.Sprintf(":%d", *port) + log.Printf("Starting qflash gateway on %s -> %s", addr, *endpoint) + if proxyURL != "" { + log.Printf("Routing through SOCKS5 proxy: %s", proxyURL) + } + + server := &http.Server{ + Addr: addr, + Handler: mux, + } + + if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Fatalf("Server failed: %v", err) + } +} diff --git a/qflash b/qflash new file mode 100755 index 0000000..c7e25cd Binary files /dev/null and b/qflash differ diff --git a/qflash_test.go b/qflash_test.go new file mode 100644 index 0000000..7fbbb7e --- /dev/null +++ b/qflash_test.go @@ -0,0 +1,321 @@ +// qflash test suite +// Created by Luxferre in 2026, released into the public domain + +package main + +import ( + "bufio" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestChatMessageGetContentString(t *testing.T) { + // String content + msg1 := ChatMessage{Role: "user", Content: "Hello world"} + if msg1.GetContentString() != "Hello world" { + t.Fatalf("expected 'Hello world', got %q", msg1.GetContentString()) + } + + // Multi-part content + msg2 := ChatMessage{ + Role: "user", + Content: []interface{}{ + map[string]interface{}{"type": "text", "text": "Part 1 "}, + map[string]interface{}{"type": "text", "text": "Part 2"}, + }, + } + if msg2.GetContentString() != "Part 1 Part 2" { + t.Fatalf("expected 'Part 1 Part 2', got %q", msg2.GetContentString()) + } + + // Nil content + msg3 := ChatMessage{Role: "assistant", Content: nil} + if msg3.GetContentString() != "" { + t.Fatalf("expected empty string, got %q", msg3.GetContentString()) + } +} + +func TestSOCKS5Parsing(t *testing.T) { + cfg, err := ParseSOCKS5URL("socks5://user:pass@127.0.0.1:9050") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if cfg.Address != "127.0.0.1:9050" || cfg.Username != "user" || cfg.Password != "pass" { + t.Fatalf("mismatched parsed socks5 config: %+v", cfg) + } + + cfg2, err := ParseSOCKS5URL("socks5h://proxy.internal:1080") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if cfg2.Address != "proxy.internal:1080" || cfg2.Username != "" { + t.Fatalf("mismatched parsed socks5 config: %+v", cfg2) + } + + cfg3, err := ParseSOCKS5URL("10.0.0.5") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if cfg3.Address != "10.0.0.5:1080" { + t.Fatalf("expected default port 1080, got %s", cfg3.Address) + } +} + +func TestEffectiveModelID(t *testing.T) { + def := "Qwen/Qwen3.8-Flash-Next" + cases := map[string]string{ + "": def, + "qwen3.8-flash-next": def, + "qwen-flash-next": def, + "qwen-flash": def, + "qwen3.8-flash": def, + "Qwen/Qwen3.8-Flash-Next": def, + "custom-org/my-model": "custom-org/my-model", + } + + for in, exp := range cases { + res := EffectiveModelID(in, def) + if res != exp { + t.Errorf("EffectiveModelID(%q) = %q; expected %q", in, res, exp) + } + } +} + +func TestSeparateReasoningAndContentThinkTags(t *testing.T) { + raw := "\nAnalyzing prompt step by step.\n\n\nHere is the answer." + reasoning, content := SeparateReasoningAndContent(raw) + if reasoning != "Analyzing prompt step by step." { + t.Fatalf("unexpected reasoning: %q", reasoning) + } + if content != "Here is the answer." { + t.Fatalf("unexpected content: %q", content) + } +} + +func TestSeparateReasoningAndContentGradioFormat(t *testing.T) { + raw := "> 💭 **Thinking Process (QSA Micro-block Reasoning):**\n>\n> Thinking Process:\n>\n> 1. Step one\n> 2. Step two\n\n---\n\n### Answer Header\n\nDetailed answer here." + reasoning, content := SeparateReasoningAndContent(raw) + if !strings.Contains(reasoning, "1. Step one") || !strings.Contains(reasoning, "2. Step two") { + t.Fatalf("expected reasoning to contain steps, got: %q", reasoning) + } + if strings.Contains(reasoning, ">") { + t.Fatalf("expected blockquote markers to be stripped, got: %q", reasoning) + } + if content != "### Answer Header\n\nDetailed answer here." { + t.Fatalf("unexpected content: %q", content) + } +} + +func TestSeparateReasoningAndContentStreamingDivider(t *testing.T) { + raw := "> 💭 **Thinking Process (QSA Micro-block Reasoning):**\n>\n> Thinking Process:\n>\n> 1. Formulating response...\n\n---\n*Generating response...*" + reasoning, content := SeparateReasoningAndContent(raw) + if !strings.Contains(reasoning, "1. Formulating response...") { + t.Fatalf("expected reasoning, got: %q", reasoning) + } + if content != "" { + t.Fatalf("expected empty content during thought phase, got: %q", content) + } +} + +func TestToolCallParsingAndDetection(t *testing.T) { + rawJSON := `{"name": "get_weather", "arguments": {"city": "Tokyo"}}` + tc, ok := parseSingleToolCall(rawJSON) + if !ok { + t.Fatalf("expected successful single tool call parse") + } + if tc.Function.Name != "get_weather" { + t.Fatalf("expected 'get_weather', got %q", tc.Function.Name) + } + + rawXML := ` +{"name": "fetch_data", "arguments": "{\"id\": 42}"} +` + calls, rem, hasCalls := DetectToolCalls(rawXML) + if !hasCalls || len(calls) != 1 { + t.Fatalf("expected 1 detected tool call, got %d", len(calls)) + } + if calls[0].Function.Name != "fetch_data" { + t.Fatalf("expected 'fetch_data', got %q", calls[0].Function.Name) + } + if rem != "" { + t.Fatalf("expected empty remaining content, got %q", rem) + } +} + +func TestStreamToolCallFilterNoLeak(t *testing.T) { + filter := &StreamToolCallFilter{} + var streamedContent strings.Builder + var emittedCalls []ToolCall + + onContent := func(s string) { + streamedContent.WriteString(s) + } + onTool := func(tc ToolCall) { + emittedCalls = append(emittedCalls, tc) + } + + // Stream in small split chunks that split the tag + chunks := []string{ + "Here is the data: ", + "\n", + `{"name": "query_db", "arguments": {"sql": "SELECT 1"}}`, + "\n", + } + + for _, c := range chunks { + filter.Feed(c, onContent, onTool) + } + filter.Flush(onContent, onTool) + + if strings.Contains(streamedContent.String(), "") || strings.Contains(streamedContent.String(), "") { + t.Fatalf("tool call tags leaked into content: %q", streamedContent.String()) + } + if streamedContent.String() != "Here is the data: " { + t.Fatalf("unexpected content: %q", streamedContent.String()) + } + if len(emittedCalls) != 1 { + t.Fatalf("expected 1 emitted tool call, got %d", len(emittedCalls)) + } + if emittedCalls[0].Function.Name != "query_db" { + t.Fatalf("expected 'query_db', got %q", emittedCalls[0].Function.Name) + } +} + +func TestParseAssistantText(t *testing.T) { + dataJSON := `[[ + {"role": "user", "metadata": null, "content": [{"text": "hi", "type": "text"}], "options": null}, + {"role": "assistant", "metadata": null, "content": [{"text": "Hello, human!", "type": "text"}], "options": null} + ]]` + + text, ok := parseAssistantText(dataJSON) + if !ok { + t.Fatalf("expected successful parse of assistant text") + } + if text != "Hello, human!" { + t.Fatalf("expected 'Hello, human!', got %q", text) + } +} + +func TestQwenServiceChatMock(t *testing.T) { + // Mock upstream Gradio space server + mockServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/gradio_api/call/chat_response" && r.Method == http.MethodPost { + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"event_id": "test_event_123"}`)) + return + } + + if r.URL.Path == "/gradio_api/call/chat_response/test_event_123" && r.Method == http.MethodGet { + w.Header().Set("Content-Type", "text/event-stream") + flusher, _ := w.(http.Flusher) + + // Step 1: Thinking progress + chunk1 := `event: generating` + "\n" + + `data: [[{"role": "user", "content": [{"text": "Hello", "type": "text"}]}, {"role": "assistant", "content": [{"text": "> 💭 **Thinking Process:**\n>\n> Thinking Process:\n>\n> 1. Step 1\n\n---\n*Generating response...*", "type": "text"}]}]]` + "\n\n" + w.Write([]byte(chunk1)) + flusher.Flush() + + // Step 2: Final completion + chunk2 := `event: complete` + "\n" + + `data: [[{"role": "user", "content": [{"text": "Hello", "type": "text"}]}, {"role": "assistant", "content": [{"text": "> 💭 **Thinking Process:**\n>\n> Thinking Process:\n>\n> 1. Step 1\n\n---\n\nGreetings from mock Qwen!", "type": "text"}]}]]` + "\n\n" + w.Write([]byte(chunk2)) + flusher.Flush() + return + } + + http.NotFound(w, r) + })) + defer mockServer.Close() + + svc := NewQwenService(mockServer.URL, "Qwen/Qwen3.8-Flash-Next", "", "", "", "", true) + + // 1. Test Non-streaming completion + rec := httptest.NewRecorder() + req := ChatCompletionRequest{ + Model: "qwen3.8-flash-next", + Messages: []ChatMessage{ + {Role: "user", Content: "Hello"}, + }, + Stream: false, + } + + err := svc.Chat(rec, nil, req) + if err != nil { + t.Fatalf("unexpected error in Chat 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 completion response: %v", err) + } + + if len(resp.Choices) == 0 { + t.Fatalf("expected choices, got 0") + } + if resp.Choices[0].Message.Content != "Greetings from mock Qwen!" { + t.Fatalf("unexpected message content: %v", resp.Choices[0].Message.Content) + } + if !strings.Contains(resp.Choices[0].Message.ReasoningContent, "1. Step 1") { + t.Fatalf("unexpected reasoning content: %v", resp.Choices[0].Message.ReasoningContent) + } + + // 2. Test Streaming completion + recStream := httptest.NewRecorder() + reqStream := ChatCompletionRequest{ + Model: "qwen-flash", + Messages: []ChatMessage{ + {Role: "user", Content: "Hello"}, + }, + Stream: true, + } + + errStream := svc.Chat(recStream, nil, reqStream) + if errStream != nil { + t.Fatalf("unexpected error in Chat streaming: %v", errStream) + } + + scanner := bufio.NewScanner(recStream.Body) + var receivedReasoning strings.Builder + var receivedContent strings.Builder + var sawDone bool + + for scanner.Scan() { + line := scanner.Text() + if strings.HasPrefix(line, "data: ") { + payload := strings.TrimPrefix(line, "data: ") + if payload == "[DONE]" { + sawDone = true + continue + } + var sResp StreamResponse + if err := json.Unmarshal([]byte(payload), &sResp); err == nil && len(sResp.Choices) > 0 { + delta := sResp.Choices[0].Delta + if delta.ReasoningContent != "" { + receivedReasoning.WriteString(delta.ReasoningContent) + } + if delta.Content != "" { + receivedContent.WriteString(delta.Content) + } + } + } + } + + if !sawDone { + t.Fatalf("expected [DONE] chunk in stream") + } + if !strings.Contains(receivedReasoning.String(), "1. Step 1") { + t.Fatalf("expected streamed reasoning, got %q", receivedReasoning.String()) + } + if !strings.Contains(receivedContent.String(), "Greetings from mock Qwen!") { + t.Fatalf("expected streamed content, got %q", receivedContent.String()) + } +}