test: session compaction tests

Part of #748

agent/session_compaction.py

@@ -0,0 +1,221 @@
+"""
+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

agent/session_model_metadata.py

@@ -1,223 +0,0 @@
-"""
-Session Model Metadata — Persist model context info per session
-
-When a session switches models mid-conversation, context length and
-token budget need to be updated to prevent silent truncation.
-
-Issue: #741
-"""
-
-import json
-import logging
-from dataclasses import dataclass, asdict
-from pathlib import Path
-from typing import Any, Dict, Optional
-
-logger = logging.getLogger(__name__)
-
-HERMES_HOME = Path.home() / ".hermes"
-
-
-# Common model context lengths (tokens)
-KNOWN_CONTEXT_LENGTHS = {
-    # Anthropic
-    "claude-opus-4-6": 200000,
-    "claude-sonnet-4": 200000,
-    "claude-3.5-sonnet": 200000,
-    "claude-3-haiku": 200000,
-    
-    # OpenAI
-    "gpt-4o": 128000,
-    "gpt-4-turbo": 128000,
-    "gpt-4": 8192,
-    "gpt-3.5-turbo": 16385,
-    
-    # Nous / open models
-    "hermes-3-llama-3.1-405b": 131072,
-    "hermes-3-llama-3.1-70b": 131072,
-    "deepseek-r1": 131072,
-    "deepseek-v3": 131072,
-    
-    # Local
-    "llama-3.1-8b": 131072,
-    "llama-3.1-70b": 131072,
-    "qwen-2.5-72b": 131072,
-    
-    # Xiaomi
-    "mimo-v2-pro": 131072,
-    "mimo-v2-flash": 131072,
-    
-    # Defaults
-    "default": 4096,
-}
-
-# Reserve tokens for system prompt, response, and overhead
-TOKEN_RESERVE = 2000
-
-
-@dataclass
-class ModelMetadata:
-    """Metadata for a model in a session."""
-    model: str
-    provider: str
-    context_length: int
-    available_for_input: int  # context_length - reserve
-    current_tokens_used: int = 0
-    
-    @property
-    def remaining_tokens(self) -> int:
-        """Tokens remaining for new input."""
-        return max(0, self.available_for_input - self.current_tokens_used)
-    
-    @property
-    def utilization_pct(self) -> float:
-        """Percentage of context used."""
-        if self.available_for_input == 0:
-            return 0.0
-        return (self.current_tokens_used / self.available_for_input) * 100
-    
-    def to_dict(self) -> Dict[str, Any]:
-        return asdict(self)
-
-
-def get_context_length(model: str) -> int:
-    """Get context length for a model."""
-    model_lower = model.lower()
-    
-    # Check exact match
-    if model_lower in KNOWN_CONTEXT_LENGTHS:
-        return KNOWN_CONTEXT_LENGTHS[model_lower]
-    
-    # Check partial match
-    for key, length in KNOWN_CONTEXT_LENGTHS.items():
-        if key in model_lower:
-            return length
-    
-    return KNOWN_CONTEXT_LENGTHS["default"]
-
-
-def create_metadata(model: str, provider: str = "", current_tokens: int = 0) -> ModelMetadata:
-    """Create model metadata."""
-    context_length = get_context_length(model)
-    available = max(0, context_length - TOKEN_RESERVE)
-    
-    return ModelMetadata(
-        model=model,
-        provider=provider,
-        context_length=context_length,
-        available_for_input=available,
-        current_tokens_used=current_tokens
-    )
-
-
-def check_model_switch(
-    old_model: str,
-    new_model: str,
-    current_tokens: int
-) -> Dict[str, Any]:
-    """
-    Check impact of switching models mid-session.
-    
-    Returns:
-        Dict with switch analysis including warnings
-    """
-    old_ctx = get_context_length(old_model)
-    new_ctx = get_context_length(new_model)
-    
-    old_available = old_ctx - TOKEN_RESERVE
-    new_available = new_ctx - TOKEN_RESERVE
-    
-    result = {
-        "old_model": old_model,
-        "new_model": new_model,
-        "old_context": old_ctx,
-        "new_context": new_ctx,
-        "current_tokens": current_tokens,
-        "fits_in_new": current_tokens <= new_available,
-        "truncation_needed": max(0, current_tokens - new_available),
-        "warning": None,
-    }
-    
-    if not result["fits_in_new"]:
-        result["warning"] = (
-            f"Switching to {new_model} ({new_ctx:,} ctx) with {current_tokens:,} tokens "
-            f"will truncate {result['truncation_needed']:,} tokens of history. "
-            f"Consider starting a new session."
-        )
-    
-    if new_ctx < old_ctx:
-        reduction = old_ctx - new_ctx
-        result["warning"] = (
-            f"New model has {reduction:,} fewer tokens of context. "
-            f"({old_ctx:,} -> {new_ctx:,})"
-        )
-    
-    return result
-
-
-class SessionModelTracker:
-    """Track model metadata for a session."""
-    
-    def __init__(self, session_id: str):
-        self.session_id = session_id
-        self.metadata: Optional[ModelMetadata] = None
-        self.history: list = []  # Model switch history
-    
-    def set_model(self, model: str, provider: str = "", tokens_used: int = 0):
-        """Set the current model for the session."""
-        old_model = self.metadata.model if self.metadata else None
-        
-        self.metadata = create_metadata(model, provider, tokens_used)
-        
-        # Record switch in history
-        if old_model and old_model != model:
-            self.history.append({
-                "from": old_model,
-                "to": model,
-                "tokens_at_switch": tokens_used,
-                "context_length": self.metadata.context_length
-            })
-        
-        logger.info(
-            "Session %s: model=%s context=%d available=%d",
-            self.session_id[:12], model,
-            self.metadata.context_length,
-            self.metadata.available_for_input
-        )
-    
-    def update_tokens(self, tokens: int):
-        """Update current token usage."""
-        if self.metadata:
-            self.metadata.current_tokens_used = tokens
-    
-    def get_remaining(self) -> int:
-        """Get remaining tokens."""
-        if not self.metadata:
-            return 0
-        return self.metadata.remaining_tokens
-    
-    def can_fit(self, additional_tokens: int) -> bool:
-        """Check if additional tokens fit in context."""
-        if not self.metadata:
-            return False
-        return self.metadata.remaining_tokens >= additional_tokens
-    
-    def get_warning(self) -> Optional[str]:
-        """Get warning if context is running low."""
-        if not self.metadata:
-            return None
-        
-        util = self.metadata.utilization_pct
-        if util > 90:
-            return f"Context {util:.0f}% full. Consider compression or new session."
-        if util > 75:
-            return f"Context {util:.0f}% full."
-        return None
-    
-    def to_dict(self) -> Dict[str, Any]:
-        """Export state."""
-        return {
-            "session_id": self.session_id,
-            "metadata": self.metadata.to_dict() if self.metadata else None,
-            "history": self.history
-        }

tests/test_session_compaction.py

@@ -0,0 +1,84 @@
+"""Tests for session compaction with fact extraction (#748)."""
+
+import sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from agent.session_compaction import (
+    extract_facts_from_messages,
+    extract_preferences,
+    compact_session,
+    should_compact,
+)
+
+
+def test_extract_preferences():
+    msgs = [
+        {"role": "user", "content": "I prefer using Python for this"},
+        {"role": "assistant", "content": "OK"},
+        {"role": "user", "content": "Always use tabs, not spaces"},
+    ]
+    prefs = extract_preferences(msgs)
+    assert len(prefs) >= 1
+
+
+def test_extract_facts():
+    msgs = [
+        {"role": "user", "content": "The server runs on port 8080"},
+        {"role": "user", "content": "Actually, the port is 8081"},
+        {"role": "user", "content": "Hello"},  # Too short, should be skipped
+    ]
+    facts = extract_facts_from_messages(msgs)
+    assert len(facts) >= 1
+    assert any("technical" in f["category"] for f in facts)
+
+
+def test_extract_deduplicates():
+    msgs = [
+        {"role": "user", "content": "I prefer Python"},
+        {"role": "user", "content": "I prefer Python"},
+    ]
+    facts = extract_facts_from_messages(msgs)
+    assert len(facts) == 1
+
+
+def test_compact_session():
+    messages = []
+    for i in range(30):
+        messages.append({"role": "user", "content": f"Message {i}: I prefer Python for server {i}"})
+        messages.append({"role": "assistant", "content": f"Response {i}"})
+    
+    compacted, count = compact_session(messages, keep_recent=10)
+    assert len(compacted) < len(messages)
+    assert count >= 0
+
+
+def test_compact_keeps_recent():
+    messages = []
+    for i in range(30):
+        messages.append({"role": "user", "content": f"Message {i}"})
+        messages.append({"role": "assistant", "content": f"Response {i}"})
+    
+    compacted, _ = compact_session(messages, keep_recent=10)
+    # Should have summary + facts + 10 recent
+    assert len(compacted) >= 10
+
+
+def test_should_compact_short():
+    messages = [{"role": "user", "content": "hi"} for _ in range(10)]
+    assert not should_compact(messages)
+
+
+def test_should_compact_long():
+    messages = [{"role": "user", "content": "x" * 1000} for _ in range(100)]
+    assert should_compact(messages)
+
+
+if __name__ == "__main__":
+    tests = [test_extract_preferences, test_extract_facts, test_extract_deduplicates,
+             test_compact_session, test_compact_keeps_recent, test_should_compact_short, test_should_compact_long]
+    for t in tests:
+        print(f"Running {t.__name__}...")
+        t()
+        print("  PASS")
+    print("\nAll tests passed.")

tests/test_session_model_metadata.py

@@ -1,105 +0,0 @@
-"""
-Tests for session model metadata
-
-Issue: #741
-"""
-
-import unittest
-from agent.session_model_metadata import (
-    get_context_length,
-    create_metadata,
-    check_model_switch,
-    SessionModelTracker,
-)
-
-
-class TestContextLength(unittest.TestCase):
-    
-    def test_known_model(self):
-        ctx = get_context_length("claude-opus-4-6")
-        self.assertEqual(ctx, 200000)
-    
-    def test_partial_match(self):
-        ctx = get_context_length("anthropic/claude-sonnet-4")
-        self.assertEqual(ctx, 200000)
-    
-    def test_unknown_model(self):
-        ctx = get_context_length("unknown-model-xyz")
-        self.assertEqual(ctx, 4096)
-
-
-class TestModelMetadata(unittest.TestCase):
-    
-    def test_create(self):
-        meta = create_metadata("gpt-4o", "openai", 1000)
-        self.assertEqual(meta.context_length, 128000)
-        self.assertEqual(meta.current_tokens_used, 1000)
-        self.assertGreater(meta.remaining_tokens, 0)
-    
-    def test_utilization(self):
-        meta = create_metadata("gpt-4o", "openai", 64000)
-        self.assertAlmostEqual(meta.utilization_pct, 50.0, delta=1)
-
-
-class TestModelSwitch(unittest.TestCase):
-    
-    def test_safe_switch(self):
-        result = check_model_switch("gpt-3.5-turbo", "gpt-4o", 5000)
-        self.assertTrue(result["fits_in_new"])
-        self.assertIsNone(result["warning"])
-    
-    def test_truncation_warning(self):
-        result = check_model_switch("gpt-4o", "gpt-3.5-turbo", 20000)
-        self.assertFalse(result["fits_in_new"])
-        self.assertIsNotNone(result["warning"])
-        self.assertIn("truncate", result["warning"].lower())
-    
-    def test_downgrade_warning(self):
-        result = check_model_switch("claude-opus-4-6", "gpt-4", 5000)
-        self.assertIsNotNone(result["warning"])
-
-
-class TestSessionModelTracker(unittest.TestCase):
-    
-    def test_set_model(self):
-        tracker = SessionModelTracker("test")
-        tracker.set_model("gpt-4o", "openai")
-        self.assertEqual(tracker.metadata.model, "gpt-4o")
-    
-    def test_update_tokens(self):
-        tracker = SessionModelTracker("test")
-        tracker.set_model("gpt-4o")
-        tracker.update_tokens(5000)
-        self.assertEqual(tracker.metadata.current_tokens_used, 5000)
-    
-    def test_remaining(self):
-        tracker = SessionModelTracker("test")
-        tracker.set_model("gpt-4o")
-        tracker.update_tokens(10000)
-        self.assertGreater(tracker.get_remaining(), 0)
-    
-    def test_can_fit(self):
-        tracker = SessionModelTracker("test")
-        tracker.set_model("gpt-4o")
-        tracker.update_tokens(10000)
-        self.assertTrue(tracker.can_fit(5000))
-        self.assertFalse(tracker.can_fit(200000))
-    
-    def test_warning_low_context(self):
-        tracker = SessionModelTracker("test")
-        tracker.set_model("gpt-4o")
-        tracker.update_tokens(115000)  # ~90% used
-        warning = tracker.get_warning()
-        self.assertIsNotNone(warning)
-    
-    def test_model_switch_history(self):
-        tracker = SessionModelTracker("test")
-        tracker.set_model("gpt-4o", "openai")
-        tracker.update_tokens(5000)
-        tracker.set_model("claude-opus-4-6", "anthropic")
-        self.assertEqual(len(tracker.history), 1)
-        self.assertEqual(tracker.history[0]["from"], "gpt-4o")
-
-
-if __name__ == "__main__":
-    unittest.main()