core/server/handlers/stream_loop.go

package handlers

import (
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"log"
	"net/http"
	"strings"
	"time"

	"github.com/gin-gonic/gin"

	"chat-switchboard/events"
	"chat-switchboard/metrics"
	"chat-switchboard/providers"
	"chat-switchboard/sandbox"
	"chat-switchboard/store"
	"chat-switchboard/tools"
)

// recordHealthFn records provider health from standalone streaming functions
// and updates Prometheus completion metrics (v0.33.0).
func recordHealthFn(hr HealthRecorder, configID string, start time.Time, err error) {
	duration := time.Since(start)
	latencyMs := int(duration.Milliseconds())

	if configID != "" {
		metrics.CompletionDuration.WithLabelValues(configID, "").Observe(duration.Seconds())
	}

	if hr == nil || configID == "" {
		return
	}
	if err != nil {
		errMsg := err.Error()
		status := "error"
		if strings.Contains(errMsg, "HTTP 429") || strings.Contains(errMsg, "rate limit") || strings.Contains(errMsg, "Too Many Requests") {
			hr.RecordRateLimit(configID, latencyMs, errMsg)
			status = "rate_limited"
		} else {
			hr.RecordError(configID, latencyMs, errMsg)
		}
		metrics.CompletionsTotal.WithLabelValues(configID, "", status).Inc()
	} else {
		hr.RecordSuccess(configID, latencyMs)
		metrics.CompletionsTotal.WithLabelValues(configID, "", "success").Inc()
	}
}

// streamWithToolLoop is the SSE streaming entry point for single-model
// completion with tool execution. Sets SSE headers, delegates to
// CoreToolLoop with an sseSink, and returns the accumulated result.
//
// Used by:
//   - streamCompletion (normal chat — persists via persistMessage)
//   - Regenerate       (regen/edit — persists as sibling with tree metadata)
func streamWithToolLoop(
	c *gin.Context,
	provider providers.Provider,
	cfg providers.ProviderConfig,
	req *providers.CompletionRequest,
	model, providerType, userID, channelID, personaID, workspaceID, configID, teamID string,
	hub *events.Hub,
	health HealthRecorder,
	runner *sandbox.Runner,
	extTools map[string]*store.PackageRegistration,
) LoopResult {
	// Set SSE headers
	c.Header("Content-Type", "text/event-stream")
	c.Header("Cache-Control", "no-cache")
	c.Header("Connection", "keep-alive")
	c.Header("X-Accel-Buffering", "no")
	c.Status(http.StatusOK)

	sink := newSSESink(c, model)

	return CoreToolLoop(c.Request.Context(), LoopConfig{
		Provider:     provider,
		Cfg:          cfg,
		Req:          req,
		Model:        model,
		ProviderType: providerType,
		ExecCtx: tools.ExecutionContext{
			UserID:      userID,
			ChannelID:   channelID,
			PersonaID:   personaID,
			WorkspaceID: workspaceID,
			TeamID:      teamID,
		},
		Hub:       hub,
		Health:    health,
		ConfigID:  configID,
		Budget:    LoopBudget{},
		Streaming: true,
		Runner:    runner,
		ExtTools:  extTools,
	}, sink)
}

const browserToolTimeout = 30 * time.Second

// streamModelResponse is the multi-model streaming variant. Runs on an
// already-established SSE connection — does NOT set SSE headers or send
// [DONE]. SSE deltas include a "model_display" field for frontend attribution.
func streamModelResponse(
	c *gin.Context,
	provider providers.Provider,
	cfg providers.ProviderConfig,
	req *providers.CompletionRequest,
	model, providerType, displayName, userID, channelID, personaID, workspaceID, configID, teamID string,
	hub *events.Hub,
	health HealthRecorder,
	runner *sandbox.Runner,
	extTools map[string]*store.PackageRegistration,
) LoopResult {
	sink := newSSEModelSink(c, model, displayName)

	return CoreToolLoop(c.Request.Context(), LoopConfig{
		Provider:     provider,
		Cfg:          cfg,
		Req:          req,
		Model:        model,
		ProviderType: providerType,
		ExecCtx: tools.ExecutionContext{
			UserID:      userID,
			ChannelID:   channelID,
			PersonaID:   personaID,
			WorkspaceID: workspaceID,
			TeamID:      teamID,
		},
		Hub:       hub,
		Health:    health,
		ConfigID:  configID,
		Budget:    LoopBudget{},
		Streaming: true,
		Runner:    runner,
		ExtTools:  extTools,
	}, sink)
}

// 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 {
	b := make([]byte, 16)
	rand.Read(b)
	callID := hex.EncodeToString(b)

	// Publish tool.call event to the user's browser
	payload, _ := json.Marshal(map[string]interface{}{
		"call_id":   callID,
		"tool":      call.Name,
		"arguments": json.RawMessage(call.Arguments),
	})

	hub.PublishToUser(userID, events.Event{
		Label:   "tool.call." + callID,
		Payload: payload,
		Ts:      time.Now().UnixMilli(),
	})

	// Block until browser sends tool.result.{callID} or timeout
	event, ok := hub.GetBus().WaitFor("tool.result."+callID, browserToolTimeout)
	if !ok {
		log.Printf("⚠️ Browser tool %s timed out after %v (call %s)", call.Name, browserToolTimeout, callID)
		return tools.ToolResult{
			ToolCallID: call.ID,
			Name:       call.Name,
			Content:    `{"error":"browser tool timed out (30s) — browser may be disconnected"}`,
			IsError:    true,
		}
	}

	// Parse result from the browser event payload
	var result struct {
		CallID string `json:"call_id"`
		Result string `json:"result"`
		Error  string `json:"error"`
	}
	if err := json.Unmarshal(event.Payload, &result); err != nil {
		return tools.ToolResult{
			ToolCallID: call.ID,
			Name:       call.Name,
			Content:    `{"error":"invalid result from browser tool"}`,
			IsError:    true,
		}
	}

	if result.Error != "" {
		return tools.ToolResult{
			ToolCallID: call.ID,
			Name:       call.Name,
			Content:    result.Error,
			IsError:    true,
		}
	}

	return tools.ToolResult{
		ToolCallID: call.ID,
		Name:       call.Name,
		Content:    result.Result,
	}
}