hermes-agent/agent/session_compaction.py

"""
Session Compaction with Fact Extraction — #748

Before compressing a long conversation, extracts durable facts
(user preferences, corrections, project details) and saves them
to the fact store. Then compresses the conversation.

This ensures key information survives context limits.

Usage:
    from agent.session_compaction import compact_session
    
    # In the conversation loop, when context is near limit:
    compact_session(messages, fact_store)
"""

import json
import re
from typing import Any, Dict, List, Optional, Tuple

# ---------------------------------------------------------------------------
# Fact Extraction Patterns
# ---------------------------------------------------------------------------

# Patterns that indicate durable facts worth preserving
_FACT_PATTERNS = [
    # User preferences
    (r"(?:i prefer|i like|i always|my preference is|remember that i)\s+(.+?)(?:\.|$)", "user_pref"),
    (r"(?:call me|my name is|i\'m)\s+([A-Z][a-z]+)", "user_name"),
    (r"(?:don\'t|do not|never)\s+(?:use|do|show|tell)\s+(.+?)(?:\.|$)", "user_constraint"),
    
    # Corrections
    (r"(?:actually|no,?|correction:?)\s+(.+?)(?:\.|$)", "correction"),
    (r"(?:that\'s wrong|not correct|i meant)\s+(.+?)(?:\.|$)", "correction"),
    
    # Project facts
    (r"(?:the project|this repo|the codebase)\s+(?:is|has|uses|runs)\s+(.+?)(?:\.|$)", "project_fact"),
    (r"(?:we use|our stack is|deployed on)\s+(.+?)(?:\.|$)", "project_fact"),
    
    # Technical facts
    (r"(?:the server|the service|the endpoint)\s+(?:is|runs on|listens on)\s+(.+?)(?:\.|$)", "technical"),
    (r"(?:port|url|address|host)\s*(?::|is|=)\s*(.+?)(?:\.|$)", "technical"),
]


def extract_facts_from_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """
    Scan conversation messages for durable facts.
    
    Returns list of fact dicts suitable for fact_store.
    """
    facts = []
    seen = set()  # Deduplicate
    
    for msg in messages:
        if msg.get("role") != "user":
            continue
        
        content = msg.get("content", "")
        if not isinstance(content, str) or len(content) < 10:
            continue
        
        for pattern, category in _FACT_PATTERNS:
            matches = re.findall(pattern, content, re.IGNORECASE)
            for match in matches:
                if isinstance(match, tuple):
                    match = match[0] if match else ""
                fact_text = match.strip()
                
                if len(fact_text) < 5 or len(fact_text) > 200:
                    continue
                
                # Deduplicate
                dedup_key = f"{category}:{fact_text.lower()}"
                if dedup_key in seen:
                    continue
                seen.add(dedup_key)
                
                facts.append({
                    "content": fact_text,
                    "category": category,
                    "source": "session_compaction",
                    "trust": 0.7,  # Medium trust — extracted, not explicitly stated
                })
    
    return facts


def extract_preferences(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Extract user preferences specifically."""
    prefs = []
    
    pref_patterns = [
        r"(?:i prefer|i like|i want|use|always)\s+(.+?)(?:\.|$)",
        r"(?:my (?:preferred|favorite|default))\s+(?:is|are)\s+(.+?)(?:\.|$)",
        r"(?:set|configure|make)\s+(?:it to|the default to)\s+(.+?)(?:\.|$)",
    ]
    
    for msg in messages:
        if msg.get("role") != "user":
            continue
        content = msg.get("content", "")
        if not isinstance(content, str):
            continue
        
        for pattern in pref_patterns:
            matches = re.findall(pattern, content, re.IGNORECASE)
            for match in matches:
                if isinstance(match, str) and len(match) > 5 and len(match) < 200:
                    prefs.append({
                        "content": match.strip(),
                        "category": "user_pref",
                        "source": "session_compaction",
                        "trust": 0.8,
                    })
    
    return prefs


def compact_session(
    messages: List[Dict[str, Any]],
    fact_store: Any = None,
    keep_recent: int = 10,
) -> Tuple[List[Dict[str, Any]], int]:
    """
    Compact a session by extracting facts and compressing old messages.
    
    Args:
        messages: Full conversation history
        fact_store: Optional fact_store instance for saving facts
        keep_recent: Number of recent messages to keep uncompressed
    
    Returns:
        Tuple of (compacted_messages, facts_extracted)
    """
    if len(messages) <= keep_recent * 2:
        return messages, 0
    
    # Split into old (to compress) and recent (to keep)
    split_point = len(messages) - keep_recent
    old_messages = messages[:split_point]
    recent_messages = messages[split_point:]
    
    # Extract facts from old messages
    facts = extract_facts_from_messages(old_messages)
    prefs = extract_preferences(old_messages)
    all_facts = facts + prefs
    
    # Save facts to store if available
    saved_count = 0
    if fact_store and all_facts:
        for fact in all_facts:
            try:
                if hasattr(fact_store, 'store'):
                    fact_store.store(
                        content=fact["content"],
                        category=fact["category"],
                        tags=["session_compaction"],
                    )
                    saved_count += 1
                elif hasattr(fact_store, 'add'):
                    fact_store.add(fact["content"])
                    saved_count += 1
            except Exception:
                pass  # Don't let fact saving block compaction
    
    # Create summary of old messages
    summary_parts = []
    if saved_count > 0:
        summary_parts.append(f"[Session compacted: {saved_count} facts extracted and saved]")
    
    # Count message types
    user_msgs = sum(1 for m in old_messages if m.get("role") == "user")
    asst_msgs = sum(1 for m in old_messages if m.get("role") == "assistant")
    summary_parts.append(f"[Previous conversation: {user_msgs} user messages, {asst_msgs} assistant responses]")
    
    summary = " ".join(summary_parts)
    
    # Build compacted messages
    compacted = []
    
    # Add summary as system message
    if summary:
        compacted.append({
            "role": "system",
            "content": summary,
            "_compacted": True,
        })
    
    # Add extracted facts as system context
    if all_facts:
        facts_text = "Known facts from previous conversation:\n"
        for fact in all_facts[:20]:  # Limit to 20 facts
            facts_text += f"- [{fact['category']}] {fact['content']}\n"
        
        compacted.append({
            "role": "system",
            "content": facts_text,
            "_extracted_facts": True,
        })
    
    # Add recent messages
    compacted.extend(recent_messages)
    
    return compacted, saved_count


def should_compact(messages: List[Dict[str, Any]], max_tokens: int = 80000) -> bool:
    """
    Determine if compaction is needed based on message count/length.
    
    Simple heuristic: compact if we have many messages or very long content.
    """
    if len(messages) < 50:
        return False
    
    # Estimate token count (rough: 4 chars per token)
    total_chars = sum(len(str(m.get("content", ""))) for m in messages)
    estimated_tokens = total_chars // 4
    
    return estimated_tokens > max_tokens * 0.8  # Compact at 80% of limit