cline/cli/pkg/cli/display/streaming.go

package display

import (
	"encoding/json"
	"fmt"
	"strings"
	"sync"

	"github.com/cline/cli/pkg/cli/types"
)

// StreamingDisplay manages streaming message display with deduplication
type StreamingDisplay struct {
	mu       sync.RWMutex
	state    *types.ConversationState
	renderer *Renderer
	dedupe   *MessageDeduplicator
}

// NewStreamingDisplay creates a new streaming display manager
func NewStreamingDisplay(state *types.ConversationState, renderer *Renderer) *StreamingDisplay {
	return &StreamingDisplay{
		state:    state,
		renderer: renderer,
		dedupe:   NewMessageDeduplicator(),
	}
}

// HandlePartialMessage processes partial messages with streaming support
func (s *StreamingDisplay) HandlePartialMessage(msg *types.ClineMessage) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	messageKey := fmt.Sprintf("%d", msg.Timestamp)
	timestamp := msg.GetTimestamp()

	// Check for deduplication
	if s.dedupe.IsDuplicate(msg) {
		return nil
	}

	// Get current streaming state
	streamingMsg := s.state.GetStreamingMessage()

	switch msg.Type {
	case types.MessageTypeAsk:
		return s.handleStreamingAsk(msg, messageKey, timestamp, streamingMsg)
	case types.MessageTypeSay:
		return s.handleStreamingSay(msg, messageKey, timestamp, streamingMsg)
	default:
		return s.renderer.RenderMessage(timestamp, "🤖", msg.Text)
	}
}

// handleStreamingAsk handles streaming ASK messages
func (s *StreamingDisplay) handleStreamingAsk(msg *types.ClineMessage, messageKey, timestamp string, streamingMsg *types.StreamingMessage) error {
	if msg.Text == "" {
		return nil
	}

	cleanText := s.renderer.sanitizeText(msg.Text)
	if cleanText == "" {
		return nil
	}

	// Check if this is an update to the same ASK message
	if streamingMsg.CurrentKey == messageKey {
		// This is an update to the same ASK message - stream the changes
		if cleanText != streamingMsg.LastText {
			s.streamAskMessageUpdate(cleanText, streamingMsg.LastText, timestamp)
			s.state.SetStreamingMessage(messageKey, cleanText)
		}
	} else {
		// This is a new ASK message
		s.finishCurrentStream()
		s.streamAskMessage(cleanText, timestamp, true)
		s.state.SetStreamingMessage(messageKey, cleanText)
	}

	return nil
}

// handleStreamingSay handles streaming SAY messages
func (s *StreamingDisplay) handleStreamingSay(msg *types.ClineMessage, messageKey, timestamp string, streamingMsg *types.StreamingMessage) error {
	switch msg.Say {
	case string(types.SayTypeText), string(types.SayTypeCompletionResult):
		return s.handleStreamingText(msg, messageKey, timestamp, streamingMsg)
	case string(types.SayTypeCommand):
		return s.handleStreamingCommand(msg, messageKey, timestamp, streamingMsg)
	case string(types.SayTypeCommandOutput):
		return s.handleStreamingCommandOutput(msg, messageKey, timestamp, streamingMsg)
	case string(types.SayTypeTool):
		return s.handleStreamingTool(msg, messageKey, timestamp, streamingMsg)
	case string(types.SayTypeShellIntegrationWarning):
		return s.handleShellIntegrationWarning(msg, messageKey, timestamp, streamingMsg)
	default:
		// For non-streaming message types, use regular display
		return s.renderer.RenderMessage(timestamp, s.getMessagePrefix(msg.Say), msg.Text)
	}
}

// handleStreamingText handles streaming text messages
func (s *StreamingDisplay) handleStreamingText(msg *types.ClineMessage, messageKey, timestamp string, streamingMsg *types.StreamingMessage) error {
	cleanText := s.renderer.sanitizeText(msg.Text)
	if cleanText == "" {
		return nil
	}

	// Check if we've already displayed this exact message
	if streamingMsg.CurrentKey == messageKey && streamingMsg.LastText == cleanText {
		return nil // Duplicate - ignore it
	}

	// Check if this is an update to the same message
	if streamingMsg.CurrentKey == messageKey {
		// Show incremental changes
		if len(cleanText) > len(streamingMsg.LastText) && strings.HasPrefix(cleanText, streamingMsg.LastText) {
			// Show only the new characters with typewriter effect
			newChars := cleanText[len(streamingMsg.LastText):]
			s.typewriterPrint(newChars)
			s.state.SetStreamingMessage(messageKey, cleanText)
		} else {
			// Text changed in a non-incremental way - replace the line
			s.renderer.ClearLine()
			prefix := s.getMessagePrefix(msg.Say)
			s.renderer.typewriter.PrintfInstant("[%s] %s: ", timestamp, prefix)
			s.typewriterPrint(cleanText)
			s.state.SetStreamingMessage(messageKey, cleanText)
		}
	} else {
		// This is a new message
		s.finishCurrentStream()
		prefix := s.getMessagePrefix(msg.Say)
		s.renderer.typewriter.PrintfInstant("[%s] %s: ", timestamp, prefix)

		// Add typewriter animation for new messages
		s.typewriterPrint(cleanText)

		s.state.SetStreamingMessage(messageKey, cleanText)
	}

	// If message is complete, add newline
	if !msg.Partial {
		fmt.Println()
		s.state.SetStreamingMessage("", "")
	}

	return nil
}

// handleStreamingCommand handles command execution messages
func (s *StreamingDisplay) handleStreamingCommand(msg *types.ClineMessage, messageKey, timestamp string, streamingMsg *types.StreamingMessage) error {
	cleanText := s.renderer.sanitizeText(msg.Text)
	if cleanText == "" {
		return nil
	}

	// Show command being executed with typewriter effect
	s.finishCurrentStream()
	s.renderer.typewriter.PrintfInstant("[%s] 🖥️  CMD: ", timestamp)
	s.typewriterPrint(cleanText)
	fmt.Println()

	return nil
}

// handleStreamingCommandOutput handles streaming command output
func (s *StreamingDisplay) handleStreamingCommandOutput(msg *types.ClineMessage, messageKey, timestamp string, streamingMsg *types.StreamingMessage) error {
	cleanText := s.renderer.sanitizeText(msg.Text)
	if cleanText == "" {
		return nil
	}

	// Check if we've already displayed this exact message
	if streamingMsg.CurrentKey == messageKey && streamingMsg.LastText == cleanText {
		return nil
	}

	// Check if this is an update to the same message
	if streamingMsg.CurrentKey == messageKey {
		// Show incremental changes with typewriter effect
		if len(cleanText) > len(streamingMsg.LastText) && strings.HasPrefix(cleanText, streamingMsg.LastText) {
			newChars := cleanText[len(streamingMsg.LastText):]
			s.typewriterPrint(newChars)
			s.state.SetStreamingMessage(messageKey, cleanText)
		} else {
			// Non-incremental change - replace the line
			s.renderer.ClearLine()
			s.renderer.typewriter.PrintfInstant("[%s] 🖥️  OUT: ", timestamp)
			s.typewriterPrint(cleanText)
			s.state.SetStreamingMessage(messageKey, cleanText)
		}
	} else {
		// New command output message
		s.finishCurrentStream()
		s.renderer.typewriter.PrintfInstant("[%s] 🖥️  OUT: ", timestamp)
		s.typewriterPrint(cleanText)
		s.state.SetStreamingMessage(messageKey, cleanText)
	}

	// If message is complete, add newline
	if !msg.Partial {
		fmt.Println()
		s.state.SetStreamingMessage("", "")
	}

	return nil
}

// handleShellIntegrationWarning handles shell integration warning messages
func (s *StreamingDisplay) handleShellIntegrationWarning(msg *types.ClineMessage, messageKey, timestamp string, streamingMsg *types.StreamingMessage) error {
	cleanText := s.renderer.sanitizeText(msg.Text)
	if cleanText == "" {
		return nil
	}

	// Show a more concise shell integration warning
	s.finishCurrentStream()
	s.renderer.typewriter.PrintfInstant("[%s] ℹ️  NOTE: ", timestamp)
	s.typewriterPrint("Command executed (output not streamed due to shell integration)")
	fmt.Println()

	return nil
}

// handleStreamingTool handles streaming tool messages with deduplication
func (s *StreamingDisplay) handleStreamingTool(msg *types.ClineMessage, messageKey, timestamp string, streamingMsg *types.StreamingMessage) error {
	cleanText := s.renderer.sanitizeText(msg.Text)
	if cleanText == "" {
		return nil
	}

	formattedTool := s.formatToolMessage(cleanText)

	// Check if this is the exact same tool message we just displayed
	if streamingMsg.LastToolMessage == formattedTool {
		return nil // Exact duplicate - ignore it
	}

	// Check if this is a very similar tool message
	if streamingMsg.LastToolMessage != "" && s.isSimilarToolMessage(streamingMsg.LastToolMessage, formattedTool) {
		return nil // Similar duplicate - ignore it
	}

	// This is a genuinely new/different tool message
	s.finishCurrentStream()
	fmt.Printf("[%s] 🔧 TOOL: %s\n", timestamp, formattedTool)

	// Store the formatted tool message for deduplication
	s.state.StreamingMessage.LastToolMessage = formattedTool

	return nil
}

// streamAskMessage streams an ASK message in a natural format
func (s *StreamingDisplay) streamAskMessage(text, timestamp string, isNew bool) {
	// Try to parse as JSON
	var askData types.AskData
	if err := s.parseJSON(text, &askData); err != nil {
		// Display as text but sanitized
		fmt.Printf("[%s] 🤖 ASK: %s", timestamp, text)
		return
	}

	fmt.Printf("[%s] 🤖 ASK: %s", timestamp, askData.Response)

	// Display options if available
	if len(askData.Options) > 0 {
		fmt.Print("\n\nOptions:")
		for i, option := range askData.Options {
			fmt.Printf("\n%d. %s", i+1, option)
		}
	}
}

// streamAskMessageUpdate handles updates to an existing ASK message
func (s *StreamingDisplay) streamAskMessageUpdate(newText, oldText, timestamp string) {
	var oldAskData, newAskData types.AskData

	oldErr := s.parseJSON(oldText, &oldAskData)
	newErr := s.parseJSON(newText, &newAskData)

	if oldErr != nil || newErr != nil {
		// Handle plain text incremental updates
		if len(newText) > len(oldText) && strings.HasPrefix(newText, oldText) {
			newChars := newText[len(oldText):]
			fmt.Print(newChars)
		} else {
			// Non-incremental change - clear line and reprint everything
			s.renderer.ClearLine()
			fmt.Printf("[%s] 🤖 ASK: %s", timestamp, newText)
		}
		return
	}

	// Handle structured updates
	if len(newAskData.Response) > len(oldAskData.Response) && strings.HasPrefix(newAskData.Response, oldAskData.Response) {
		newChars := newAskData.Response[len(oldAskData.Response):]
		fmt.Print(newChars)
	} else if oldAskData.Response != newAskData.Response {
		s.renderer.ClearLine()
		fmt.Printf("[%s] 🤖 ASK: %s", timestamp, newAskData.Response)
	}

	// Handle options changes
	if len(newAskData.Options) > len(oldAskData.Options) {
		if len(oldAskData.Options) == 0 {
			fmt.Print("\n\nOptions:")
		}

		for i := len(oldAskData.Options); i < len(newAskData.Options); i++ {
			fmt.Printf("\n%d. %s", i+1, newAskData.Options[i])
		}
	}
}

// typewriterPrint displays text with a typewriter animation effect
func (s *StreamingDisplay) typewriterPrint(text string) {
	// Use the renderer's typewriter for consistent animation
	s.renderer.typewriter.Print(text)
}

// finishCurrentStream completes any ongoing streaming message
func (s *StreamingDisplay) finishCurrentStream() {
	streamingMsg := s.state.GetStreamingMessage()
	if streamingMsg.CurrentKey != "" {
		//fmt.Println() // Add newline to finish the current streaming message
		s.state.SetStreamingMessage("", "")
	}
}

// getMessagePrefix returns the appropriate prefix for a message type
func (s *StreamingDisplay) getMessagePrefix(say string) string {
	switch say {
	case string(types.SayTypeCompletionResult):
		return "✅ RESULT"
	case string(types.SayTypeText):
		return "🤖"
	default:
		return "🤖"
	}
}

// formatToolMessage formats tool call messages for better readability
func (s *StreamingDisplay) formatToolMessage(text string) string {
	var toolCall map[string]interface{}
	if err := s.parseJSON(text, &toolCall); err == nil {
		if tool, ok := toolCall["tool"].(string); ok {
			parts := []string{tool}

			if path, ok := toolCall["path"].(string); ok && path != "" {
				parts = append(parts, fmt.Sprintf("path=%s", path))
			}

			if content, ok := toolCall["content"].(string); ok && content != "" {
				if len(content) > 50 {
					parts = append(parts, fmt.Sprintf("content=%s...", content[:50]))
				} else {
					parts = append(parts, fmt.Sprintf("content=%s", content))
				}
			}

			return strings.Join(parts, " ")
		}
	}

	// If not JSON or doesn't have expected structure, return truncated
	if len(text) > 100 {
		return text[:100] + "..."
	}
	return text
}

// isSimilarToolMessage checks if two tool messages are similar enough to be considered duplicates
func (s *StreamingDisplay) isSimilarToolMessage(msg1, msg2 string) bool {
	parts1 := strings.Fields(msg1)
	parts2 := strings.Fields(msg2)

	if len(parts1) == 0 || len(parts2) == 0 {
		return false
	}

	// If the first word (tool name) is the same, check for similarity
	if parts1[0] == parts2[0] {
		// For file operations, check if the path is the same
		if strings.Contains(msg1, "path=") && strings.Contains(msg2, "path=") {
			path1 := s.extractPathFromToolMessage(msg1)
			path2 := s.extractPathFromToolMessage(msg2)

			if path1 != "" && path1 == path2 {
				return true
			}
		}

		// For very similar content (>80% similarity), consider them duplicates
		similarity := s.calculateStringSimilarity(msg1, msg2)
		return similarity > 0.8
	}

	return false
}

// extractPathFromToolMessage extracts the path parameter from a tool message
func (s *StreamingDisplay) extractPathFromToolMessage(msg string) string {
	parts := strings.Fields(msg)
	for _, part := range parts {
		if strings.HasPrefix(part, "path=") {
			return strings.TrimPrefix(part, "path=")
		}
	}
	return ""
}

// calculateStringSimilarity calculates a simple similarity ratio between two strings
func (s *StreamingDisplay) calculateStringSimilarity(s1, s2 string) float64 {
	if s1 == s2 {
		return 1.0
	}

	if len(s1) == 0 || len(s2) == 0 {
		return 0.0
	}

	shorter, longer := s1, s2
	if len(s1) > len(s2) {
		shorter, longer = s2, s1
	}

	matches := 0
	for i, r := range shorter {
		if i < len(longer) && rune(longer[i]) == r {
			matches++
		}
	}

	return float64(matches) / float64(len(longer))
}

// parseJSON is a helper function to parse JSON with error handling
func (s *StreamingDisplay) parseJSON(text string, v interface{}) error {
	return json.Unmarshal([]byte(text), v)
}

// Cleanup cleans up streaming display resources
func (s *StreamingDisplay) Cleanup() {
	if s.dedupe != nil {
		s.dedupe.Stop()
	}
}