Changeset 0.20.0 (#85)

server/handlers/stream_loop.go

@@ -240,6 +240,181 @@ func streamWithToolLoop(
 
 const browserToolTimeout = 30 * time.Second
 
+// streamModelResponse is a variant of streamWithToolLoop used by multi-model
+// fan-out (v0.20.0). It streams a single model's response within an already-
+// established SSE connection — it does NOT set SSE headers or send [DONE].
+//
+// SSE deltas include a "model_display" field so the frontend can attribute
+// each response to the correct model.
+func streamModelResponse(
+	c *gin.Context,
+	provider providers.Provider,
+	cfg providers.ProviderConfig,
+	req *providers.CompletionRequest,
+	model, displayName, userID, channelID, personaID string,
+	hub *events.Hub,
+) streamResult {
+	flusher, _ := c.Writer.(http.Flusher)
+	flush := func() {
+		if flusher != nil {
+			flusher.Flush()
+		}
+	}
+	sendSSE := func(data string) {
+		fmt.Fprintf(c.Writer, "data: %s\n\n", data)
+		flush()
+	}
+	sendEvent := func(event, data string) {
+		fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event, data)
+		flush()
+	}
+
+	var result streamResult
+
+	for iteration := 0; iteration < maxToolIterations; iteration++ {
+		ch, err := provider.StreamCompletion(c.Request.Context(), cfg, *req)
+		if err != nil {
+			sendSSE(fmt.Sprintf(`{"error":"%s"}`, escapeJSON(err.Error())))
+			return result
+		}
+
+		var iterContent string
+		var iterReasoning string
+		var toolCalls []providers.ToolCall
+
+		for event := range ch {
+			if event.Error != nil {
+				sendSSE(fmt.Sprintf(`{"error":"%s"}`, escapeJSON(event.Error.Error())))
+				return result
+			}
+
+			if event.Reasoning != "" {
+				iterReasoning += event.Reasoning
+				sendSSE(fmt.Sprintf(
+					`{"choices":[{"delta":{"reasoning_content":%q},"finish_reason":null}],"model":%q,"model_display":%q}`,
+					event.Reasoning, model, displayName))
+			}
+
+			if event.Delta != "" {
+				iterContent += event.Delta
+				sendSSE(fmt.Sprintf(
+					`{"choices":[{"delta":{"content":%q},"finish_reason":null}],"model":%q,"model_display":%q}`,
+					event.Delta, model, displayName))
+			}
+
+			if event.Done {
+				if event.FinishReason == "tool_calls" && len(event.ToolCalls) > 0 {
+					toolCalls = event.ToolCalls
+					result.InputTokens += event.InputTokens
+					result.OutputTokens += event.OutputTokens
+					result.CacheCreationTokens += event.CacheCreationTokens
+					result.CacheReadTokens += event.CacheReadTokens
+				} else {
+					finishReason := event.FinishReason
+					if finishReason == "" {
+						finishReason = "stop"
+					}
+					if iterReasoning != "" {
+						result.Content += "<think>" + iterReasoning + "</think>"
+					}
+					result.Content += iterContent
+					result.InputTokens += event.InputTokens
+					result.OutputTokens += event.OutputTokens
+					result.CacheCreationTokens += event.CacheCreationTokens
+					result.CacheReadTokens += event.CacheReadTokens
+					sendSSE(fmt.Sprintf(
+						`{"choices":[{"delta":{},"finish_reason":%q}],"model":%q,"model_display":%q}`,
+						finishReason, model, displayName))
+					// Do NOT send [DONE] — caller manages that for multi-model
+					return result
+				}
+				break
+			}
+		}
+
+		// Tool execution (same logic as streamWithToolLoop)
+		if len(toolCalls) == 0 {
+			if iterReasoning != "" {
+				result.Content += "<think>" + iterReasoning + "</think>"
+			}
+			result.Content += iterContent
+			return result
+		}
+
+		toolCallsJSON, _ := json.Marshal(toolCalls)
+		sendEvent("tool_use", string(toolCallsJSON))
+
+		assistantMsg := providers.Message{
+			Role:    "assistant",
+			Content: iterContent,
+		}
+		for _, tc := range toolCalls {
+			assistantMsg.ToolCalls = append(assistantMsg.ToolCalls, tc)
+		}
+		req.Messages = append(req.Messages, assistantMsg)
+
+		execCtx := tools.ExecutionContext{
+			UserID:    userID,
+			ChannelID: channelID,
+			PersonaID: personaID,
+		}
+		for _, tc := range toolCalls {
+			call := tools.ToolCall{
+				ID:        tc.ID,
+				Name:      tc.Function.Name,
+				Arguments: tc.Function.Arguments,
+			}
+
+			var toolResult tools.ToolResult
+			if tools.Get(call.Name) != nil {
+				toolResult = tools.ExecuteCall(c.Request.Context(), execCtx, call)
+			} else if hub != nil && hub.IsConnected(userID) {
+				toolResult = executeBrowserTool(hub, userID, call)
+			} else {
+				toolResult = tools.ToolResult{
+					ToolCallID: call.ID,
+					Name:       call.Name,
+					Content:    `{"error":"unknown tool or browser not connected"}`,
+					IsError:    true,
+				}
+			}
+
+			resultJSON, _ := json.Marshal(map[string]interface{}{
+				"tool_call_id": toolResult.ToolCallID,
+				"name":         toolResult.Name,
+				"content":      toolResult.Content,
+				"is_error":     toolResult.IsError,
+			})
+			sendEvent("tool_result", string(resultJSON))
+
+			result.ToolActivity = append(result.ToolActivity, map[string]interface{}{
+				"id":        tc.ID,
+				"name":      tc.Function.Name,
+				"arguments": tc.Function.Arguments,
+				"result":    toolResult.Content,
+				"is_error":  toolResult.IsError,
+			})
+
+			req.Messages = append(req.Messages, providers.Message{
+				Role:       "tool",
+				Content:    toolResult.Content,
+				ToolCallID: toolResult.ToolCallID,
+				Name:       toolResult.Name,
+			})
+		}
+
+		result.Content += iterContent
+	}
+
+	// Hit max iterations
+	log.Printf("⚠️ Multi-model tool loop hit max iterations (%d) for model %s", maxToolIterations, displayName)
+	sendSSE(fmt.Sprintf(
+		`{"choices":[{"delta":{"content":"%s"},"finish_reason":"stop"}],"model":%q,"model_display":%q}`,
+		escapeJSON("[Tool execution limit reached]"), model, displayName))
+
+	return result
+}
+
 // executeBrowserTool sends a tool call to the user's browser via WebSocket
 // and blocks until a result is returned or the timeout elapses.
 func executeBrowserTool(hub *events.Hub, userID string, call tools.ToolCall) tools.ToolResult {