package handlers import ( "database/sql" "encoding/json" "fmt" "log" "net/http" "strings" "github.com/gin-gonic/gin" "git.gobha.me/xcaliber/chat-switchboard/database" "git.gobha.me/xcaliber/chat-switchboard/models" "git.gobha.me/xcaliber/chat-switchboard/providers" "git.gobha.me/xcaliber/chat-switchboard/crypto" capspkg "git.gobha.me/xcaliber/chat-switchboard/capabilities" "git.gobha.me/xcaliber/chat-switchboard/tools" ) // ── Request Types ─────────────────────────── type completionRequest struct { ChannelID string `json:"channel_id"` // preferred; validated manually below ChatID string `json:"chat_id"` // deprecated alias — maps to channel_id Content string `json:"content" binding:"required"` Model string `json:"model,omitempty"` PresetID string `json:"preset_id,omitempty"` // if set, unwraps preset → base model + config APIConfigID string `json:"provider_config_id,omitempty"` MaxTokens int `json:"max_tokens,omitempty"` Temperature *float64 `json:"temperature,omitempty"` TopP *float64 `json:"top_p,omitempty"` Stream *bool `json:"stream,omitempty"` } // CompletionHandler proxies LLM requests through the backend. type CompletionHandler struct{ vault *crypto.KeyResolver } // NewCompletionHandler creates a new handler. func NewCompletionHandler(vault *crypto.KeyResolver) *CompletionHandler { return &CompletionHandler{vault: vault} } // ── Chat Completion ───────────────────────── // POST /api/v1/chat/completions // // Flow: // 1. Validate request, verify chat ownership // 2. Resolve api_config (from request, chat, or user default) // 3. Load conversation history from messages // 4. Persist user message // 5. Attach tool definitions if model supports tools // 6. Call provider (stream or non-stream) // 7. Tool loop: if LLM requests tool calls, execute them and re-call // 8. Stream SSE to client / return JSON // 9. Persist assistant message with token counts func (h *CompletionHandler) Complete(c *gin.Context) { var req completionRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } // Support chat_id as alias during frontend transition channelID := req.ChannelID if channelID == "" { channelID = req.ChatID } if channelID == "" { c.JSON(http.StatusBadRequest, gin.H{"error": "channel_id is required"}) return } userID := getUserID(c) // Verify channel ownership if !userOwnsChannel(c, channelID, userID) { return } // ── Preset unwrap: preset overrides defaults, explicit request fields win ── var presetSystemPrompt string if req.PresetID != "" { preset := ResolvePreset(req.PresetID, userID) if preset == nil { c.JSON(http.StatusBadRequest, gin.H{"error": "preset not found or not accessible"}) return } // Preset provides defaults; explicit request fields take priority if req.Model == "" { req.Model = preset.BaseModelID } if req.APIConfigID == "" && preset.ProviderConfigID != nil { req.APIConfigID = *preset.ProviderConfigID } if req.Temperature == nil && preset.Temperature != nil { req.Temperature = preset.Temperature } if req.MaxTokens == 0 && preset.MaxTokens != nil { req.MaxTokens = *preset.MaxTokens } if preset.SystemPrompt != "" { presetSystemPrompt = preset.SystemPrompt } } // Resolve provider config providerCfg, providerID, model, configID, err := h.resolveConfig(userID, channelID, req) if err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } // ── Team policy: require_private_providers ── if err := enforcePrivateProviderPolicy(userID, configID); err != nil { c.JSON(http.StatusForbidden, gin.H{"error": err.Error()}) return } provider, err := providers.Get(providerID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } // Load conversation history messages, err := h.loadConversation(channelID, userID, presetSystemPrompt) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load conversation"}) return } // Add the new user message messages = append(messages, providers.Message{ Role: "user", Content: req.Content, }) // Persist user message if _, err := h.persistMessage(channelID, userID, "user", req.Content, "", 0, 0, nil, nil); err != nil { log.Printf("Failed to persist user message: %v", err) } // Build provider request provReq := providers.CompletionRequest{ Model: model, Messages: messages, } // Resolve capabilities for this model — auto-set defaults caps := h.getModelCapabilities(c, model, configID) if req.MaxTokens > 0 { provReq.MaxTokens = req.MaxTokens } else { // ResolveMaxOutput checks: caps → known models → context/8 → 4096 provReq.MaxTokens = capspkg.ResolveMaxOutput(model, caps) } if req.Temperature != nil { provReq.Temperature = req.Temperature } if req.TopP != nil { provReq.TopP = req.TopP } // Attach tool definitions if model supports tool calling and tools are available if caps.ToolCalling && tools.HasTools() { provReq.Tools = h.buildToolDefs() } // Determine streaming stream := true // default if req.Stream != nil { stream = *req.Stream } if stream { h.streamCompletion(c, provider, providerCfg, provReq, channelID, userID, model) } else { h.syncCompletion(c, provider, providerCfg, provReq, channelID, userID, model) } } // buildToolDefs converts registered tools to provider-format tool definitions. func (h *CompletionHandler) buildToolDefs() []providers.ToolDef { allTools := tools.AllDefinitions() defs := make([]providers.ToolDef, len(allTools)) for i, t := range allTools { defs[i] = providers.ToolDef{ Type: "function", Function: providers.FunctionDef{ Name: t.Name, Description: t.Description, Parameters: t.Parameters, }, } } return defs } // ── Streaming Completion (SSE) with Tool Loop ── const maxToolIterations = 10 // safety limit on tool call rounds func (h *CompletionHandler) streamCompletion( c *gin.Context, provider providers.Provider, cfg providers.ProviderConfig, req providers.CompletionRequest, channelID, userID, model string, ) { result := streamWithToolLoop(c, provider, cfg, &req, model, userID, channelID) // Persist assistant response if result.Content != "" { if _, err := h.persistMessage(channelID, userID, "assistant", result.Content, model, 0, 0, nil, toolActivityJSON(result.ToolActivity)); err != nil { log.Printf("Failed to persist assistant message: %v", err) } } } // ── Non-Streaming Completion with Tool Loop ── func (h *CompletionHandler) syncCompletion( c *gin.Context, provider providers.Provider, cfg providers.ProviderConfig, req providers.CompletionRequest, channelID, userID, model string, ) { var totalInput, totalOutput int var finalContent string var allToolActivity []map[string]interface{} for iteration := 0; iteration < maxToolIterations; iteration++ { resp, err := provider.ChatCompletion(c.Request.Context(), cfg, req) if err != nil { c.JSON(http.StatusBadGateway, gin.H{"error": "provider error: " + err.Error()}) return } totalInput += resp.InputTokens totalOutput += resp.OutputTokens // No tool calls — normal response if len(resp.ToolCalls) == 0 || resp.FinishReason != "tool_calls" { finalContent = resp.Content // Persist assistant response if _, err := h.persistMessage(channelID, userID, "assistant", finalContent, model, totalInput, totalOutput, nil, toolActivityJSON(allToolActivity)); err != nil { log.Printf("Failed to persist assistant message: %v", err) } // Return OpenAI-compatible response c.JSON(http.StatusOK, gin.H{ "model": resp.Model, "choices": []gin.H{ { "message": gin.H{ "role": "assistant", "content": finalContent, }, "finish_reason": resp.FinishReason, }, }, "usage": gin.H{ "prompt_tokens": totalInput, "completion_tokens": totalOutput, "total_tokens": totalInput + totalOutput, }, }) return } // ── Tool execution round ──────────────── assistantMsg := providers.Message{ Role: "assistant", Content: resp.Content, } for _, tc := range resp.ToolCalls { assistantMsg.ToolCalls = append(assistantMsg.ToolCalls, tc) } req.Messages = append(req.Messages, assistantMsg) execCtx := tools.ExecutionContext{ UserID: userID, ChannelID: channelID, } for _, tc := range resp.ToolCalls { call := tools.ToolCall{ ID: tc.ID, Name: tc.Function.Name, Arguments: tc.Function.Arguments, } log.Printf("🔧 Executing tool: %s (call %s)", call.Name, call.ID) result := tools.ExecuteCall(c.Request.Context(), execCtx, call) // Collect for persistence allToolActivity = append(allToolActivity, map[string]interface{}{ "id": tc.ID, "name": tc.Function.Name, "arguments": tc.Function.Arguments, "result": result.Content, "is_error": result.IsError, }) req.Messages = append(req.Messages, providers.Message{ Role: "tool", Content: result.Content, ToolCallID: result.ToolCallID, Name: result.Name, }) } } // Max iterations reached c.JSON(http.StatusOK, gin.H{ "model": model, "choices": []gin.H{ { "message": gin.H{ "role": "assistant", "content": "[Tool execution limit reached]", }, "finish_reason": "stop", }, }, }) } // ── Model Capabilities ────────────────────── // escapeJSON escapes a string for safe embedding in a JSON string value. func escapeJSON(s string) string { s = strings.ReplaceAll(s, `\`, `\\`) s = strings.ReplaceAll(s, `"`, `\"`) s = strings.ReplaceAll(s, "\n", `\n`) s = strings.ReplaceAll(s, "\r", `\r`) s = strings.ReplaceAll(s, "\t", `\t`) return s } // getModelCapabilities looks up capabilities from model_catalog DB, // then overlays with known model defaults and heuristic detection. func (h *CompletionHandler) getModelCapabilities(c *gin.Context, model, apiConfigID string) models.ModelCapabilities { return ResolveModelCaps(c, model, apiConfigID) } // ── Config Resolution ─────────────────────── // Priority: request.provider_config_id → chat.provider_config_id → user's first active config func (h *CompletionHandler) resolveConfig(userID string, channelID string, req completionRequest) (providers.ProviderConfig, string, string, string, error) { var configID string // 1. Explicit config from request if req.APIConfigID != "" { configID = req.APIConfigID } // 2. Config from channel if configID == "" { var channelConfigID *string err := database.DB.QueryRow( `SELECT provider_config_id FROM channels WHERE id = $1`, channelID, ).Scan(&channelConfigID) if err == nil && channelConfigID != nil { configID = *channelConfigID } } // 3. User's first active config (personal first, then global — excludes team providers) if configID == "" { err := database.DB.QueryRow(` SELECT id FROM provider_configs WHERE is_active = true AND ( (scope = 'personal' AND owner_id = $1) OR scope = 'global' ) ORDER BY scope ASC, created_at ASC LIMIT 1 `, userID).Scan(&configID) if err != nil { return providers.ProviderConfig{}, "", "", "", fmt.Errorf("no API config found — add one at /api-configs") } } // Load the config — allow personal, global, OR team configs the user belongs to var providerID, endpoint string var modelDefault *string var apiKeyEnc, keyNonce []byte var keyScope string var customHeadersJSON, providerSettingsJSON []byte err := database.DB.QueryRow(` SELECT provider, endpoint, api_key_enc, key_nonce, key_scope, model_default, headers, settings FROM provider_configs WHERE id = $1 AND is_active = true AND (scope = 'global' OR (scope = 'personal' AND owner_id = $2) OR (scope = 'team' AND owner_id IN (SELECT team_id FROM team_members WHERE user_id = $2))) `, configID, userID).Scan(&providerID, &endpoint, &apiKeyEnc, &keyNonce, &keyScope, &modelDefault, &customHeadersJSON, &providerSettingsJSON) if err == sql.ErrNoRows { return providers.ProviderConfig{}, "", "", "", fmt.Errorf("API config not found or not accessible") } if err != nil { return providers.ProviderConfig{}, "", "", "", fmt.Errorf("failed to load API config: %w", err) } // Resolve model: request > config default model := req.Model if model == "" && modelDefault != nil { model = *modelDefault } if model == "" { return providers.ProviderConfig{}, "", "", "", fmt.Errorf("no model specified and no default model in config") } // Decrypt API key using the appropriate tier key := "" if len(apiKeyEnc) > 0 { if h.vault != nil { var err error key, err = h.vault.Decrypt(apiKeyEnc, keyNonce, keyScope, userID) if err != nil { if err == crypto.ErrVaultLocked { return providers.ProviderConfig{}, "", "", "", fmt.Errorf("personal vault is locked — please log in again") } return providers.ProviderConfig{}, "", "", "", fmt.Errorf("failed to decrypt API key: %w", err) } } else { // No vault — key stored as raw bytes (unencrypted fallback) key = string(apiKeyEnc) } } // Parse custom headers customHeaders := make(map[string]string) if customHeadersJSON != nil { _ = json.Unmarshal(customHeadersJSON, &customHeaders) } // Parse provider-specific settings providerSettings := make(map[string]interface{}) if providerSettingsJSON != nil { _ = json.Unmarshal(providerSettingsJSON, &providerSettings) } return providers.ProviderConfig{ Endpoint: endpoint, APIKey: key, CustomHeaders: customHeaders, Settings: providerSettings, }, providerID, model, configID, nil } // ── Conversation Loader ───────────────────── // loadConversation builds the message history for the LLM by walking the // active path through the message tree. Only the current branch is sent — // sibling branches are never included in context. func (h *CompletionHandler) loadConversation(channelID, userID, presetSystemPrompt string) ([]providers.Message, error) { // Load system prompt from channel var systemPrompt *string _ = database.DB.QueryRow( `SELECT system_prompt FROM channels WHERE id = $1`, channelID, ).Scan(&systemPrompt) messages := make([]providers.Message, 0) // Preset system prompt takes priority; channel system prompt is fallback if presetSystemPrompt != "" { messages = append(messages, providers.Message{ Role: "system", Content: presetSystemPrompt, }) } else if systemPrompt != nil && *systemPrompt != "" { messages = append(messages, providers.Message{ Role: "system", Content: *systemPrompt, }) } // Walk the active path (root → leaf) instead of loading all messages path, err := getActivePath(channelID, userID) if err != nil { return nil, err } for _, m := range path { if m.Role == "system" { continue // system prompts handled above } messages = append(messages, providers.Message{ Role: m.Role, Content: m.Content, }) } return messages, nil } // ── Message Persistence ───────────────────── // persistMessage inserts a message into the tree, updates the cursor, and // returns the new message's ID. // // Parent resolution: // - parentOverride (explicit parent for edit/regen operations) // - cursor's active_leaf_id (normal conversation flow) // - latest message by created_at (fallback for legacy/missing cursor) // toolActivityJSON marshals tool activity for persistence. Returns nil if empty. func toolActivityJSON(activity []map[string]interface{}) json.RawMessage { if len(activity) == 0 { return nil } b, _ := json.Marshal(activity) return b } func (h *CompletionHandler) persistMessage(channelID, userID, role, content, model string, inputTokens, outputTokens int, parentOverride *string, toolCallsJSON json.RawMessage) (string, error) { var tokensUsed *int if inputTokens > 0 || outputTokens > 0 { total := inputTokens + outputTokens tokensUsed = &total } // Determine parent var parentID *string if parentOverride != nil { parentID = parentOverride } else { parentID, _ = getActiveLeaf(channelID, userID) } // Compute sibling_index siblingIdx := nextSiblingIndex(channelID, parentID) // Determine participant participantType := "user" participantID := userID if role == "assistant" { participantType = "model" participantID = model } // Prepare tool_calls JSONB (nil → NULL) var tcVal interface{} if len(toolCallsJSON) > 0 { tcVal = string(toolCallsJSON) } // Insert with RETURNING id var newID string err := database.DB.QueryRow(` INSERT INTO messages (channel_id, role, content, model, tokens_used, tool_calls, parent_id, participant_type, participant_id, sibling_index) VALUES ($1, $2, $3, NULLIF($4, ''), $5, $6, $7, $8, $9, $10) RETURNING id `, channelID, role, content, model, tokensUsed, tcVal, parentID, participantType, participantID, siblingIdx, ).Scan(&newID) if err != nil { return "", err } // Update cursor to point at new message if userID != "" { _ = updateCursor(channelID, userID, newID) } // Touch channel updated_at _, _ = database.DB.Exec(`UPDATE channels SET updated_at = NOW() WHERE id = $1`, channelID) return newID, nil }