test: Add session analytics tests (#753)

agent/session_analytics.py

@@ -0,0 +1,189 @@
+"""
+Session Analytics — Per-session token/cost/time tracking
+
+Tracks resource consumption per session for transparency.
+
+Issue: #753
+"""
+
+import json
+import time
+from dataclasses import dataclass, asdict, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+HERMES_HOME = Path.home() / ".hermes"
+ANALYTICS_DIR = HERMES_HOME / "analytics"
+
+
+# Cost per 1K tokens by provider (input/output)
+COST_TABLE = {
+    "anthropic": {"input": 0.015, "output": 0.075},
+    "openai": {"input": 0.005, "output": 0.015},
+    "nous": {"input": 0.002, "output": 0.006},
+    "openrouter": {"input": 0.005, "output": 0.015},
+    "ollama": {"input": 0.0, "output": 0.0},
+    "local": {"input": 0.0, "output": 0.0},
+}
+
+
+@dataclass
+class SessionStats:
+    """Statistics for a single session."""
+    session_id: str
+    start_time: str
+    end_time: Optional[str] = None
+    
+    # Token counts
+    input_tokens: int = 0
+    output_tokens: int = 0
+    total_tokens: int = 0
+    
+    # Tool usage
+    tool_calls: int = 0
+    tool_errors: int = 0
+    
+    # Timing
+    wall_time_seconds: float = 0.0
+    api_calls: int = 0
+    
+    # Cost
+    estimated_cost_usd: float = 0.0
+    provider: str = ""
+    model: str = ""
+    
+    def to_dict(self) -> Dict[str, Any]:
+        return asdict(self)
+
+
+class SessionTracker:
+    """Track per-session analytics."""
+    
+    def __init__(self, session_id: str, provider: str = "", model: str = ""):
+        self.session_id = session_id
+        self.provider = provider.lower() if provider else ""
+        self.model = model
+        self.start_time = time.time()
+        self.stats = SessionStats(
+            session_id=session_id,
+            start_time=datetime.now(timezone.utc).isoformat(),
+            provider=provider,
+            model=model
+        )
+    
+    def record_tokens(self, input_tokens: int, output_tokens: int):
+        """Record token usage."""
+        self.stats.input_tokens += input_tokens
+        self.stats.output_tokens += output_tokens
+        self.stats.total_tokens = self.stats.input_tokens + self.stats.output_tokens
+        
+        # Estimate cost
+        costs = COST_TABLE.get(self.provider, {"input": 0.01, "output": 0.03})
+        cost = (input_tokens / 1000) * costs["input"] + (output_tokens / 1000) * costs["output"]
+        self.stats.estimated_cost_usd += cost
+    
+    def record_tool_call(self, success: bool = True):
+        """Record a tool call."""
+        self.stats.tool_calls += 1
+        if not success:
+            self.stats.tool_errors += 1
+    
+    def record_api_call(self):
+        """Record an API call."""
+        self.stats.api_calls += 1
+    
+    def finish(self) -> SessionStats:
+        """Finish tracking and return stats."""
+        self.stats.end_time = datetime.now(timezone.utc).isoformat()
+        self.stats.wall_time_seconds = time.time() - self.start_time
+        return self.stats
+    
+    def get_current_stats(self) -> SessionStats:
+        """Get current stats without finishing."""
+        self.stats.wall_time_seconds = time.time() - self.start_time
+        return self.stats
+
+
+def format_stats(stats: SessionStats) -> str:
+    """Format stats for display."""
+    lines = []
+    lines.append(f"Session: {stats.session_id[:20]}...")
+    lines.append(f"Provider: {stats.provider or 'unknown'}")
+    lines.append(f"Model: {stats.model or 'unknown'}")
+    lines.append("")
+    lines.append(f"Tokens: {stats.input_tokens:,} in / {stats.output_tokens:,} out ({stats.total_tokens:,} total)")
+    lines.append(f"Cost: ${stats.estimated_cost_usd:.4f}")
+    lines.append(f"API calls: {stats.api_calls}")
+    lines.append(f"Tool calls: {stats.tool_calls} ({stats.tool_errors} errors)")
+    lines.append(f"Wall time: {stats.wall_time_seconds:.1f}s")
+    return "\n".join(lines)
+
+
+def save_session_stats(stats: SessionStats):
+    """Save session stats to disk."""
+    ANALYTICS_DIR.mkdir(parents=True, exist_ok=True)
+    
+    # Daily file
+    date_str = datetime.now().strftime("%Y-%m-%d")
+    stats_file = ANALYTICS_DIR / f"sessions_{date_str}.jsonl"
+    
+    with open(stats_file, "a") as f:
+        f.write(json.dumps(stats.to_dict()) + "\n")
+
+
+def get_daily_stats(date_str: Optional[str] = None) -> Dict[str, Any]:
+    """Get aggregate stats for a day."""
+    if date_str is None:
+        date_str = datetime.now().strftime("%Y-%m-%d")
+    
+    stats_file = ANALYTICS_DIR / f"sessions_{date_str}.jsonl"
+    if not stats_file.exists():
+        return {"date": date_str, "sessions": 0}
+    
+    sessions = []
+    with open(stats_file) as f:
+        for line in f:
+            line = line.strip()
+            if line:
+                try:
+                    sessions.append(json.loads(line))
+                except json.JSONDecodeError:
+                    pass
+    
+    if not sessions:
+        return {"date": date_str, "sessions": 0}
+    
+    total_tokens = sum(s.get("total_tokens", 0) for s in sessions)
+    total_cost = sum(s.get("estimated_cost_usd", 0) for s in sessions)
+    total_time = sum(s.get("wall_time_seconds", 0) for s in sessions)
+    total_tool_calls = sum(s.get("tool_calls", 0) for s in sessions)
+    total_errors = sum(s.get("tool_errors", 0) for s in sessions)
+    
+    return {
+        "date": date_str,
+        "sessions": len(sessions),
+        "total_tokens": total_tokens,
+        "total_cost_usd": round(total_cost, 4),
+        "total_wall_time_seconds": round(total_time, 1),
+        "total_tool_calls": total_tool_calls,
+        "total_tool_errors": total_errors,
+        "avg_tokens_per_session": total_tokens // len(sessions) if sessions else 0,
+        "avg_cost_per_session": round(total_cost / len(sessions), 4) if sessions else 0,
+    }
+
+
+def format_daily_report(stats: Dict[str, Any]) -> str:
+    """Format daily stats as report."""
+    lines = []
+    lines.append(f"# Session Analytics — {stats['date']}")
+    lines.append("")
+    lines.append(f"Sessions: {stats['sessions']}")
+    lines.append(f"Total tokens: {stats.get('total_tokens', 0):,}")
+    lines.append(f"Total cost: ${stats.get('total_cost_usd', 0):.4f}")
+    lines.append(f"Total wall time: {stats.get('total_wall_time_seconds', 0):.1f}s")
+    lines.append(f"Tool calls: {stats.get('total_tool_calls', 0)} ({stats.get('total_tool_errors', 0)} errors)")
+    lines.append("")
+    lines.append(f"Avg tokens/session: {stats.get('avg_tokens_per_session', 0):,}")
+    lines.append(f"Avg cost/session: ${stats.get('avg_cost_per_session', 0):.4f}")
+    return "\n".join(lines)

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_analytics.py

@@ -0,0 +1,111 @@
+"""
+Tests for session analytics
+
+Issue: #753
+"""
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from agent.session_analytics import (
+    SessionTracker,
+    SessionStats,
+    format_stats,
+    get_daily_stats,
+    format_daily_report,
+)
+
+
+class TestSessionStats(unittest.TestCase):
+    
+    def test_defaults(self):
+        stats = SessionStats(session_id="test", start_time="2026-01-01")
+        self.assertEqual(stats.input_tokens, 0)
+        self.assertEqual(stats.output_tokens, 0)
+        self.assertEqual(stats.tool_calls, 0)
+    
+    def test_to_dict(self):
+        stats = SessionStats(session_id="test", start_time="2026-01-01")
+        d = stats.to_dict()
+        self.assertEqual(d["session_id"], "test")
+        self.assertIn("input_tokens", d)
+
+
+class TestSessionTracker(unittest.TestCase):
+    
+    def test_record_tokens(self):
+        tracker = SessionTracker("test", provider="openai")
+        tracker.record_tokens(100, 50)
+        stats = tracker.get_current_stats()
+        self.assertEqual(stats.input_tokens, 100)
+        self.assertEqual(stats.output_tokens, 50)
+        self.assertGreater(stats.estimated_cost_usd, 0)
+    
+    def test_record_tool_call(self):
+        tracker = SessionTracker("test")
+        tracker.record_tool_call(success=True)
+        tracker.record_tool_call(success=False)
+        stats = tracker.get_current_stats()
+        self.assertEqual(stats.tool_calls, 2)
+        self.assertEqual(stats.tool_errors, 1)
+    
+    def test_free_provider(self):
+        tracker = SessionTracker("test", provider="ollama")
+        tracker.record_tokens(1000, 500)
+        stats = tracker.get_current_stats()
+        self.assertEqual(stats.estimated_cost_usd, 0.0)
+    
+    def test_finish(self):
+        tracker = SessionTracker("test")
+        stats = tracker.finish()
+        self.assertIsNotNone(stats.end_time)
+        self.assertGreater(stats.wall_time_seconds, 0)
+
+
+class TestFormatStats(unittest.TestCase):
+    
+    def test_format(self):
+        stats = SessionStats(
+            session_id="test123",
+            start_time="2026-01-01",
+            input_tokens=1000,
+            output_tokens=500,
+            total_tokens=1500,
+            tool_calls=5,
+            tool_errors=1,
+            wall_time_seconds=30.5,
+            api_calls=3
+        )
+        formatted = format_stats(stats)
+        self.assertIn("1,000", formatted)
+        self.assertIn("500", formatted)
+
+
+class TestDailyStats(unittest.TestCase):
+    
+    def test_empty(self):
+        with patch("agent.session_analytics.ANALYTICS_DIR", Path(tempfile.mkdtemp())):
+            stats = get_daily_stats("2020-01-01")
+            self.assertEqual(stats["sessions"], 0)
+    
+    def test_format_report(self):
+        stats = {
+            "date": "2026-04-14",
+            "sessions": 10,
+            "total_tokens": 50000,
+            "total_cost_usd": 0.50,
+            "total_wall_time_seconds": 300,
+            "total_tool_calls": 100,
+            "total_tool_errors": 5,
+            "avg_tokens_per_session": 5000,
+            "avg_cost_per_session": 0.05,
+        }
+        report = format_daily_report(stats)
+        self.assertIn("10", report)
+        self.assertIn("50,000", report)
+
+
+if __name__ == "__main__":
+    unittest.main()

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()