Refactor message compaction and legacy migration handling

- Removed unnecessary logging statements in `compaction.go` to streamline message processing and improve clarity.
- Updated `MigrateLegacySessionToAGUI` to handle legacy messages in JSONL format, enhancing compatibility with existing data.
- Improved error handling and logging during legacy message migration, ensuring better traceability and user feedback.

These changes enhance the efficiency of message compaction and improve the robustness of legacy data handling.

Author: Gkrumbach07

components/backend/websocket/compaction.go

@@ -57,16 +57,12 @@ func (c *MessageCompactor) HandleEvent(event map[string]interface{}) {
 		c.handleMessagesSnapshot(event)
 	case types.EventTypeRunStarted, types.EventTypeRunFinished, types.EventTypeRunError:
 		// Lifecycle events - skip, don't affect message compaction
-		log.Printf("Compaction: Skipping lifecycle event %s", eventType)
 	case types.EventTypeStepStarted, types.EventTypeStepFinished:
 		// Step events - skip, don't affect message compaction
-		log.Printf("Compaction: Skipping step event %s", eventType)
 	case types.EventTypeStateSnapshot, types.EventTypStateDelta:
 		// State events - skip, don't affect message compaction
-		log.Printf("Compaction: Skipping state event %s", eventType)
 	case types.EventTypeActivitySnapshot, types.EventTypeActivityDelta:
 		// Activity events - skip, don't affect message compaction
-		log.Printf("Compaction: Skipping activity event %s", eventType)
 	default:
 		log.Printf("Compaction: WARNING - Unhandled event type: %s", eventType)
 	}
@@ -87,9 +83,6 @@ func (c *MessageCompactor) GetMessages() []types.Message {
 	// If we included "running" status tools here, they would duplicate when
 	// the active run's TOOL_CALL_END events are replayed.
 	if len(c.activeToolCalls) > 0 {
-		log.Printf("Compaction: WARNING - %d tool calls never completed (missing TOOL_CALL_END)", len(c.activeToolCalls))
-		log.Printf("Compaction: This indicates events from an active/incomplete run were included in compaction")
-		log.Printf("Compaction: Discarding incomplete tool calls to prevent duplicates")
 		// Clear activeToolCalls - don't include them in snapshot
 		c.activeToolCalls = make(map[string]*ActiveToolCall)
 	}
@@ -99,17 +92,12 @@ func (c *MessageCompactor) GetMessages() []types.Message {
 	hiddenCount := 0
 	for _, msg := range c.messages {
 		if c.hiddenMessages[msg.ID] {
-			log.Printf("Compaction: Filtering hidden message %s from snapshot", msg.ID[:8])
 			hiddenCount++
 			continue
 		}
 		visibleMessages = append(visibleMessages, msg)
 	}
 
-	if hiddenCount > 0 {
-		log.Printf("Compaction: Filtered %d hidden messages, returning %d visible", hiddenCount, len(visibleMessages))
-	}
-
 	return visibleMessages
 }
 
@@ -118,7 +106,6 @@ func (c *MessageCompactor) GetMessages() []types.Message {
 func (c *MessageCompactor) handleTextMessageStart(event map[string]interface{}) {
 	// Flush previous message if any
 	if c.currentMessage != nil {
-		log.Printf("Compaction: Flushing previous message (id=%s, content=%d chars)", c.currentMessage.ID, len(c.currentMessage.Content))
 		c.messages = append(c.messages, *c.currentMessage)
 	}
 
@@ -137,34 +124,26 @@ func (c *MessageCompactor) handleTextMessageStart(event map[string]interface{})
 		Role:    role,
 		Content: "",
 	}
-	log.Printf("Compaction: TEXT_MESSAGE_START role=%s, messageId=%s", role, messageID)
 }
 
 func (c *MessageCompactor) handleTextMessageContent(event map[string]interface{}) {
 	if c.currentMessage == nil {
-		log.Printf("Compaction: WARNING - TEXT_MESSAGE_CONTENT without START")
 		return
 	}
 
 	delta, _ := event["delta"].(string)
 	c.currentMessage.Content += delta
-	log.Printf("Compaction: TEXT_MESSAGE_CONTENT delta=%d chars, total=%d chars", len(delta), len(c.currentMessage.Content))
 }
 
 func (c *MessageCompactor) handleTextMessageEnd(event map[string]interface{}) {
 	if c.currentMessage != nil {
 		// User messages never have tool calls - flush immediately
+		// Assistant messages might have tool calls - keep open
+		// We'll flush when a new TEXT_MESSAGE_START arrives or at the end of compaction
 		if c.currentMessage.Role == types.RoleUser {
-			log.Printf("Compaction: TEXT_MESSAGE_END (user) id=%s, flushing immediately", c.currentMessage.ID)
 			c.messages = append(c.messages, *c.currentMessage)
 			c.currentMessage = nil
-		} else {
-			// Assistant messages might have tool calls - keep open
-			log.Printf("Compaction: TEXT_MESSAGE_END id=%s, content=%d chars (keeping message open for tool calls)", c.currentMessage.ID, len(c.currentMessage.Content))
-			// We'll flush when a new TEXT_MESSAGE_START arrives or at the end of compaction
 		}
-	} else {
-		log.Printf("Compaction: WARNING - TEXT_MESSAGE_END without current message")
 	}
 }
 
@@ -196,13 +175,6 @@ func (c *MessageCompactor) handleToolCallStart(event map[string]interface{}) {
 			ParentToolUseID: parentToolUseID,
 			Status:          "running",
 		}
-		if parentToolUseID != "" {
-			log.Printf("Compaction: Started tool %s (%s) with parent %s", toolName, toolID[:8], parentToolUseID[:8])
-		} else {
-			log.Printf("Compaction: Started tool %s (%s)", toolName, toolID[:8])
-		}
-	} else {
-		log.Printf("Compaction: WARNING - TOOL_CALL_START without toolCallId")
 	}
 }
 
@@ -215,15 +187,11 @@ func (c *MessageCompactor) handleToolCallArgs(event map[string]interface{}) {
 	delta, _ := event["delta"].(string)
 
 	if toolID == "" {
-		log.Printf("Compaction: WARNING - TOOL_CALL_ARGS without toolCallId")
 		return
 	}
 
 	if active, ok := c.activeToolCalls[toolID]; ok {
 		active.Args += delta
-		log.Printf("Compaction: Accumulated args for %s: %d chars", active.Name, len(active.Args))
-	} else {
-		log.Printf("Compaction: WARNING - TOOL_CALL_ARGS for unknown tool %s", toolID[:8])
 	}
 }
 
@@ -237,13 +205,11 @@ func (c *MessageCompactor) handleToolCallEnd(event map[string]interface{}) {
 	errorStr, _ := event["error"].(string)
 
 	if toolID == "" {
-		log.Printf("Compaction: WARNING - TOOL_CALL_END without toolCallId")
 		return
 	}
 
 	active, ok := c.activeToolCalls[toolID]
 	if !ok {
-		log.Printf("Compaction: WARNING - TOOL_CALL_END for unknown tool %s", toolID[:8])
 		return
 	}
 
@@ -262,9 +228,6 @@ func (c *MessageCompactor) handleToolCallEnd(event map[string]interface{}) {
 		tc.Status = "error"
 	}
 
-	log.Printf("Compaction: Completed tool %s (%s), args=%d chars, result=%d chars",
-		tc.Name, tc.ID[:8], len(tc.Args), len(tc.Result))
-
 	// Add to message
 	// Check if we need to create a new message or add to current
 	if c.currentMessage != nil && c.currentMessage.Role == types.RoleAssistant {
@@ -299,7 +262,6 @@ func (c *MessageCompactor) handleRawEvent(event map[string]interface{}) {
 		if hidden, _ := data["hidden"].(bool); hidden {
 			if messageID, ok := data["messageId"].(string); ok {
 				c.hiddenMessages[messageID] = true
-				log.Printf("Compaction: Marking message %s as hidden", messageID[:8])
 			}
 		}
 		return
@@ -404,48 +366,25 @@ func (c *MessageCompactor) handleMessagesSnapshot(event map[string]interface{})
 		c.messages = append(c.messages, msg)
 	}
 
-	log.Printf("Compaction: Received MESSAGES_SNAPSHOT with %d messages from runner", len(c.messages))
 }
 
 // CompactEvents is the main entry point for event compaction
 func CompactEvents(events []map[string]interface{}) []types.Message {
-	log.Printf("Compaction: Starting compaction of %d events", len(events))
 
 	// Count event types to help debug
 	eventTypeCounts := make(map[string]int)
 	for _, event := range events {
 		eventType, _ := event["type"].(string)
 		eventTypeCounts[eventType]++
 	}
-	log.Printf("Compaction: Event type breakdown: %v", eventTypeCounts)
 
 	compactor := NewMessageCompactor()
 
-	for i, event := range events {
-		eventType, _ := event["type"].(string)
-		if i < 10 || i >= len(events)-10 {
-			// Log first and last 10 events in detail
-			log.Printf("Compaction: [%d/%d] Processing event type=%s", i+1, len(events), eventType)
-		}
+	for _, event := range events {
 		compactor.HandleEvent(event)
 	}
 
 	messages := compactor.GetMessages()
-	log.Printf("Compaction: Finished compaction - produced %d messages from %d events", len(messages), len(events))
-
-	// Log summary of messages
-	for i, msg := range messages {
-		toolCallCount := len(msg.ToolCalls)
-		log.Printf("Compaction:   Message[%d]: role=%s, id=%s, content=%d chars, toolCalls=%d",
-			i, msg.Role, msg.ID, len(msg.Content), toolCallCount)
-	}
-
-	if len(messages) == 0 && len(events) > 0 {
-		log.Printf("Compaction: WARNING - 0 messages produced! This indicates:")
-		log.Printf("Compaction:   - No TEXT_MESSAGE_START events found, OR")
-		log.Printf("Compaction:   - No MESSAGES_SNAPSHOT events found, OR")
-		log.Printf("Compaction:   - Only lifecycle events (which are skipped)")
-	}
 
 	return messages
 }

components/backend/websocket/legacy_translator.go

@@ -13,8 +13,8 @@ import (
 // MigrateLegacySessionToAGUI converts old message format to AG-UI events
 // Creates a MESSAGES_SNAPSHOT from legacy messages and persists it
 func MigrateLegacySessionToAGUI(sessionID string) error {
-	// Check if session has legacy messages
-	legacyPath := StateBaseDir + "/sessions/" + sessionID + "/messages.json"
+	// Check if session has legacy messages (JSONL format)
+	legacyPath := StateBaseDir + "/sessions/" + sessionID + "/messages.jsonl"
 	data, err := os.ReadFile(legacyPath)
 	if err != nil {
 		if os.IsNotExist(err) {
@@ -24,25 +24,28 @@ func MigrateLegacySessionToAGUI(sessionID string) error {
 		return err
 	}
 
-	log.Printf("LegacyMigration: Found legacy messages.json for %s, converting to AG-UI", sessionID)
+	log.Printf("LegacyMigration: Found legacy messages.jsonl for %s, converting to AG-UI", sessionID)
 
-	var legacyData struct {
-		Messages  []map[string]interface{} `json:"messages"`
-		SessionID string                   `json:"sessionId"`
-	}
-
-	if err := json.Unmarshal(data, &legacyData); err != nil {
-		log.Printf("LegacyMigration: Failed to parse legacy messages: %v", err)
-		return err
+	// Parse JSONL - each line is a complete message
+	var legacyMessages []map[string]interface{}
+	lines := splitLines(data)
+	for _, line := range lines {
+		if len(line) == 0 {
+			continue
+		}
+		var msg map[string]interface{}
+		if err := json.Unmarshal(line, &msg); err == nil {
+			legacyMessages = append(legacyMessages, msg)
+		}
 	}
 
 	// Convert to AG-UI Message format
 	messages := make([]types.Message, 0)
-	
-	for _, legacyMsg := range legacyData.Messages {
+
+	for _, legacyMsg := range legacyMessages {
 		msgType, _ := legacyMsg["type"].(string)
 		payload, _ := legacyMsg["payload"].(map[string]interface{})
-		
+
 		switch msgType {
 		case "user_message":
 			content, _ := payload["content"].(string)
@@ -66,8 +69,8 @@ func MigrateLegacySessionToAGUI(sessionID string) error {
 				}
 			}
 			// Tool calls will be reconstructed from tool_result pairs
-			
-		// system.message, agent.running, agent.waiting are not chat messages, skip
+
+			// system.message, agent.running, agent.waiting are not chat messages, skip
 		}
 	}
 
@@ -96,6 +99,8 @@ func MigrateLegacySessionToAGUI(sessionID string) error {
 	migratedPath := legacyPath + ".migrated"
 	if err := os.Rename(legacyPath, migratedPath); err != nil {
 		log.Printf("LegacyMigration: Warning - failed to rename legacy file: %v", err)
+	} else {
+		log.Printf("LegacyMigration: Renamed %s to %s", legacyPath, migratedPath)
 	}
 
 	return nil
@@ -107,4 +112,3 @@ func generateEventID() string {
 	rand.Read(b)
 	return hex.EncodeToString(b)
 }
-