Merge branch 'main' into plus

internal/api/modules/amp/proxy.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
@@ -108,6 +109,19 @@ func createReverseProxy(upstreamURL string, secretSource SecretSource) (*httputi
 	// Modify incoming responses to handle gzip without Content-Encoding
 	// This addresses the same issue as inline handler gzip handling, but at the proxy level
 	proxy.ModifyResponse = func(resp *http.Response) error {
+		// Log upstream error responses for diagnostics (502, 503, etc.)
+		// These are NOT proxy connection errors - the upstream responded with an error status
+		if resp.StatusCode >= 500 {
+			log.Errorf("amp upstream responded with error [%d] for %s %s", resp.StatusCode, resp.Request.Method, resp.Request.URL.Path)
+		} else if resp.StatusCode >= 400 {
+			log.Warnf("amp upstream responded with client error [%d] for %s %s", resp.StatusCode, resp.Request.Method, resp.Request.URL.Path)
+		}
+
+		// Only process successful responses for gzip decompression
+		if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+			return nil
+		}
+
 		// Skip if already marked as gzip (Content-Encoding set)
 		if resp.Header.Get("Content-Encoding") != "" {
 			return nil
@@ -187,13 +201,29 @@ func createReverseProxy(upstreamURL string, secretSource SecretSource) (*httputi
 		return nil
 	}
 
-	// Error handler for proxy failures
+	// Error handler for proxy failures with detailed error classification for diagnostics
 	proxy.ErrorHandler = func(rw http.ResponseWriter, req *http.Request, err error) {
-		// Client-side cancellations are common during polling; suppress logging in this case
+		// Classify the error type for better diagnostics
+		var errType string
+		if errors.Is(err, context.DeadlineExceeded) {
+			errType = "timeout"
+		} else if errors.Is(err, context.Canceled) {
+			errType = "canceled"
+		} else if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
+			errType = "dial_timeout"
+		} else if _, ok := err.(net.Error); ok {
+			errType = "network_error"
+		} else {
+			errType = "connection_error"
+		}
+
+		// Don't log as error for context canceled - it's usually client closing connection
 		if errors.Is(err, context.Canceled) {
 			return
+		} else {
+			log.Errorf("amp upstream proxy error [%s] for %s %s: %v", errType, req.Method, req.URL.Path, err)
 		}
-		log.Errorf("amp upstream proxy error for %s %s: %v", req.Method, req.URL.Path, err)
+
 		rw.Header().Set("Content-Type", "application/json")
 		rw.WriteHeader(http.StatusBadGateway)
 		_, _ = rw.Write([]byte(`{"error":"amp_upstream_proxy_error","message":"Failed to reach Amp upstream"}`))

internal/api/modules/amp/response_rewriter.go

@@ -36,6 +36,8 @@ func NewResponseRewriter(w gin.ResponseWriter, originalModel string) *ResponseRe
 const maxBufferedResponseBytes = 2 * 1024 * 1024 // 2MB safety cap
 
 func looksLikeSSEChunk(data []byte) bool {
+	// Fallback detection: some upstreams may omit/lie about Content-Type, causing SSE to be buffered.
+	// We conservatively detect SSE by checking for "data:" / "event:" at the start of any line.
 	for _, line := range bytes.Split(data, []byte("\n")) {
 		trimmed := bytes.TrimSpace(line)
 		if bytes.HasPrefix(trimmed, []byte("data:")) ||
@@ -52,8 +54,10 @@ func (rw *ResponseRewriter) enableStreaming(reason string) error {
 	}
 	rw.isStreaming = true
 
+	// Flush any previously buffered data to avoid reordering or data loss.
 	if rw.body != nil && rw.body.Len() > 0 {
 		buf := rw.body.Bytes()
+		// Copy before Reset() to keep bytes stable.
 		toFlush := make([]byte, len(buf))
 		copy(toFlush, buf)
 		rw.body.Reset()
@@ -70,7 +74,9 @@ func (rw *ResponseRewriter) enableStreaming(reason string) error {
 	return nil
 }
 
+// Write intercepts response writes and buffers them for model name replacement
 func (rw *ResponseRewriter) Write(data []byte) (int, error) {
+	// Detect streaming on first write (header-based)
 	if !rw.isStreaming && rw.body.Len() == 0 {
 		contentType := rw.Header().Get("Content-Type")
 		rw.isStreaming = strings.Contains(contentType, "text/event-stream") ||
@@ -78,11 +84,13 @@ func (rw *ResponseRewriter) Write(data []byte) (int, error) {
 	}
 
 	if !rw.isStreaming {
+		// Content-based fallback: detect SSE-like chunks even if Content-Type is missing/wrong.
 		if looksLikeSSEChunk(data) {
 			if err := rw.enableStreaming("sse heuristic"); err != nil {
 				return 0, err
 			}
 		} else if rw.body.Len()+len(data) > maxBufferedResponseBytes {
+			// Safety cap: avoid unbounded buffering on large responses.
 			log.Warnf("amp response rewriter: buffer exceeded %d bytes, switching to streaming", maxBufferedResponseBytes)
 			if err := rw.enableStreaming("buffer limit"); err != nil {
 				return 0, err