From d68efdbf5980dc8ae56af540e8e25a386480dc19 Mon Sep 17 00:00:00 2001 From: Luxferre Date: Thu, 27 Aug 2026 13:54:10 +0300 Subject: [PATCH] feat: initial release of q38max gateway --- .gitignore | 2 + Makefile | 13 + README.md | 184 ++++++ go.mod | 3 + main.go | 1640 ++++++++++++++++++++++++++++++++++++++++++++++++ q38max_test.go | 119 ++++ xtest.sh | 67 ++ 7 files changed, 2028 insertions(+) create mode 100644 .gitignore create mode 100644 Makefile create mode 100644 README.md create mode 100644 go.mod create mode 100644 main.go create mode 100644 q38max_test.go create mode 100755 xtest.sh diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d7d5d82 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +bin/ +*.log diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..fd27039 --- /dev/null +++ b/Makefile @@ -0,0 +1,13 @@ +all: q38max + +q38max: + mkdir -p bin + go build -trimpath -ldflags="-s -w" -o bin/q38max main.go + +test: + go test -v ./... + +clean: + rm -rf bin + +.PHONY: all q38max test clean diff --git a/README.md b/README.md new file mode 100644 index 0000000..9d8d1e9 --- /dev/null +++ b/README.md @@ -0,0 +1,184 @@ +# q38max + +Standalone, zero-dependency OpenAI-compatible proxy gateway in Go for the **Qwen 3.8 Max** model (`Qwen/Qwen3.8-Max`) hosted on Hugging Face Spaces (`harpreetsahota-qwen38-max-openlogo-demo.hf.space`). + +## Overview + +`q38max` reverse-engineers the FiftyOne plugin backend operator interface of the Hugging Face space and transforms it into a standard, production-ready OpenAI API endpoint (`/v1/chat/completions` and `/v1/models`). + +### Features + +- **OpenAI Standard Compatibility**: Full drop-in replacement for OpenAI API clients (Curl, Python `openai`, LangChain, LiteLLM, Open-WebUI). +- **Zero External Dependencies**: Pure standard library Go implementation (`net/http`, `encoding/json`, `crypto/rand`, `time`). +- **Live Streaming SSE & Reasoning**: Streams real-time tokens with separation of reasoning content (`delta.reasoning_content`) and message content (`delta.content`). +- **Function / Tool Calling Interception**: Supports OpenAI `tools` specification, system prompt tool schema injection, and stateful streaming interception of tool calls (`delta.tool_calls` and `finish_reason: "tool_calls"`). +- **Session Lifecycle Management**: Thread-safe automatic session creation (`/__session/start`), periodic background heartbeats (`/__session/heartbeat`), and auto-reconnect recovery. +- **Fibonacci Backoff Retry**: Resilient against network hiccups and transient timeouts. + +--- + +## Architecture & Upstream Protocol + +``` ++---------------------------+ OpenAI HTTP / SSE +------------------------+ +| Client (Python / Curl / | ===========================> | q38max Gateway | +| OpenAI SDK / Open-WebUI) | | (localhost:8080) | ++---------------------------+ +------------------------+ + | + | FiftyOne Session & + | Operator API + v + +------------------------+ + | HuggingFace Space | + | FiftyOne Backend | + | (Qwen 3.8 Max Model) | + +------------------------+ +``` + +### Upstream Flow: +1. `POST /__session/start` -> Allocates an ephemeral session token `X-FiftyOne-Session` and dataset clone. +2. `POST /operators/execute` -> Dispatches the `@harpreetsahota/qwen38-max/qwen38_chat` operator with method `"ask"`. +3. Polling Loops: + - `get_thinking_chunk`: Extracts newly generated reasoning tokens in real-time. + - `get_stream_chunk`: Extracts newly generated message content tokens in real-time. + +--- + +## Build & Run + +### Build +```bash +make q38max +``` + +Binary is output to `bin/q38max`. + +### Run +```bash +./bin/q38max -port 8080 +``` + +### CLI Flags + +| Flag | Default | Description | +|------|---------|-------------| +| `-port` | `8080` | Port to listen on | +| `-space-url` | `https://harpreetsahota-qwen38-max-openlogo-demo.hf.space` | Upstream Hugging Face Space URL | +| `-sample-path` | `/home/user/datasets/openlogo/data/data_0/logos32plus_002359.jpg` | Container image sample path | +| `-model` | `qwen-3.8-max` | Default model identifier | +| `-timeout` | `300` | Upstream timeout in seconds | +| `-user-agent` / `-ua` | `""` | Custom User-Agent header | +| `-hf-token` | `""` | Optional Hugging Face token | + +--- + +## API Usage Examples + +### 1. List Models +```bash +curl http://localhost:8080/v1/models +``` + +### 2. Non-Streaming Chat Completion +```bash +curl -X POST http://localhost:8080/v1/chat/completions \ + -H "Content-Type: application/json" \ + -d '{ + "model": "qwen-3.8-max", + "messages": [ + {"role": "user", "content": "What is the capital of Germany? Answer in 1 word."} + ], + "reasoning_effort": "none", + "max_tokens": 50 + }' +``` + +### 3. Streaming Chat Completion with Reasoning +```bash +curl -N -X POST http://localhost:8080/v1/chat/completions \ + -H "Content-Type: application/json" \ + -d '{ + "model": "qwen-3.8-max", + "messages": [ + {"role": "user", "content": "Calculate 25 * 25 and explain in one sentence."} + ], + "stream": true, + "reasoning_effort": "medium", + "max_tokens": 150 + }' +``` + +### 4. Function / Tool Calling +```bash +curl -X POST http://localhost:8080/v1/chat/completions \ + -H "Content-Type: application/json" \ + -d '{ + "model": "qwen-3.8-max", + "messages": [ + {"role": "user", "content": "What is the weather in Berlin?"} + ], + "tools": [ + { + "type": "function", + "function": { + "name": "get_weather", + "description": "Get current weather for a city", + "parameters": { + "type": "object", + "properties": { + "location": {"type": "string"} + }, + "required": ["location"] + } + } + } + ], + "reasoning_effort": "none" + }' +``` + +### 5. Python OpenAI Client Example +```python +from openai import OpenAI + +client = OpenAI( + base_url="http://localhost:8080/v1", + api_key="none" +) + +response = client.chat.completions.create( + model="qwen-3.8-max", + messages=[ + {"role": "user", "content": "Write a short haiku about computers."} + ], + stream=True +) + +for chunk in response: + delta = chunk.choices[0].delta + if hasattr(delta, "reasoning_content") and delta.reasoning_content: + print(f"[Thinking] {delta.reasoning_content}", end="", flush=True) + if delta.content: + print(delta.content, end="", flush=True) +print() +``` + +--- + +## Testing + +Run unit tests: +```bash +make test +``` + +Run end-to-end integration tests: +```bash +./xtest.sh 8080 +``` + +--- + +## License + +Public Domain / Unlicense diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..00f0b9a --- /dev/null +++ b/go.mod @@ -0,0 +1,3 @@ +module q38max + +go 1.22 diff --git a/main.go b/main.go new file mode 100644 index 0000000..7898705 --- /dev/null +++ b/main.go @@ -0,0 +1,1640 @@ +// q38max: Standalone OpenAI-compatible gateway for Qwen 3.8 Max HuggingFace Spaces +// Created by Luxferre in 2026, released into the public domain + +package main + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/json" + "flag" + "fmt" + "io" + "log" + "net/http" + "os" + "regexp" + "strings" + "sync" + "time" +) + +var ( + DefaultSpaceURL = "https://harpreetsahota-qwen38-max-openlogo-demo.hf.space" + DefaultSamplePath = "/home/user/datasets/openlogo/data/data_0/logos32plus_002359.jpg" + DefaultModelID = "qwen-3.8-max" + DefaultUserAgent = "Mozilla/5.0 (X11; Linux x86_64; rv:153.0) Gecko/20100101 Firefox/153.0" + ConfiguredUserAgent string + ConfiguredToken string +) + +// --------------------------------------------------------------------------- +// 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"` + Thinking interface{} `json:"thinking,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"` +} + +// --------------------------------------------------------------------------- +// 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 || resp.StatusCode == http.StatusCreated || resp.StatusCode == http.StatusNoContent) { + 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") +} + +func ResolveMaxTokens(req ChatCompletionRequest) int { + mt := req.MaxTokens + if mt == 0 && req.MaxCompletionTokens > 0 { + mt = req.MaxCompletionTokens + } + if mt <= 0 { + mt = 4000 + } + 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 EffectiveToken(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 ") { + token := strings.TrimSpace(auth[7:]) + if token != "" && token != "-" { + return token + } + } + } + } + if ConfiguredToken != "" { + return ConfiguredToken + } + if envTok := os.Getenv("HF_TOKEN"); envTok != "" { + return envTok + } + return "" +} + +// --------------------------------------------------------------------------- +// FiftyOne HF Space Session Manager +// --------------------------------------------------------------------------- + +type FiftyOneSessionManager struct { + spaceURL string + client *http.Client + mu sync.Mutex + token string + datasetName string + lastSeen time.Time +} + +func NewFiftyOneSessionManager(spaceURL string, client *http.Client) *FiftyOneSessionManager { + sm := &FiftyOneSessionManager{ + spaceURL: strings.TrimRight(spaceURL, "/"), + client: client, + } + go sm.heartbeatLoop() + return sm +} + +func (sm *FiftyOneSessionManager) EnsureSession(ctx context.Context, reqUserAgent string, hfToken string) (string, string, error) { + sm.mu.Lock() + defer sm.mu.Unlock() + + if sm.token != "" && sm.datasetName != "" && time.Since(sm.lastSeen) < 20*time.Minute { + sm.lastSeen = time.Now() + return sm.token, sm.datasetName, nil + } + + return sm.startNewSessionLocked(ctx, reqUserAgent, hfToken) +} + +func (sm *FiftyOneSessionManager) startNewSessionLocked(ctx context.Context, reqUserAgent string, hfToken string) (string, string, error) { + url := sm.spaceURL + "/__session/start" + req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader([]byte("{}"))) + if err != nil { + return "", "", err + } + req.Header.Set("Content-Type", "application/json") + if reqUserAgent != "" { + req.Header.Set("User-Agent", reqUserAgent) + } else { + req.Header.Set("User-Agent", DefaultUserAgent) + } + if hfToken != "" { + req.Header.Set("Authorization", "Bearer "+hfToken) + } + + resp, err := sm.client.Do(req) + if err != nil { + return "", "", fmt.Errorf("failed to connect to session start: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + bodyBytes, _ := io.ReadAll(resp.Body) + return "", "", fmt.Errorf("session start returned status %d: %s", resp.StatusCode, string(bodyBytes)) + } + + var startRes struct { + Token string `json:"token"` + URL string `json:"url"` + Error string `json:"error"` + } + if err := json.NewDecoder(resp.Body).Decode(&startRes); err != nil { + return "", "", fmt.Errorf("failed to decode session start json: %w", err) + } + + if startRes.Error != "" { + return "", "", fmt.Errorf("session start error: %s", startRes.Error) + } + if startRes.Token == "" { + return "", "", fmt.Errorf("session start returned empty token") + } + + parts := strings.Split(strings.TrimRight(startRes.URL, "/"), "/") + datasetName := parts[len(parts)-1] + if datasetName == "" { + datasetName = "openlogo-session-" + startRes.Token + } + + sm.token = startRes.Token + sm.datasetName = datasetName + sm.lastSeen = time.Now() + + log.Printf("[Session] Allocated new session token=%s dataset=%s", sm.token, sm.datasetName) + return sm.token, sm.datasetName, nil +} + +func (sm *FiftyOneSessionManager) Invalidate() { + sm.mu.Lock() + defer sm.mu.Unlock() + sm.token = "" + sm.datasetName = "" +} + +func (sm *FiftyOneSessionManager) heartbeatLoop() { + ticker := time.NewTicker(45 * time.Second) + defer ticker.Stop() + + for range ticker.C { + sm.mu.Lock() + token := sm.token + dataset := sm.datasetName + sm.mu.Unlock() + + if token == "" || dataset == "" { + continue + } + + url := fmt.Sprintf("%s/__session/heartbeat?dataset=%s", sm.spaceURL, dataset) + req, err := http.NewRequest("POST", url, nil) + if err == nil { + req.Header.Set("X-FiftyOne-Session", token) + req.Header.Set("User-Agent", DefaultUserAgent) + if ConfiguredToken != "" { + req.Header.Set("Authorization", "Bearer "+ConfiguredToken) + } + resp, err := sm.client.Do(req) + if err == nil { + resp.Body.Close() + if resp.StatusCode == http.StatusUnauthorized { + sm.Invalidate() + } + } + } + } +} + +// --------------------------------------------------------------------------- +// Response Framing +// --------------------------------------------------------------------------- + +type FinalOutput struct { + Content interface{} + ReasoningContent string + ToolCalls []ToolCall + FinishReason string + PromptTokens int + CompletionTokens int +} + +func WriteCompletionResponse(w http.ResponseWriter, completionID string, created int64, model string, out FinalOutput) { + finish := out.FinishReason + if finish == "" { + finish = "stop" + } + totalTokens := out.PromptTokens + out.CompletionTokens + 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: out.PromptTokens, + CompletionTokens: out.CompletionTokens, + TotalTokens: totalTokens, + }, + } + 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]) + } + } + + 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]) + } + } + + 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) (calls []ToolCall, remainingText string) { + trimmed := strings.TrimSpace(content) + if trimmed == "" { + return nil, "" + } + + if strings.Contains(content, "") { + blocks, rem := ExtractToolCallBlocks(content) + for _, blk := range blocks { + if tc, ok := parseXMLToolCall(blk); ok { + calls = append(calls, tc) + } + } + if len(calls) > 0 { + return calls, rem + } + } + + var rawArray []interface{} + cleanedContent := cleanJSONBlock(trimmed) + if strings.HasPrefix(cleanedContent, "[") && strings.HasSuffix(cleanedContent, "]") { + if err := json.Unmarshal([]byte(cleanedContent), &rawArray); err == nil && len(rawArray) > 0 { + allValid := true + var tempCalls []ToolCall + for _, item := range rawArray { + b, _ := json.Marshal(item) + if tc, ok := parseSingleToolCall(string(b)); ok { + tempCalls = append(tempCalls, tc) + } else { + allValid = false + break + } + } + if allValid && len(tempCalls) > 0 { + return tempCalls, "" + } + } + } + + if tc, ok := parseSingleToolCall(trimmed); ok { + return []ToolCall{tc}, "" + } + + return nil, content +} + +// --------------------------------------------------------------------------- +// Stateful Streaming Tool Call Filter +// --------------------------------------------------------------------------- + +type StreamToolCallFilter struct { + streamer *Streamer + buffer string + insideToolCall bool + insideJSON bool + hasEmittedTool bool +} + +func NewStreamToolCallFilter(streamer *Streamer) *StreamToolCallFilter { + return &StreamToolCallFilter{ + streamer: streamer, + } +} + +func (f *StreamToolCallFilter) Feed(chunk string) { + f.buffer += chunk + + for { + if !f.insideToolCall && !f.insideJSON { + idx := strings.Index(f.buffer, "") + if idx != -1 { + cleanPrefix := f.buffer[:idx] + if cleanPrefix != "" { + f.streamer.Content(cleanPrefix) + } + f.buffer = f.buffer[idx:] + f.insideToolCall = true + continue + } + + trimmedBuf := strings.TrimSpace(f.buffer) + if (strings.HasPrefix(trimmedBuf, "{\"name\"") || strings.HasPrefix(trimmedBuf, "{\"function\"")) && strings.HasSuffix(trimmedBuf, "}") { + if tc, ok := parseSingleToolCall(trimmedBuf); ok { + f.hasEmittedTool = true + f.streamer.ToolCallDelta(tc) + f.buffer = "" + return + } + } + + const safeLen = 12 + if len(f.buffer) > safeLen { + emitLen := len(f.buffer) - safeLen + f.streamer.Content(f.buffer[:emitLen]) + f.buffer = f.buffer[emitLen:] + } + return + } + + if f.insideToolCall { + eIdx := strings.Index(f.buffer, "") + if eIdx != -1 { + block := f.buffer[:eIdx+len("")] + f.buffer = f.buffer[eIdx+len(""):] + f.insideToolCall = false + + if tc, ok := parseXMLToolCall(block); ok { + f.hasEmittedTool = true + f.streamer.ToolCallDelta(tc) + } + continue + } + return + } + + return + } +} + +func (f *StreamToolCallFilter) Flush() { + if f.buffer == "" { + return + } + if f.insideToolCall { + if tc, ok := parseXMLToolCall(f.buffer); ok { + f.hasEmittedTool = true + f.streamer.ToolCallDelta(tc) + f.buffer = "" + return + } + } + + if tc, ok := parseSingleToolCall(f.buffer); ok { + f.hasEmittedTool = true + f.streamer.ToolCallDelta(tc) + f.buffer = "" + return + } + + f.streamer.Content(f.buffer) + f.buffer = "" +} + +func (f *StreamToolCallFilter) HasEmittedTools() bool { + return f.hasEmittedTool +} + +// --------------------------------------------------------------------------- +// Thinking Filter +// --------------------------------------------------------------------------- + +var thinkTagRegex = regexp.MustCompile(`(?s)(.*?)(?:|$)`) + +func ExtractThinkingContent(text string) (thinking string, cleanText string) { + matches := thinkTagRegex.FindAllStringSubmatch(text, -1) + if len(matches) > 0 { + var thinkParts []string + for _, m := range matches { + if len(m) > 1 { + thinkParts = append(thinkParts, strings.TrimSpace(m[1])) + } + } + thinking = strings.Join(thinkParts, "\n\n") + cleanText = thinkTagRegex.ReplaceAllString(text, "") + cleanText = strings.TrimSpace(cleanText) + return thinking, cleanText + } + return "", text +} + +// --------------------------------------------------------------------------- +// Conversation Message Formatting +// --------------------------------------------------------------------------- + +func formatToolPrompt(tools []Tool) string { + if len(tools) == 0 { + return "" + } + var sb strings.Builder + sb.WriteString("\n\n[AVAILABLE TOOLS]\n") + sb.WriteString("You have access to the following tools/functions:\n") + for _, t := range tools { + b, err := json.Marshal(t) + if err == nil { + sb.WriteString(string(b)) + sb.WriteString("\n") + } + } + sb.WriteString("\nTo invoke a tool, output a single tool call formatted exactly like this:\n") + sb.WriteString("{\"name\": \"function_name\", \"arguments\": {\"param1\": \"val1\"}}\n") + sb.WriteString("Do not add extra explanation when calling a tool.\n\n") + return sb.String() +} + +func PrepareConversation(req ChatCompletionRequest) (history []map[string]interface{}, currentQuestion string, thinkingMode string) { + var systemParts []string + + // Add default baseline directive to maintain conversational focus + systemParts = append(systemParts, "Respond directly and concisely to the user's instructions and queries. Do not mention background image dimensions or unrelated demo metadata unless specifically asked.") + + if len(req.Tools) > 0 { + systemParts = append(systemParts, formatToolPrompt(req.Tools)) + } + + // Determine thinking mode + thinkingMode = "auto" + if req.Thinking != nil { + switch tv := req.Thinking.(type) { + case bool: + if tv { + thinkingMode = "true" + } else { + thinkingMode = "false" + } + case string: + s := strings.ToLower(strings.TrimSpace(tv)) + if s == "true" || s == "on" || s == "enabled" { + thinkingMode = "true" + } else if s == "false" || s == "off" || s == "disabled" { + thinkingMode = "false" + } + } + } else if req.ReasoningEffort != "" { + re := strings.ToLower(strings.TrimSpace(req.ReasoningEffort)) + if re == "none" || re == "false" || re == "off" { + thinkingMode = "false" + } else if re == "low" || re == "medium" || re == "high" || re == "true" { + thinkingMode = "true" + } + } + + var turns []map[string]string + for _, m := range req.Messages { + role := strings.ToLower(strings.TrimSpace(m.Role)) + content := m.GetContentString() + + if role == "system" { + if content != "" { + systemParts = append(systemParts, content) + } + continue + } + + if role == "tool" || role == "function" { + toolName := m.Name + if toolName == "" { + toolName = "tool" + } + content = fmt.Sprintf("{\"name\": %q, \"content\": %s}", toolName, content) + role = "user" + } + + if len(m.ToolCalls) > 0 { + var tcParts []string + for _, tc := range m.ToolCalls { + tcParts = append(tcParts, fmt.Sprintf("{\"name\": %q, \"arguments\": %s}", tc.Function.Name, tc.Function.Arguments)) + } + if content != "" { + content = content + "\n" + strings.Join(tcParts, "\n") + } else { + content = strings.Join(tcParts, "\n") + } + } + + turns = append(turns, map[string]string{ + "role": role, + "content": content, + }) + } + + systemText := strings.TrimSpace(strings.Join(systemParts, "\n\n")) + + if len(turns) == 0 { + emptyHist := make([]map[string]interface{}, 0) + if systemText != "" { + return emptyHist, systemText, thinkingMode + } + return emptyHist, "Hello", thinkingMode + } + + lastTurn := turns[len(turns)-1] + priorTurns := turns[:len(turns)-1] + + historyItems := make([]map[string]interface{}, 0) + for idx, t := range priorTurns { + c := t["content"] + if idx == 0 && systemText != "" { + c = "[SYSTEM INSTRUCTION]\n" + systemText + "\n\n" + c + } + historyItems = append(historyItems, map[string]interface{}{ + "role": t["role"], + "content": c, + }) + } + + currentQuestion = lastTurn["content"] + if len(priorTurns) == 0 && systemText != "" { + currentQuestion = "[SYSTEM INSTRUCTION]\n" + systemText + "\n\n" + currentQuestion + } + + return historyItems, currentQuestion, thinkingMode +} + +// --------------------------------------------------------------------------- +// Main Gateway Service +// --------------------------------------------------------------------------- + +type Q38Gateway struct { + spaceURL string + samplePath string + modelID string + client *http.Client + sessionMgr *FiftyOneSessionManager +} + +func NewQ38Gateway(spaceURL string, samplePath string, modelID string, timeout time.Duration) *Q38Gateway { + tr := &http.Transport{ + MaxIdleConns: 100, + MaxIdleConnsPerHost: 20, + IdleConnTimeout: 90 * time.Second, + } + client := &http.Client{ + Transport: tr, + Timeout: timeout, + } + return &Q38Gateway{ + spaceURL: strings.TrimRight(spaceURL, "/"), + samplePath: samplePath, + modelID: modelID, + client: client, + sessionMgr: NewFiftyOneSessionManager(spaceURL, client), + } +} + +func (gw *Q38Gateway) HandleModels(w http.ResponseWriter, r *http.Request) { + EnableCORS(w) + if r.Method == http.MethodOptions { + w.WriteHeader(http.StatusNoContent) + return + } + + now := time.Now().Unix() + resp := ModelsResponse{ + Object: "list", + Data: []ModelItem{ + { + ID: gw.modelID, + Object: "model", + Created: now, + OwnedBy: "qwen", + }, + { + ID: "qwen3.8-max", + Object: "model", + Created: now, + OwnedBy: "qwen", + }, + { + ID: "qwen38-max", + Object: "model", + Created: now, + OwnedBy: "qwen", + }, + { + ID: "Qwen/Qwen3.8-Max", + Object: "model", + Created: now, + OwnedBy: "qwen", + }, + }, + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(resp) +} + +func (gw *Q38Gateway) HandleChatCompletions(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 + bodyBytes, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, "Failed to read request body", http.StatusBadRequest) + return + } + + if err := json.Unmarshal(bodyBytes, &req); err != nil { + http.Error(w, fmt.Sprintf("Invalid JSON request: %v", err), http.StatusBadRequest) + return + } + + modelName := req.Model + if modelName == "" { + modelName = gw.modelID + } + + maxTokens := ResolveMaxTokens(req) + history, question, thinkingMode := PrepareConversation(req) + if history == nil { + history = make([]map[string]interface{}, 0) + } + + reqUA := EffectiveUserAgent(r) + reqHF := EffectiveToken(r) + + // Obtain or refresh FiftyOne Session + token, datasetName, err := gw.sessionMgr.EnsureSession(r.Context(), reqUA, reqHF) + if err != nil { + log.Printf("[Error] Failed to obtain FiftyOne session: %v", err) + http.Error(w, fmt.Sprintf("Upstream session error: %v", err), http.StatusBadGateway) + return + } + + // Execute ask operator + askPayload := map[string]interface{}{ + "operator_uri": "@harpreetsahota/qwen38-max/qwen38_chat", + "dataset_name": datasetName, + "view": []interface{}{}, + "selected": []interface{}{}, + "selected_samples": []interface{}{}, + "selected_labels": []interface{}{}, + "filters": map[string]interface{}{}, + "extended": map[string]interface{}{}, + "params": map[string]interface{}{ + "panel_id": "qwen38_chat", + "__method__": "ask", + "filepath": gw.samplePath, + "question": question, + "history": history, + "thinking": thinkingMode, + "hint_format": "auto", + "hint_text": "", + "image_max_side": 1280, + "max_tokens": maxTokens, + }, + } + + askBody, _ := json.Marshal(askPayload) + executeURL := gw.spaceURL + "/operators/execute" + + makeAskReq := func() (*http.Request, error) { + hReq, err := http.NewRequestWithContext(r.Context(), "POST", executeURL, bytes.NewReader(askBody)) + if err != nil { + return nil, err + } + hReq.Header.Set("Content-Type", "application/json") + hReq.Header.Set("X-FiftyOne-Session", token) + hReq.Header.Set("User-Agent", reqUA) + if reqHF != "" { + hReq.Header.Set("Authorization", "Bearer "+reqHF) + } + return hReq, nil + } + + askResp, err := DoWithFibonacciRetry(gw.client, makeAskReq, 3) + if err != nil { + log.Printf("[Error] ask execution failed: %v", err) + gw.sessionMgr.Invalidate() + http.Error(w, fmt.Sprintf("Upstream operator execution failed: %v", err), http.StatusBadGateway) + return + } + defer askResp.Body.Close() + + var askResult struct { + Result struct { + Status string `json:"status"` + RunID string `json:"run_id"` + Error string `json:"error"` + } `json:"result"` + Error string `json:"error"` + ErrorMessage string `json:"error_message"` + } + + if err := json.NewDecoder(askResp.Body).Decode(&askResult); err != nil { + http.Error(w, fmt.Sprintf("Failed to parse operator response: %v", err), http.StatusBadGateway) + return + } + + if askResult.Error != "" || askResult.ErrorMessage != "" { + errMsg := askResult.Error + if errMsg == "" { + errMsg = askResult.ErrorMessage + } + http.Error(w, fmt.Sprintf("Upstream error: %s", errMsg), http.StatusBadGateway) + return + } + + runID := askResult.Result.RunID + if runID == "" { + http.Error(w, "Upstream did not return a run_id", http.StatusBadGateway) + return + } + + log.Printf("[Inference] Started run_id=%s model=%s stream=%t thinking=%s", runID, modelName, req.Stream, thinkingMode) + + completionID := "chatcmpl-" + GenerateUUID() + createdTime := time.Now().Unix() + + if req.Stream { + flusher, ok := w.(http.Flusher) + if !ok { + http.Error(w, "Streaming unsupported", http.StatusInternalServerError) + return + } + + streamer := NewStreamer(w, flusher, completionID, createdTime, modelName) + streamer.Role() + + toolFilter := NewStreamToolCallFilter(streamer) + + streamCursor := 0 + thinkingCursor := 0 + pollTicker := time.NewTicker(150 * time.Millisecond) + defer pollTicker.Stop() + + for { + select { + case <-r.Context().Done(): + log.Printf("[Stream] Client disconnected for run_id=%s", runID) + return + case <-pollTicker.C: + // 1. Poll thinking chunk + thinkPayload, _ := json.Marshal(map[string]interface{}{ + "operator_uri": "@harpreetsahota/qwen38-max/qwen38_chat", + "dataset_name": datasetName, + "view": []interface{}{}, + "selected": []interface{}{}, + "selected_samples": []interface{}{}, + "selected_labels": []interface{}{}, + "filters": map[string]interface{}{}, + "extended": map[string]interface{}{}, + "params": map[string]interface{}{ + "panel_id": "qwen38_chat", + "__method__": "get_thinking_chunk", + "run_id": runID, + "cursor": thinkingCursor, + }, + }) + tReq, err := http.NewRequestWithContext(r.Context(), "POST", executeURL, bytes.NewReader(thinkPayload)) + if err == nil { + tReq.Header.Set("Content-Type", "application/json") + tReq.Header.Set("X-FiftyOne-Session", token) + tReq.Header.Set("User-Agent", reqUA) + if reqHF != "" { + tReq.Header.Set("Authorization", "Bearer "+reqHF) + } + tResp, tErr := gw.client.Do(tReq) + if tErr == nil { + var tResult struct { + Result struct { + Text string `json:"text"` + Cursor int `json:"cursor"` + Done bool `json:"done"` + } `json:"result"` + } + if json.NewDecoder(tResp.Body).Decode(&tResult) == nil { + if tResult.Result.Text != "" { + streamer.Reasoning(tResult.Result.Text) + thinkingCursor = tResult.Result.Cursor + } + } + tResp.Body.Close() + } + } + + // 2. Poll stream chunk + streamPayload, _ := json.Marshal(map[string]interface{}{ + "operator_uri": "@harpreetsahota/qwen38-max/qwen38_chat", + "dataset_name": datasetName, + "view": []interface{}{}, + "selected": []interface{}{}, + "selected_samples": []interface{}{}, + "selected_labels": []interface{}{}, + "filters": map[string]interface{}{}, + "extended": map[string]interface{}{}, + "params": map[string]interface{}{ + "panel_id": "qwen38_chat", + "__method__": "get_stream_chunk", + "run_id": runID, + "cursor": streamCursor, + }, + }) + sReq, err := http.NewRequestWithContext(r.Context(), "POST", executeURL, bytes.NewReader(streamPayload)) + if err == nil { + sReq.Header.Set("Content-Type", "application/json") + sReq.Header.Set("X-FiftyOne-Session", token) + sReq.Header.Set("User-Agent", reqUA) + if reqHF != "" { + sReq.Header.Set("Authorization", "Bearer "+reqHF) + } + sResp, sErr := gw.client.Do(sReq) + if sErr == nil { + var sResult struct { + Result struct { + Text string `json:"text"` + Cursor int `json:"cursor"` + Done bool `json:"done"` + FinalStatus struct { + Status string `json:"status"` + PromptTokens int `json:"prompt_tokens"` + CompletionTokens int `json:"completion_tokens"` + Error string `json:"error"` + } `json:"final_status"` + } `json:"result"` + } + if json.NewDecoder(sResp.Body).Decode(&sResult) == nil { + if sResult.Result.Text != "" { + toolFilter.Feed(sResult.Result.Text) + streamCursor = sResult.Result.Cursor + } + if sResult.Result.Done { + sResp.Body.Close() + toolFilter.Flush() + + finishReason := "stop" + if toolFilter.HasEmittedTools() { + finishReason = "tool_calls" + } + streamer.Finish(finishReason) + streamer.Done() + log.Printf("[Stream] Completed run_id=%s finish_reason=%s", runID, finishReason) + return + } + } + sResp.Body.Close() + } + } + } + } + } else { + // Non-streaming mode: poll until done + var accumulatedContent strings.Builder + var accumulatedThinking strings.Builder + promptTokens := 0 + completionTokens := 0 + + streamCursor := 0 + thinkingCursor := 0 + pollTicker := time.NewTicker(200 * time.Millisecond) + defer pollTicker.Stop() + + for { + select { + case <-r.Context().Done(): + log.Printf("[Non-Stream] Client canceled run_id=%s", runID) + return + case <-pollTicker.C: + // Poll thinking chunk + thinkPayload, _ := json.Marshal(map[string]interface{}{ + "operator_uri": "@harpreetsahota/qwen38-max/qwen38_chat", + "dataset_name": datasetName, + "view": []interface{}{}, + "selected": []interface{}{}, + "selected_samples": []interface{}{}, + "selected_labels": []interface{}{}, + "filters": map[string]interface{}{}, + "extended": map[string]interface{}{}, + "params": map[string]interface{}{ + "panel_id": "qwen38_chat", + "__method__": "get_thinking_chunk", + "run_id": runID, + "cursor": thinkingCursor, + }, + }) + tReq, err := http.NewRequestWithContext(r.Context(), "POST", executeURL, bytes.NewReader(thinkPayload)) + if err == nil { + tReq.Header.Set("Content-Type", "application/json") + tReq.Header.Set("X-FiftyOne-Session", token) + tReq.Header.Set("User-Agent", reqUA) + if reqHF != "" { + tReq.Header.Set("Authorization", "Bearer "+reqHF) + } + tResp, tErr := gw.client.Do(tReq) + if tErr == nil { + var tResult struct { + Result struct { + Text string `json:"text"` + Cursor int `json:"cursor"` + Done bool `json:"done"` + } `json:"result"` + } + if json.NewDecoder(tResp.Body).Decode(&tResult) == nil { + if tResult.Result.Text != "" { + accumulatedThinking.WriteString(tResult.Result.Text) + thinkingCursor = tResult.Result.Cursor + } + } + tResp.Body.Close() + } + } + + // Poll stream chunk + streamPayload, _ := json.Marshal(map[string]interface{}{ + "operator_uri": "@harpreetsahota/qwen38-max/qwen38_chat", + "dataset_name": datasetName, + "view": []interface{}{}, + "selected": []interface{}{}, + "selected_samples": []interface{}{}, + "selected_labels": []interface{}{}, + "filters": map[string]interface{}{}, + "extended": map[string]interface{}{}, + "params": map[string]interface{}{ + "panel_id": "qwen38_chat", + "__method__": "get_stream_chunk", + "run_id": runID, + "cursor": streamCursor, + }, + }) + sReq, err := http.NewRequestWithContext(r.Context(), "POST", executeURL, bytes.NewReader(streamPayload)) + if err == nil { + sReq.Header.Set("Content-Type", "application/json") + sReq.Header.Set("X-FiftyOne-Session", token) + sReq.Header.Set("User-Agent", reqUA) + if reqHF != "" { + sReq.Header.Set("Authorization", "Bearer "+reqHF) + } + sResp, sErr := gw.client.Do(sReq) + if sErr == nil { + var sResult struct { + Result struct { + Text string `json:"text"` + Cursor int `json:"cursor"` + Done bool `json:"done"` + FinalStatus struct { + Status string `json:"status"` + PromptTokens int `json:"prompt_tokens"` + CompletionTokens int `json:"completion_tokens"` + Error string `json:"error"` + } `json:"final_status"` + } `json:"result"` + } + if json.NewDecoder(sResp.Body).Decode(&sResult) == nil { + if sResult.Result.Text != "" { + accumulatedContent.WriteString(sResult.Result.Text) + streamCursor = sResult.Result.Cursor + } + if sResult.Result.Done { + sResp.Body.Close() + promptTokens = sResult.Result.FinalStatus.PromptTokens + completionTokens = sResult.Result.FinalStatus.CompletionTokens + + fullContent := accumulatedContent.String() + rawThinking := accumulatedThinking.String() + + // Check for embedded tags in content + embeddedThink, cleanedContent := ExtractThinkingContent(fullContent) + if embeddedThink != "" { + if rawThinking != "" { + rawThinking = rawThinking + "\n\n" + embeddedThink + } else { + rawThinking = embeddedThink + } + fullContent = cleanedContent + } + + toolCalls, remainingText := DetectToolCalls(fullContent) + finishReason := "stop" + var finalMsgContent interface{} = strings.TrimSpace(remainingText) + + if len(toolCalls) > 0 { + finishReason = "tool_calls" + if strings.TrimSpace(remainingText) == "" { + finalMsgContent = nil + } + } + + out := FinalOutput{ + Content: finalMsgContent, + ReasoningContent: strings.TrimSpace(rawThinking), + ToolCalls: toolCalls, + FinishReason: finishReason, + PromptTokens: promptTokens, + CompletionTokens: completionTokens, + } + + WriteCompletionResponse(w, completionID, createdTime, modelName, out) + log.Printf("[Non-Stream] Completed run_id=%s finish_reason=%s tokens=%d/%d", runID, finishReason, promptTokens, completionTokens) + return + } + } + sResp.Body.Close() + } + } + } + } + } +} + +// --------------------------------------------------------------------------- +// Gateway Entrypoint +// --------------------------------------------------------------------------- + +func main() { + port := flag.Int("port", 8080, "Port to listen on") + spaceURL := flag.String("space-url", DefaultSpaceURL, "Upstream Hugging Face space URL") + samplePath := flag.String("sample-path", DefaultSamplePath, "Target image sample filepath inside the container") + modelID := flag.String("model", DefaultModelID, "Default model identifier") + timeoutSec := flag.Int("timeout", 300, "Request timeout in seconds") + userAge := flag.String("user-agent", "", "Custom User-Agent header") + uaShort := flag.String("ua", "", "Custom User-Agent header (short)") + hfToken := flag.String("hf-token", "", "Optional Hugging Face access token") + flag.Parse() + + if *userAge != "" { + ConfiguredUserAgent = *userAge + } else if *uaShort != "" { + ConfiguredUserAgent = *uaShort + } + if *hfToken != "" { + ConfiguredToken = *hfToken + } + + gw := NewQ38Gateway(*spaceURL, *samplePath, *modelID, time.Duration(*timeoutSec)*time.Second) + + mux := http.NewServeMux() + + // OpenAI routes + mux.HandleFunc("/v1/models", gw.HandleModels) + mux.HandleFunc("/models", gw.HandleModels) + mux.HandleFunc("/v1/chat/completions", gw.HandleChatCompletions) + mux.HandleFunc("/chat/completions", gw.HandleChatCompletions) + + // Health and index routes + mux.HandleFunc("/__health", func(w http.ResponseWriter, r *http.Request) { + EnableCORS(w) + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"status":"ok","gateway":"q38max"}`)) + }) + mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) { + EnableCORS(w) + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"status":"ok","gateway":"q38max"}`)) + }) + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + EnableCORS(w) + if r.URL.Path == "/" { + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"service":"q38max","description":"OpenAI-compatible gateway for Qwen 3.8 Max","models_endpoint":"/v1/models","completions_endpoint":"/v1/chat/completions"}`)) + return + } + http.NotFound(w, r) + }) + + addr := fmt.Sprintf(":%d", *port) + log.Printf("[q38max] Listening on http://localhost%s (Upstream: %s)", addr, *spaceURL) + log.Printf("[q38max] OpenAI compatible endpoints: http://localhost%s/v1/chat/completions", addr) + + if err := http.ListenAndServe(addr, mux); err != nil { + log.Fatalf("[q38max] Server failed: %v", err) + } +} diff --git a/q38max_test.go b/q38max_test.go new file mode 100644 index 0000000..4969640 --- /dev/null +++ b/q38max_test.go @@ -0,0 +1,119 @@ +// Unit and integration tests for q38max +// Created by Luxferre in 2026, released into the public domain + +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestChatMessageGetContentString(t *testing.T) { + msg1 := ChatMessage{Role: "user", Content: "Hello world"} + if msg1.GetContentString() != "Hello world" { + t.Fatalf("expected 'Hello world', got %q", msg1.GetContentString()) + } + + 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()) + } +} + +func TestExtractThinkingContent(t *testing.T) { + raw := "\nAnalyzing the user's request...\n\nHere is the answer." + thinking, clean := ExtractThinkingContent(raw) + if thinking != "Analyzing the user's request..." { + t.Fatalf("unexpected thinking extraction: %q", thinking) + } + if clean != "Here is the answer." { + t.Fatalf("unexpected clean text: %q", clean) + } +} + +func TestDetectToolCalls(t *testing.T) { + xmlInput := "Let me check the weather.\n{\"name\": \"get_weather\", \"arguments\": {\"location\": \"Tokyo\"}}" + calls, rem := DetectToolCalls(xmlInput) + if len(calls) != 1 { + t.Fatalf("expected 1 tool call, got %d", len(calls)) + } + if calls[0].Function.Name != "get_weather" { + t.Fatalf("expected function name 'get_weather', got %q", calls[0].Function.Name) + } + if !strings.Contains(calls[0].Function.Arguments, "Tokyo") { + t.Fatalf("expected argument with Tokyo, got %q", calls[0].Function.Arguments) + } + if strings.TrimSpace(rem) != "Let me check the weather." { + t.Fatalf("unexpected remaining text: %q", rem) + } +} + +func TestPrepareConversation(t *testing.T) { + req := ChatCompletionRequest{ + Model: "qwen-3.8-max", + Messages: []ChatMessage{ + {Role: "system", Content: "You are a helpful assistant."}, + {Role: "user", Content: "Tell me a joke."}, + {Role: "assistant", Content: "Why did the chicken cross the road?"}, + {Role: "user", Content: "Why?"}, + }, + ReasoningEffort: "medium", + } + + history, question, thinkingMode := PrepareConversation(req) + if thinkingMode != "true" { + t.Fatalf("expected thinkingMode 'true', got %q", thinkingMode) + } + if len(history) != 2 { + t.Fatalf("expected 2 history items, got %d", len(history)) + } + if question != "Why?" { + t.Fatalf("expected question 'Why?', got %q", question) + } + if !strings.Contains(history[0]["content"].(string), "You are a helpful assistant.") { + t.Fatalf("expected system prompt inside first turn, got %v", history[0]["content"]) + } +} + +func TestModelsHandler(t *testing.T) { + gw := NewQ38Gateway("https://mock.hf.space", "/dummy/path.jpg", "qwen-3.8-max", 5*time.Second) + + req := httptest.NewRequest("GET", "/v1/models", nil) + w := httptest.NewRecorder() + + gw.HandleModels(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200 OK, got %d", w.Code) + } + + var res ModelsResponse + if err := json.NewDecoder(w.Body).Decode(&res); err != nil { + t.Fatalf("failed to decode response: %v", err) + } + + if len(res.Data) == 0 { + t.Fatalf("expected at least 1 model in response") + } + + found := false + for _, m := range res.Data { + if m.ID == "qwen-3.8-max" { + found = true + break + } + } + if !found { + t.Fatalf("qwen-3.8-max not found in models list") + } +} diff --git a/xtest.sh b/xtest.sh new file mode 100755 index 0000000..e720358 --- /dev/null +++ b/xtest.sh @@ -0,0 +1,67 @@ +#!/usr/bin/env bash +set -e + +PORT=${1:-18080} +BASE_URL="http://localhost:${PORT}" + +echo "=== 1. Testing Models Endpoint ===" +curl -s "${BASE_URL}/v1/models" | jq . + +echo "" +echo "=== 2. Testing Non-Streaming Chat Completion ===" +curl -s -X POST "${BASE_URL}/v1/chat/completions" \ + -H "Content-Type: application/json" \ + -d '{ + "model": "qwen-3.8-max", + "messages": [ + {"role": "user", "content": "What is the capital of Italy? Answer in 1 word."} + ], + "reasoning_effort": "none", + "max_tokens": 50 + }' | jq . + +echo "" +echo "=== 3. Testing Streaming SSE Completion (with reasoning) ===" +curl -N -s -X POST "${BASE_URL}/v1/chat/completions" \ + -H "Content-Type: application/json" \ + -d '{ + "model": "qwen-3.8-max", + "messages": [ + {"role": "user", "content": "Calculate 25 * 25 and explain briefly in one sentence."} + ], + "stream": true, + "reasoning_effort": "medium", + "max_tokens": 150 + }' + +echo "" +echo "=== 4. Testing Function/Tool Calling ===" +curl -s -X POST "${BASE_URL}/v1/chat/completions" \ + -H "Content-Type: application/json" \ + -d '{ + "model": "qwen-3.8-max", + "messages": [ + {"role": "user", "content": "What is the weather in Berlin?"} + ], + "tools": [ + { + "type": "function", + "function": { + "name": "get_weather", + "description": "Get current weather for location", + "parameters": { + "type": "object", + "properties": { + "location": {"type": "string"} + }, + "required": ["location"] + } + } + } + ], + "reasoning_effort": "none", + "max_tokens": 200 + }' | jq . + +echo "" +echo "=== All integration tests finished successfully! ==="