Method CompleteStreamingHandler

runtime/server/chat.go:456–547

CompleteStreamingHandler is a HTTP handler that wraps CompleteStreaming and maps it to SSE. This is required as vanguard doesn't currently map streaming RPCs to SSE, so we register this handler manually override the behavior

(w http.ResponseWriter, req *http.Request)

// CompleteStreamingHandler is a HTTP handler that wraps CompleteStreaming and maps it to SSE.
// This is required as vanguard doesn't currently map streaming RPCs to SSE, so we register this handler manually override the behavior
func (s *Server) CompleteStreamingHandler(w http.ResponseWriter, req *http.Request) {
	// Observability
	ctx := req.Context()
	instanceID := req.PathValue("instance_id")
	observability.AddRequestAttributes(ctx,
		attribute.String("args.instance_id", instanceID),
	)

	// Access check
	if !auth.GetClaims(ctx, instanceID).Can(runtime.UseAI) {
		http.Error(w, "action not allowed", http.StatusUnauthorized)
		return
	}

	// Apply configured timeout for AI chat
	cfg, err := s.runtime.InstanceConfig(ctx, instanceID)
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to load instance config: %v", err), http.StatusBadRequest)
		return
	}
	ctx, cancel := context.WithTimeout(ctx, time.Duration(cfg.AICompletionTimeoutSeconds)*time.Second)
	defer cancel()
	req = req.WithContext(ctx)

	// Build request. Note we try to support both GET and POST.
	completeReq := &runtimev1.CompleteStreamingRequest{}
	if req.Method == http.MethodGet {
		// Parse from query parameters
		completeReq.ConversationId = req.URL.Query().Get("conversationId")
		completeReq.Prompt = req.URL.Query().Get("prompt")
	} else {
		// Parse from JSON body
		body, err := io.ReadAll(req.Body)
		if err != nil {
			http.Error(w, "failed to read request body", http.StatusBadRequest)
			return
		}
		if err := protojson.Unmarshal(body, completeReq); err != nil {
			http.Error(w, "failed to parse request body", http.StatusBadRequest)
			return
		}
	}
	completeReq.InstanceId = instanceID // Set instance ID from path

	// Start goroutine that calls CompleteStreaming and publishes responses to a channel
	events := make(chan *sseEvent)
	go func() {
		// Handle panics (it's a separate goroutine so the middleware won't catch panics)
		defer func() {
			if r := recover(); r != nil {
				s.logger.Error("panic in CompleteStreamingHandler subscription goroutine", zap.Any("recover", r), zap.Stack("stack"))
			}
		}()

		// We must close the events channel when done to make sure the SSE handler returns
		defer close(events)

		// Create the shim that implements RuntimeService_CompleteStreamingServer