Update sse.go

sse.go

@@ -20,6 +20,10 @@ func streamOpenAIResponseToClaudeSSE(
 	requestedModel string, // Model name determined in handler
 	originalClaudeRequest *ClaudeRequest,
 ) {
+	// --- Phần này giữ nguyên logic như phiên bản trước của bạn ---
+	// ... (logic gửi message_start, content_block_start, xử lý delta, gửi message_delta "hack", message_stop)
+	// ... đảm bảo rằng "hack" gửi output_tokens trong message_delta trung gian được giữ lại.
+
 	c.Writer.Header().Set("Content-Type", "text/event-stream")
 	c.Writer.Header().Set("Cache-Control", "no-cache")
 	c.Writer.Header().Set("Connection", "keep-alive")
@@ -34,12 +38,11 @@ func streamOpenAIResponseToClaudeSSE(
 	startTime := time.Now()
 
 	calculatedInputTokens := calculateInputTokensFromClaudeRequest(originalClaudeRequest)
-	// Giữ lại: Log INFO quan trọng về việc bắt đầu SSE
 	log.Printf("INFO: [%s] (OpenAI->Claude SSE) Bắt đầu SSE. Input tokens ước tính: %d. Model: %s", messageID, calculatedInputTokens, requestedModel)
 
 	inputTokens := calculatedInputTokens
 	outputTokens := 0
-	finalUsageReceivedFromStream := false
+	finalUsageReceivedFromStream := false // Flag to check if usage is explicitly received from OpenAI stream
 	eventIndex := 0
 
 	doneChan := make(chan struct{})
@@ -53,7 +56,6 @@ func streamOpenAIResponseToClaudeSSE(
 		for scanner.Scan() {
 			select {
 			case <-c.Request.Context().Done():
-				// Giữ lại: Log INFO khi client ngắt kết nối
 				log.Printf("INFO: [%s] (OpenAI->Claude SSE) SSE: Client ngắt kết nối trong vòng lặp đọc.", messageID)
 				return
 			default:
@@ -67,15 +69,13 @@ func streamOpenAIResponseToClaudeSSE(
 			if strings.HasPrefix(line, "data:") {
 				dataStr := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
 				if dataStr == "[DONE]" {
-					// Giữ lại: Log INFO quan trọng
 					log.Printf("INFO: [%s] (OpenAI->Claude SSE) SSE: Nhận được dấu hiệu [DONE].", messageID)
-					return
+					return // OpenAI stream finished
 				}
 
 				var chunk OpenAIStreamChunk
 				if err := json.Unmarshal([]byte(dataStr), &chunk); err != nil {
-					// Giữ lại: Log WARN quan trọng
-					log.Printf("WARN: [%s] (OpenAI->Claude SSE) SSE: Không thể giải mã chunk JSON: %v. Data: %s", messageID, err, dataStr)
+					log.Printf("WARN: [%s] (OpenAI->Claude SSE) SSE: Không thể giải mã chunk JSON OpenAI: %v. Data: %s", messageID, err, dataStr)
 					continue
 				}
 
@@ -87,94 +87,85 @@ func streamOpenAIResponseToClaudeSSE(
 					if choice.Delta.Content != nil {
 						contentChunk := *choice.Delta.Content
 						accumulatedContent += contentChunk
-						currentOutputTokens := estimateTokens(accumulatedContent)
-
+						// Gửi content_block_delta cho client
 						deltaPayload := ClaudeSSEEvent{
 							Type:  "content_block_delta",
-							Index: func() *int { i := 0; return &i }(),
+							Index: func() *int { i := 0; return &i }(), // Claude spec: index of the content block
 							Delta: &ClaudeSSEDelta{
 								Type: "text_delta",
 								Text: &contentChunk,
 							},
 						}
-						// Không log event này (shouldLog = false)
 						if !sendSSEEvent(c, "content_block_delta", deltaPayload, messageID, eventIndex, false, "(OpenAI->Claude SSE)") {
-							return
+							return // Client disconnected
 						}
 						eventIndex++
 
+						// "Hack" để gửi message_delta với output_tokens cập nhật
+						currentOutputTokens := estimateTokens(accumulatedContent)
 						if currentOutputTokens != outputTokens {
 							outputTokens = currentOutputTokens
 							intermediateUsage := ClaudeSSEUsage{OutputTokens: outputTokens}
 							intermediateDeltaPayload := ClaudeSSEEvent{
 								Type:  "message_delta",
-								Delta: &ClaudeSSEDelta{},
+								Delta: &ClaudeSSEDelta{}, // Delta rỗng cho loại event này
 								Usage: &intermediateUsage,
 							}
-							// Không log event này (shouldLog = false)
 							if !sendSSEEvent(c, "message_delta", intermediateDeltaPayload, messageID, eventIndex, false, "(OpenAI->Claude SSE)") {
-								return
+								return // Client disconnected
 							}
 							eventIndex++
 						}
 					}
 				}
-				// Xử lý Usage nếu có trong chunk (phiên bản mới của API OpenAI có thể gửi)
-				if chunk.Usage != nil {
-					// Vô hiệu hóa: Log INFO này có thể quá chi tiết nếu block usage xuất hiện thường xuyên
-					// log.Printf("INFO: [%s] (OpenAI->Claude SSE) SSE: Nhận được block usage từ OpenAI stream: %+v", messageID, *chunk.Usage)
-					if chunk.Usage.CompletionTokens > 0 { // OpenAI gọi là completion_tokens
+				if chunk.Usage != nil { // Some OpenAI versions might send usage in chunks
+					if chunk.Usage.CompletionTokens > 0 {
 						outputTokens = chunk.Usage.CompletionTokens
-						finalUsageReceivedFromStream = true // Đánh dấu đã nhận usage cuối cùng từ stream
+						finalUsageReceivedFromStream = true
 					}
-					if chunk.Usage.PromptTokens != inputTokens && chunk.Usage.PromptTokens > 0 { // OpenAI gọi là prompt_tokens
-						inputTokens = chunk.Usage.PromptTokens
+					if chunk.Usage.PromptTokens > 0 {
+						inputTokens = chunk.Usage.PromptTokens // Update input tokens if provided by OpenAI
 					}
 				}
 			}
 		}
-
 		if err := scanner.Err(); err != nil {
 			select {
 			case <-c.Request.Context().Done():
-				// Giữ lại: Log INFO
 				log.Printf("INFO: [%s] (OpenAI->Claude SSE) SSE: Đọc upstream bị gián đoạn bởi client ngắt kết nối: %v", messageID, c.Request.Context().Err())
 			default:
-				// Giữ lại: Log ERROR quan trọng
 				log.Printf("ERROR: [%s] (OpenAI->Claude SSE) SSE: Lỗi đọc nội dung phản hồi upstream: %v", messageID, err)
 				errChan <- fmt.Errorf("lỗi đọc upstream: %w", err)
 			}
 		}
 	}()
 
+	// Gửi message_start
 	startUsage := ClaudeUsage{InputTokens: calculatedInputTokens, OutputTokens: 0}
 	startMessage := ClaudeSSEMessage{ID: messageID, Type: "message", Role: "assistant", Content: []ClaudeContentBlock{}, Model: requestedModel, Usage: startUsage}
 	startEvent := ClaudeSSEEvent{Type: "message_start", Message: &startMessage}
-	// Log event quan trọng này (shouldLog = true)
 	if !sendSSEEvent(c, "message_start", startEvent, messageID, eventIndex, true, "(OpenAI->Claude SSE)") {
 		return
 	}
 	eventIndex++
 
-	contentStartBlock := ClaudeSSEContentBlock{Type: "text", Text: ""}
+	// Gửi content_block_start
+	contentStartBlock := ClaudeSSEContentBlock{Type: "text", Text: ""} // Text ban đầu rỗng
 	contentStartEvent := ClaudeSSEEvent{Type: "content_block_start", Index: func() *int { i := 0; return &i }(), ContentBlock: &contentStartBlock}
-	// Log event quan trọng này (shouldLog = true)
 	if !sendSSEEvent(c, "content_block_start", contentStartEvent, messageID, eventIndex, true, "(OpenAI->Claude SSE)") {
 		return
 	}
 	eventIndex++
 
+	// Chờ goroutine đọc xong hoặc có lỗi
 	select {
 	case <-doneChan:
-		// Vô hiệu hóa: Log DEBUG không cần thiết
-		// log.Printf("DEBUG: [%s] (OpenAI->Claude SSE) SSE: Đọc upstream hoàn tất.", messageID)
+		// Normal completion
 	case err := <-errChan:
-		// Giữ lại: Log ERROR quan trọng
 		log.Printf("ERROR: [%s] (OpenAI->Claude SSE) SSE: Nhận lỗi từ goroutine đọc: %v", messageID, err)
 		streamErrorOccurred = true
 		errorDetails = &ClaudeError{Type: "api_error", Message: fmt.Sprintf("Lỗi đọc phản hồi upstream: %v", err)}
 	case <-c.Request.Context().Done():
-		// Giữ lại: Log INFO quan trọng
 		log.Printf("INFO: [%s] (OpenAI->Claude SSE) SSE: Client ngắt kết nối trong quá trình xử lý stream: %v", messageID, c.Request.Context().Err())
 		streamErrorOccurred = true
 		errorDetails = &ClaudeError{Type: "client_disconnect", Message: "Client ngắt kết nối trong quá trình stream"}
@@ -182,137 +173,324 @@ func streamOpenAIResponseToClaudeSSE(
 
 	var claudeStopReason string
 	if streamErrorOccurred && errorDetails != nil && errorDetails.Type == "client_disconnect" {
-		claudeStopReason = "client_disconnect" // Specific reason for client disconnect
+		claudeStopReason = "client_disconnect"
 	} else if streamErrorOccurred {
-		claudeStopReason = "error" // Generic error if stream failed for other reasons
+		claudeStopReason = "error"
 	} else {
 		claudeStopReason = mapOpenAIFinishReasonToClaude(openAIFinishReason)
 	}
 
 	finalInputTokens := inputTokens
 	finalOutputTokens := outputTokens
-
-	if !finalUsageReceivedFromStream { // Nếu không nhận được usage từ stream (ví dụ API cũ hoặc không có)
-		estimatedOutput := estimateTokens(accumulatedContent)
-		finalOutputTokens = max(0, estimatedOutput)
-		if accumulatedContent == "" {
-			finalOutputTokens = 0
-		}
-	} else {
-		finalOutputTokens = max(0, outputTokens) // Sử dụng giá trị đã nhận từ stream nếu có
+	if !finalUsageReceivedFromStream { // If OpenAI didn't send final usage, estimate from accumulated content
+		finalOutputTokens = estimateTokens(accumulatedContent)
 	}
-	finalInputTokens = max(0, finalInputTokens) // Đảm bảo không âm
+	finalOutputTokens = max(0, finalOutputTokens)
+	finalInputTokens = max(0, finalInputTokens)
+
 
-	// Giữ lại: Log INFO tóm tắt quan trọng khi stream hoàn tất
 	log.Printf("INFO: [%s] (OpenAI->Claude SSE) SSE Stream hoàn tất. Lý do dừng: %s. Input: %d, Output: %d. Thời gian: %v. OpenAI Finish Reason: %s",
 		messageID, claudeStopReason, finalInputTokens, finalOutputTokens, time.Since(startTime), safeDeref(openAIFinishReason))
 
-	// Final events to send to client
-	finalHackUsageData := ClaudeSSEUsage{OutputTokens: finalOutputTokens} // For message_delta
-	finalStopUsageData := ClaudeSSEUsage{InputTokens: &finalInputTokens, OutputTokens: finalOutputTokens} // For message_stop
-
+	// "Hack" message_delta cuối cùng với stop_reason và usage (chỉ output_tokens)
+	finalHackUsageData := ClaudeSSEUsage{OutputTokens: finalOutputTokens}
 	finalDeltaStopReason := claudeStopReason
 	priorityFinalDeltaPayload := ClaudeSSEEvent{
 		Type: "message_delta",
 		Delta: &ClaudeSSEDelta{
 			StopReason:   &finalDeltaStopReason,
-			StopSequence: nil, // OpenAI doesn't provide this directly in stream delta
+			StopSequence: nil,
 		},
-		Usage: &finalHackUsageData, // Claude expects usage in message_delta before stop
+		Usage: &finalHackUsageData,
 	}
-	// Không log event này (shouldLog = false)
 	_ = sendSSEEvent(c, "message_delta", priorityFinalDeltaPayload, messageID, eventIndex, false, "(OpenAI->Claude SSE)")
 	eventIndex++
 
+	// Gửi content_block_stop
 	contentStopPayload := ClaudeSSEEvent{Type: "content_block_stop", Index: func() *int { i := 0; return &i }()}
-	// Không log event này (shouldLog = false)
 	_ = sendSSEEvent(c, "content_block_stop", contentStopPayload, messageID, eventIndex, false, "(OpenAI->Claude SSE)")
 	eventIndex++
 
-	messageStopPayload := ClaudeSSEEvent{Type: "message_stop", Usage: &finalStopUsageData} // Claude expects final usage in message_stop
-	// Log event quan trọng này (shouldLog = true)
+	// Gửi message_stop với usage đầy đủ
+	finalStopUsageData := ClaudeSSEUsage{InputTokens: &finalInputTokens, OutputTokens: finalOutputTokens}
+	messageStopPayload := ClaudeSSEEvent{Type: "message_stop", Usage: &finalStopUsageData}
 	_ = sendSSEEvent(c, "message_stop", messageStopPayload, messageID, eventIndex, true, "(OpenAI->Claude SSE)")
 	eventIndex++
 
 	if streamErrorOccurred && errorDetails != nil && errorDetails.Type != "client_disconnect" {
 		errorPayload := ClaudeSSEEvent{Type: "error", Error: errorDetails}
-		// Log event lỗi (shouldLog = true)
 		_ = sendSSEEvent(c, "error", errorPayload, messageID, eventIndex, true, "(OpenAI->Claude SSE)")
-		// eventIndex++ // Không cần tăng nữa vì đây là event cuối cùng trong trường hợp lỗi
 	}
 }
 
-// proxyClaudeNativeSSE directly proxies SSE events from a Claude-native upstream
+// proxyClaudeNativeSSE xử lý SSE từ upstream Claude native và **thêm "hack" usage**
 func proxyClaudeNativeSSE(
 	c *gin.Context,
 	upstreamResp *http.Response,
 	claudeRequestID string,
+	originalClaudeRequest *ClaudeRequest, // Cần để tính input tokens và lấy model
 ) {
 	c.Writer.Header().Set("Content-Type", "text/event-stream")
 	c.Writer.Header().Set("Cache-Control", "no-cache")
 	c.Writer.Header().Set("Connection", "keep-alive")
 	c.Writer.Header().Set("X-Content-Type-Options", "nosniff")
-	// Copy Anthropic-Version header if present in upstream response
 	if anthropicVersion := upstreamResp.Header.Get("anthropic-version"); anthropicVersion != "" {
 		c.Writer.Header().Set("anthropic-version", anthropicVersion)
 	}
 	c.Writer.Flush()
 
-	// Giữ lại: Log INFO quan trọng
-	log.Printf("INFO: [%s] (Claude Native SSE) SSE Proxy: Bắt đầu proxying SSE từ upstream Claude native.", claudeRequestID)
+	messageID := claudeRequestID
+	requestedModel := originalClaudeRequest.Model
+	accumulatedContent := ""
+	streamErrorOccurred := false
+	var errorDetails *ClaudeError
 	startTime := time.Now()
 
-	clientGone := c.Request.Context().Done()
-	scanner := bufio.NewScanner(upstreamResp.Body)
-	defer upstreamResp.Body.Close()
-
-	eventCount := 0
-	lineBuffer := "" // Buffer for multi-line data fields (though rare in Claude SSE)
-
-	for {
-		select {
-		case <-clientGone:
-			// Giữ lại: Log INFO
-			log.Printf("INFO: [%s] (Claude Native SSE) SSE Proxy: Client ngắt kết nối.", claudeRequestID)
-			return
-		default:
-			if !scanner.Scan() {
-				if err := scanner.Err(); err != nil {
-					// Giữ lại: Log ERROR
-					log.Printf("ERROR: [%s] (Claude Native SSE) SSE Proxy: Lỗi đọc từ upstream: %v", claudeRequestID, err)
-					// Optionally send a Claude error event if the protocol allows and client is still connected
-					// For now, just close.
-				} else {
-					// Giữ lại: Log INFO
-					log.Printf("INFO: [%s] (Claude Native SSE) SSE Proxy: Upstream đóng kết nối.", claudeRequestID)
-				}
-				// Giữ lại: Log INFO
-				log.Printf("INFO: [%s] (Claude Native SSE) SSE Proxy: Hoàn thành sau %v, %d events processed.", claudeRequestID, time.Since(startTime), eventCount)
+	// Input tokens được tính từ yêu cầu gốc, giống như khi chuyển đổi từ OpenAI
+	calculatedInputTokens := calculateInputTokensFromClaudeRequest(originalClaudeRequest)
+	log.Printf("INFO: [%s] (Claude Native SSE with Hack) Bắt đầu SSE. Input tokens ước tính: %d. Model: %s", messageID, calculatedInputTokens, requestedModel)
+
+	inputTokens := calculatedInputTokens // inputTokens sẽ không thay đổi trong quá trình stream này
+	outputTokens := 0
+	eventIndex := 0
+
+	var upstreamFinalStopReason string
+	var upstreamFinalUsage *ClaudeUsage
+
+	doneChan := make(chan struct{})
+	errChan := make(chan error, 1)
+
+	// Gửi message_start cho client (tự tạo)
+	startUsage := ClaudeUsage{InputTokens: calculatedInputTokens, OutputTokens: 0}
+	startMessage := ClaudeSSEMessage{ID: messageID, Type: "message", Role: "assistant", Content: []ClaudeContentBlock{}, Model: requestedModel, Usage: startUsage}
+	startEvent := ClaudeSSEEvent{Type: "message_start", Message: &startMessage}
+	if !sendSSEEvent(c, "message_start", startEvent, messageID, eventIndex, true, "(Claude Native SSE with Hack)") {
+		return // Client disconnected
+	}
+	eventIndex++
+
+	// Gửi content_block_start cho client (tự tạo)
+	// Claude native thường gửi content_block_start với index 0 và content_block.type="text", content_block.text=""
+	contentStartBlock := ClaudeSSEContentBlock{Type: "text", Text: ""}
+	contentStartEvent := ClaudeSSEEvent{Type: "content_block_start", Index: func() *int { i := 0; return &i }(), ContentBlock: &contentStartBlock}
+	if !sendSSEEvent(c, "content_block_start", contentStartEvent, messageID, eventIndex, true, "(Claude Native SSE with Hack)") {
+		return // Client disconnected
+	}
+	eventIndex++
+
+	go func() {
+		defer close(doneChan)
+		defer upstreamResp.Body.Close()
+
+		scanner := bufio.NewScanner(upstreamResp.Body)
+		for scanner.Scan() {
+			select {
+			case <-c.Request.Context().Done():
+				log.Printf("INFO: [%s] (Claude Native SSE with Hack) Client ngắt kết nối trong vòng lặp đọc.", messageID)
 				return
+			default:
 			}
 
 			line := scanner.Text()
-			lineBuffer += line + "\n" // Append line to buffer
+			if line == "" { // Skip empty lines between events
+				continue
+			}
 
-			// An empty line signifies end of an event in SSE protocol
-			if line == "" {
-				_, err := fmt.Fprint(c.Writer, lineBuffer) // Write the buffered event (includes the final empty line)
-				if err != nil {
-					// Giữ lại: Log WARN
-					log.Printf("WARN: [%s] (Claude Native SSE) SSE Proxy: Lỗi ghi cho client: %v. Client có thể đã ngắt kết nối.", claudeRequestID, err)
-					return // Stop if we can't write to client
+			// Chỉ xử lý các dòng data, bỏ qua event: lines vì chúng ta sẽ tự tạo/forward có chọn lọc
+			if strings.HasPrefix(line, "data:") {
+				dataStr := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
+				var upstreamEvent ClaudeSSEEvent // Sử dụng struct ClaudeSSEEvent để parse
+				if err := json.Unmarshal([]byte(dataStr), &upstreamEvent); err != nil {
+					log.Printf("WARN: [%s] (Claude Native SSE with Hack) Không thể giải mã chunk JSON Claude native: %v. Data: %s", messageID, err, dataStr)
+					continue
 				}
-				c.Writer.Flush()
-				lineBuffer = "" // Reset buffer for the next event
-				// Vô hiệu hóa: Log TRACE quá chi tiết
-				// log.Printf("TRACE: [%s] (Claude Native SSE) SSE Proxy: Flushed event %d", claudeRequestID, eventCount)
-			}
 
-			// Increment event count when an "event:" line is encountered (heuristic)
-			if strings.HasPrefix(line, "event:") {
-				eventCount++
+				// Xử lý các loại event từ Claude native upstream
+				switch upstreamEvent.Type {
+				case "content_block_delta":
+					if upstreamEvent.Delta != nil && upstreamEvent.Delta.Text != nil {
+						textChunk := *upstreamEvent.Delta.Text
+						accumulatedContent += textChunk
+
+						// Forward content_block_delta đến client
+						// Đảm bảo index được set (thường là 0 cho một content block đơn giản)
+						clientContentDelta := ClaudeSSEEvent{
+							Type:  "content_block_delta",
+							Index: upstreamEvent.Index, // Sử dụng index từ upstream nếu có, nếu không thì 0
+							Delta: &ClaudeSSEDelta{
+								Type: "text_delta",
+								Text: &textChunk,
+							},
+						}
+						if clientContentDelta.Index == nil { clientContentDelta.Index = func() *int { i := 0; return &i }() }
+
+						if !sendSSEEvent(c, "content_block_delta", clientContentDelta, messageID, eventIndex, false, "(Claude Native SSE with Hack)") {
+							return // Client disconnected
+						}
+						eventIndex++
+
+						// "Hack" để gửi message_delta với output_tokens cập nhật
+						currentOutputTokens := estimateTokens(accumulatedContent)
+						if currentOutputTokens != outputTokens {
+							outputTokens = currentOutputTokens
+							intermediateUsage := ClaudeSSEUsage{OutputTokens: outputTokens}
+							intermediateDeltaPayload := ClaudeSSEEvent{
+								Type:  "message_delta",
+								Delta: &ClaudeSSEDelta{}, // Delta rỗng
+								Usage: &intermediateUsage,
+							}
+							if !sendSSEEvent(c, "message_delta", intermediateDeltaPayload, messageID, eventIndex, false, "(Claude Native SSE with Hack)") {
+								return // Client disconnected
+							}
+							eventIndex++
+						}
+					}
+				case "message_delta": // Claude native có thể gửi message_delta với usage
+					if upstreamEvent.Usage != nil {
+						// Cập nhật outputTokens nếu upstream cung cấp
+						// Đây là thông tin chính xác hơn là ước tính
+						if upstreamEvent.Usage.OutputTokens > outputTokens {
+							outputTokens = upstreamEvent.Usage.OutputTokens
+						}
+						// Gửi message_delta "hack" với usage từ upstream (nếu có)
+						// hoặc usage đã tính toán nếu upstream không gửi
+						currentHackUsage := ClaudeSSEUsage{OutputTokens: outputTokens}
+						hackMessageDelta := ClaudeSSEEvent{
+							Type: "message_delta",
+							Delta: &ClaudeSSEDelta{
+								// Claude native message_delta có thể có stop_reason/sequence
+								StopReason: upstreamEvent.Delta.StopReason,
+								StopSequence: upstreamEvent.Delta.StopSequence,
+							},
+							Usage: &currentHackUsage,
+						}
+						if !sendSSEEvent(c, "message_delta", hackMessageDelta, messageID, eventIndex, false, "(Claude Native SSE with Hack - from upstream delta)") {
+							return
+						}
+						eventIndex++
+					}
+					if upstreamEvent.Delta != nil && upstreamEvent.Delta.StopReason != nil {
+						upstreamFinalStopReason = *upstreamEvent.Delta.StopReason
+					}
+
+				case "message_stop":
+					log.Printf("INFO: [%s] (Claude Native SSE with Hack) Nhận message_stop từ upstream.", messageID)
+					if upstreamEvent.Message != nil && upstreamEvent.Message.StopReason != nil { // Older format
+						upstreamFinalStopReason = *upstreamEvent.Message.StopReason
+					}
+					// Newer Claude API sends stop_reason in message_delta, and message_stop has final usage.
+					if upstreamEvent.Usage != nil { // Final usage from upstream
+						upstreamFinalUsage = upstreamEvent.Usage
+						if upstreamEvent.Usage.OutputTokens > outputTokens { // Prefer upstream's final count
+							outputTokens = upstreamEvent.Usage.OutputTokens
+						}
+					}
+					return // Kết thúc vòng lặp khi nhận message_stop
+
+				case "error":
+					log.Printf("ERROR: [%s] (Claude Native SSE with Hack) Nhận lỗi từ upstream: %+v", messageID, upstreamEvent.Error)
+					streamErrorOccurred = true
+					errorDetails = upstreamEvent.Error
+					return // Kết thúc vòng lặp
+				
+				// Các event khác như message_start, content_block_start, content_block_stop từ upstream sẽ được bỏ qua
+				// vì chúng ta đã tự tạo các event tương ứng cho client ở đầu luồng.
+				// Ping events có thể được forward nếu cần.
+				case "ping":
+					pingEvent := ClaudeSSEEvent{Type: "ping"}
+					if !sendSSEEvent(c, "ping", pingEvent, messageID, eventIndex, false, "(Claude Native SSE with Hack - ping)") {
+						return
+					}
+					eventIndex++
+
+				default:
+					// log.Printf("DEBUG: [%s] (Claude Native SSE with Hack) Bỏ qua event upstream không xác định: %s", messageID, upstreamEvent.Type)
+				}
 			}
 		}
+		if err := scanner.Err(); err != nil {
+			select {
+			case <-c.Request.Context().Done():
+				log.Printf("INFO: [%s] (Claude Native SSE with Hack) Đọc upstream bị gián đoạn bởi client ngắt kết nối: %v", messageID, c.Request.Context().Err())
+			default:
+				log.Printf("ERROR: [%s] (Claude Native SSE with Hack) Lỗi đọc nội dung phản hồi upstream: %v", messageID, err)
+				errChan <- fmt.Errorf("lỗi đọc upstream: %w", err)
+			}
+		}
+	}()
+
+	select {
+	case <-doneChan:
+		// Normal completion of upstream read
+	case err := <-errChan:
+		log.Printf("ERROR: [%s] (Claude Native SSE with Hack) Nhận lỗi từ goroutine đọc: %v", messageID, err)
+		streamErrorOccurred = true
+		if errorDetails == nil { // Ensure errorDetails is set
+			errorDetails = &ClaudeError{Type: "api_error", Message: fmt.Sprintf("Lỗi đọc phản hồi upstream: %v", err)}
+		}
+	case <-c.Request.Context().Done():
+		log.Printf("INFO: [%s] (Claude Native SSE with Hack) Client ngắt kết nối trong quá trình xử lý stream: %v", messageID, c.Request.Context().Err())
+		streamErrorOccurred = true
+		if errorDetails == nil { // Ensure errorDetails is set
+			errorDetails = &ClaudeError{Type: "client_disconnect", Message: "Client ngắt kết nối trong quá trình stream"}
+		}
+	}
+
+	var finalStopReasonClient string
+	if streamErrorOccurred && errorDetails != nil && errorDetails.Type == "client_disconnect" {
+		finalStopReasonClient = "client_disconnect"
+	} else if streamErrorOccurred {
+		finalStopReasonClient = "error"
+	} else if upstreamFinalStopReason != "" {
+		finalStopReasonClient = upstreamFinalStopReason
+	} else {
+		finalStopReasonClient = "end_turn" // Default if no specific reason from upstream
+	}
+
+	// Xác định finalOutputTokens
+	finalOutputTokens := outputTokens
+	if upstreamFinalUsage != nil && upstreamFinalUsage.OutputTokens > finalOutputTokens {
+		finalOutputTokens = upstreamFinalUsage.OutputTokens // Ưu tiên usage từ message_stop của upstream
+	} else {
+		finalOutputTokens = estimateTokens(accumulatedContent) // Nếu không có, ước tính lại
+	}
+	finalOutputTokens = max(0, finalOutputTokens)
+
+
+	log.Printf("INFO: [%s] (Claude Native SSE with Hack) SSE Stream hoàn tất. Lý do dừng: %s. Input: %d, Output: %d. Thời gian: %v. Upstream Stop Reason: %s",
+		messageID, finalStopReasonClient, calculatedInputTokens, finalOutputTokens, time.Since(startTime), upstreamFinalStopReason)
+
+	// Gửi "Hack" message_delta cuối cùng với stop_reason và usage (chỉ output_tokens)
+	finalHackUsageData := ClaudeSSEUsage{OutputTokens: finalOutputTokens}
+	finalDeltaStopReason := finalStopReasonClient
+	priorityFinalDeltaPayload := ClaudeSSEEvent{
+		Type: "message_delta",
+		Delta: &ClaudeSSEDelta{
+			StopReason:   &finalDeltaStopReason,
+			StopSequence: nil, // Claude native không thường gửi stop_sequence trong delta này
+		},
+		Usage: &finalHackUsageData,
+	}
+	_ = sendSSEEvent(c, "message_delta", priorityFinalDeltaPayload, messageID, eventIndex, false, "(Claude Native SSE with Hack)")
+	eventIndex++
+
+	// Gửi content_block_stop cho client
+	contentStopPayload := ClaudeSSEEvent{Type: "content_block_stop", Index: func() *int { i := 0; return &i }()}
+	_ = sendSSEEvent(c, "content_block_stop", contentStopPayload, messageID, eventIndex, false, "(Claude Native SSE with Hack)")
+	eventIndex++
+
+	// Gửi message_stop cho client với usage đầy đủ
+	// Input tokens là cái đã tính toán, output tokens là cái đã chốt cuối cùng
+	finalClientInputTokens := calculatedInputTokens
+	finalStopUsageData := ClaudeSSEUsage{InputTokens: &finalClientInputTokens, OutputTokens: finalOutputTokens}
+	messageStopPayload := ClaudeSSEEvent{Type: "message_stop", Usage: &finalStopUsageData}
+	_ = sendSSEEvent(c, "message_stop", messageStopPayload, messageID, eventIndex, true, "(Claude Native SSE with Hack)")
+	eventIndex++
+
+	if streamErrorOccurred && errorDetails != nil && errorDetails.Type != "client_disconnect" {
+		// Nếu lỗi không phải do client disconnect, gửi event error cho client
+		errorPayload := ClaudeSSEEvent{Type: "error", Error: errorDetails}
+		_ = sendSSEEvent(c, "error", errorPayload, messageID, eventIndex, true, "(Claude Native SSE with Hack)")
 	}
 }
 
@@ -330,10 +508,11 @@ func sendSSEEvent(c *gin.Context, eventName string, data interface{}, requestID
 			log.Printf("ERROR: [%s] %s Không thể marshal SSE event %d (%s): %v", requestID, logPrefix, eventIndex, eventName, err)
 			return true // Continue trying to send other events if this one fails to marshal
 		}
+		// Đảm bảo có một dòng trống sau mỗi sự kiện data:
 		_, err = fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", eventName, string(jsonData))
 		if err != nil {
 			if shouldLog || eventName == "message_stop" || eventName == "error" {
-				log.Printf("WARN: [%s] %s Không thể ghi SSE event %d (%s) cho client: %v. Client có thể đã ngắt kết nối.", requestID, logPrefix, eventIndex, eventName, err)
+				log.Printf("WARN: [%s] %s Không thể ghi SSE event %d (%s) cho client: %v. Client có thể đã ngắt kết nối.", requestID, logPrefix, eventIndex, err)
 			}
 			return false // Stop if writing to client fails
 		}