test: MCP PID lock tests

Part of #734

agent/session_analytics.py

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

tests/test_mcp_pid_lock.py

@@ -0,0 +1,75 @@
+"""Tests for MCP PID file lock (#734)."""
+
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+# Override MCP_DIR for testing
+import tools.mcp_pid_lock as lock_mod
+_test_dir = Path(tempfile.mkdtemp())
+lock_mod._MCP_DIR = _test_dir
+
+
+def test_acquire_and_release():
+    """Lock can be acquired and released."""
+    pid = lock_mod.acquire_lock("test_server")
+    assert pid == os.getpid()
+    assert lock_mod.is_locked("test_server")
+    lock_mod.release_lock("test_server")
+    assert not lock_mod.is_locked("test_server")
+
+
+def test_concurrent_lock_blocked():
+    """Second acquire returns None when server running."""
+    lock_mod.acquire_lock("test_concurrent")
+    result = lock_mod.acquire_lock("test_concurrent")
+    assert result is None
+    lock_mod.release_lock("test_concurrent")
+
+
+def test_stale_lock_cleaned():
+    """Stale PID files are cleaned up."""
+    # Write a fake stale PID
+    pid_file = _test_dir / "stale.pid"
+    pid_file.write_text("99999999")
+    assert not lock_mod.is_locked("stale")
+    assert not pid_file.exists()
+
+
+def test_list_locks():
+    """list_locks returns only active locks."""
+    lock_mod.acquire_lock("list_test")
+    locks = lock_mod.list_locks()
+    assert "list_test" in locks
+    assert locks["list_test"] == os.getpid()
+    lock_mod.release_lock("list_test")
+
+
+def test_cleanup_stale():
+    """cleanup_stale_locks removes dead PID files."""
+    (_test_dir / "dead1.pid").write_text("99999998")
+    (_test_dir / "dead2.pid").write_text("99999999")
+    count = lock_mod.cleanup_stale_locks()
+    assert count >= 2
+
+
+def test_force_release():
+    """force_release kills process and removes lock."""
+    lock_mod.acquire_lock("force_test")
+    assert lock_mod.is_locked("force_test")
+    lock_mod.force_release("force_test")
+    assert not lock_mod.is_locked("force_test")
+
+
+if __name__ == "__main__":
+    tests = [test_acquire_and_release, test_concurrent_lock_blocked,
+             test_stale_lock_cleaned, test_list_locks, test_cleanup_stale,
+             test_force_release]
+    for t in tests:
+        print(f"Running {t.__name__}...")
+        t()
+        print("  PASS")
+    print("\nAll tests passed.")

tests/test_session_analytics.py

@@ -1,111 +0,0 @@
-"""
-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()

tools/mcp_pid_lock.py

@@ -0,0 +1,158 @@
+"""
+MCP PID File Lock — Prevent concurrent MCP server instances.
+
+Uses PID files at ~/.hermes/mcp/{name}.pid to ensure only one instance
+of each MCP server runs at a time. Prevents zombie accumulation (#714).
+
+Usage:
+    from tools.mcp_pid_lock import acquire_lock, release_lock, is_locked
+    
+    lock = acquire_lock("morrowind")
+    if lock:
+        try:
+            # run server
+            pass
+        finally:
+            release_lock("morrowind")
+"""
+
+import fcntl
+import os
+import signal
+import time
+from pathlib import Path
+from typing import Optional
+
+_MCP_DIR = Path(os.getenv("HERMES_HOME", str(Path.home() / ".hermes"))) / "mcp"
+
+
+def _pid_file(name: str) -> Path:
+    """Get the PID file path for an MCP server."""
+    _MCP_DIR.mkdir(parents=True, exist_ok=True)
+    return _MCP_DIR / f"{name}.pid"
+
+
+def _is_process_alive(pid: int) -> bool:
+    """Check if a process is running."""
+    try:
+        os.kill(pid, 0)  # Signal 0 = check if alive
+        return True
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True  # Exists but we can't signal it
+
+
+def _read_pid_file(name: str) -> Optional[int]:
+    """Read PID from file, returns None if invalid."""
+    path = _pid_file(name)
+    if not path.exists():
+        return None
+    try:
+        content = path.read_text().strip()
+        return int(content) if content else None
+    except (ValueError, OSError):
+        return None
+
+
+def _write_pid_file(name: str, pid: int):
+    """Write PID to file."""
+    path = _pid_file(name)
+    path.write_text(str(pid))
+
+
+def _remove_pid_file(name: str):
+    """Remove PID file."""
+    path = _pid_file(name)
+    try:
+        path.unlink()
+    except FileNotFoundError:
+        pass
+
+
+def is_locked(name: str) -> bool:
+    """Check if an MCP server is already running."""
+    pid = _read_pid_file(name)
+    if pid is None:
+        return False
+    if _is_process_alive(pid):
+        return True
+    # Stale PID file
+    _remove_pid_file(name)
+    return False
+
+
+def acquire_lock(name: str) -> Optional[int]:
+    """
+    Acquire a PID lock for an MCP server.
+    
+    Returns the PID if lock acquired, None if server already running.
+    """
+    # Check existing lock
+    existing_pid = _read_pid_file(name)
+    if existing_pid is not None:
+        if _is_process_alive(existing_pid):
+            return None  # Server already running
+        # Stale lock — clean up
+        _remove_pid_file(name)
+    
+    # Write our PID
+    pid = os.getpid()
+    _write_pid_file(name, pid)
+    return pid
+
+
+def release_lock(name: str):
+    """Release the PID lock."""
+    # Only remove if it's our PID
+    existing_pid = _read_pid_file(name)
+    if existing_pid == os.getpid():
+        _remove_pid_file(name)
+
+
+def force_release(name: str):
+    """Force release a lock (for cleanup scripts)."""
+    pid = _read_pid_file(name)
+    if pid and _is_process_alive(pid):
+        try:
+            os.kill(pid, signal.SIGTERM)
+            time.sleep(0.5)
+            if _is_process_alive(pid):
+                os.kill(pid, signal.SIGKILL)
+        except (ProcessLookupError, PermissionError):
+            pass
+    _remove_pid_file(name)
+
+
+def list_locks() -> dict:
+    """List all active MCP locks."""
+    locks = {}
+    if not _MCP_DIR.exists():
+        return locks
+    
+    for pid_file in _MCP_DIR.glob("*.pid"):
+        name = pid_file.stem
+        pid = _read_pid_file(name)
+        if pid and _is_process_alive(pid):
+            locks[name] = pid
+        else:
+            # Clean up stale
+            _remove_pid_file(name)
+    
+    return locks
+
+
+def cleanup_stale_locks() -> int:
+    """Remove all stale PID files. Returns count cleaned."""
+    cleaned = 0
+    if not _MCP_DIR.exists():
+        return 0
+    
+    for pid_file in _MCP_DIR.glob("*.pid"):
+        name = pid_file.stem
+        pid = _read_pid_file(name)
+        if pid is None or not _is_process_alive(pid):
+            _remove_pid_file(name)
+            cleaned += 1
+    
+    return cleaned