Pathfinder MCP Server

"""Session manager for save/load of session snapshots.""" import json import os import re from pathlib import Path from pydantic import BaseModel from pathfinder_mcp.state import Phase def get_sessions_dir() -> Path: """Get sessions directory from env var or default.""" default = Path.home() / ".pathfinder-sessions" path = Path(os.environ.get("PATHFINDER_SESSIONS_DIR", default)) path.mkdir(parents=True, exist_ok=True) return path def _sanitize_session_id(session_id: str) -> str: """Sanitize session ID to prevent path traversal.""" # Remove any path separators and dangerous characters sanitized = re.sub(r"[/\\:*?\"<>|.]", "_", session_id) # Ensure not empty after sanitization if not sanitized or sanitized.startswith("_"): sanitized = f"session_{sanitized}" return sanitized[:64] # Limit length class SessionSnapshot(BaseModel): """Snapshot of session state.""" session_id: str phase: Phase task_description: str = "" research_summary: str = "" plan_summary: str = "" context_tokens: int = 0 class SessionManager: """Handles save/load of session snapshots.""" def __init__(self, base_dir: Path | None = None): self.base_dir = base_dir or get_sessions_dir() def get_session_path(self, session_id: str) -> Path: """Get path for a session directory.""" safe_id = _sanitize_session_id(session_id) return self.base_dir / safe_id def create_session(self, session_id: str) -> Path: """Create session directory. Returns path.""" path = self.get_session_path(session_id) path.mkdir(parents=True, exist_ok=True) return path def save_snapshot(self, session_id: str, data: SessionSnapshot) -> None: """Save session snapshot as JSON.""" path = self.get_session_path(session_id) path.mkdir(parents=True, exist_ok=True) snapshot_file = path / "session_state.json" snapshot_file.write_text(data.model_dump_json(indent=2)) def load_snapshot(self, session_id: str) -> SessionSnapshot | None: """Load previous session snapshot.""" path = self.get_session_path(session_id) snapshot_file = path / "session_state.json" if not snapshot_file.exists(): return None try: return SessionSnapshot.model_validate_json(snapshot_file.read_text()) except (json.JSONDecodeError, ValueError): return None def session_exists(self, session_id: str) -> bool: """Check if session directory exists.""" return self.get_session_path(session_id).exists()