Pathfinder MCP Server

"""Plan phase handler.""" import re from typing import Any from fastmcp.server.context import Context from pathfinder_mcp.handlers.base import BaseHandler from pathfinder_mcp.state import Phase class PlanHandler(BaseHandler): """Handler for plan phase operations.""" phase = Phase.PLAN async def execute( self, session_id: str, plan_content: str | None = None, ctx: Context | None = None, **kwargs: Any, ) -> dict[str, Any]: """Execute plan operation. If plan_content is None, starts plan phase (transitions from research). If plan_content is provided, saves the plan. Args: session_id: Session ID plan_content: Plan content to save (None to start plan phase) ctx: FastMCP context for elicitation Returns: Operation result """ if plan_content is not None: return self._save_plan(session_id, plan_content) return await self._start_plan(session_id, ctx) async def _start_plan(self, session_id: str, ctx: Context | None) -> dict[str, Any]: """Transition to plan phase.""" # Validate in research phase error = self.validate_phase(session_id, [Phase.RESEARCH]) if error: return error # Check research exists if not self.artifact_writer.artifact_exists(session_id, "research.md"): return { "error": "Research artifact not found. Complete research first.", "code": "MISSING_RESEARCH", } # Elicit confirmation if ctx is not None: try: result = await ctx.elicit( "Ready to transition to planning phase?", response_type=["yes", "no"], ) if result.action != "accept" or result.data == "no": return { "session_id": session_id, "phase": "research", "message": "Plan transition cancelled.", "cancelled": True, } except Exception: pass # Elicit not supported # Get task name from research research = self.artifact_writer.read_artifact(session_id, "research.md") or "" match = re.search(r"^# Research: (.+)$", research, re.MULTILINE) task_name = match.group(1) if match else "Implementation Plan" # Transition state = self.get_session(session_id) if state: new_state = state.transition_to(Phase.PLAN) self.set_session(session_id, new_state) # Create template template = self.artifact_writer.get_plan_template( name=task_name, overview=f"Implementation plan for: {task_name}" ) artifact_path = self.artifact_writer.write_plan(session_id, template) # Update snapshot snapshot = self.session_manager.load_snapshot(session_id) if snapshot: snapshot.phase = Phase.PLAN self.session_manager.save_snapshot(session_id, snapshot) return { "session_id": session_id, "phase": "plan", "artifact_path": str(artifact_path), "message": "Plan phase started. Edit plan.md.", "template_created": True, } def _save_plan(self, session_id: str, plan_content: str) -> dict[str, Any]: """Save plan content.""" # Validate in plan phase error = self.validate_phase(session_id, [Phase.PLAN]) if error: return error # Validate structure is_valid, errors = self._validate_structure(plan_content) if not is_valid: return { "error": "Plan validation failed", "code": "INVALID_PLAN_FORMAT", "validation_errors": errors, } # Save artifact_path = self.artifact_writer.write_plan(session_id, plan_content) self.context_monitor.add_message(plan_content) # Update snapshot snapshot = self.session_manager.load_snapshot(session_id) if snapshot: snapshot.plan_summary = plan_content[:500] snapshot.context_tokens = self.context_monitor.current_tokens self.session_manager.save_snapshot(session_id, snapshot) return { "session_id": session_id, "artifact_path": str(artifact_path), "validated": True, "context": self.get_context_status(), "message": "Plan saved. Use implement_phase to begin.", } def _validate_structure(self, content: str) -> tuple[bool, list[str]]: """Validate plan structure.""" errors: list[str] = [] if not content.startswith("---"): errors.append("Missing YAML frontmatter") else: parts = content.split("---", 2) if len(parts) < 3: errors.append("Invalid YAML frontmatter") else: fm = parts[1] for field in ["name:", "overview:", "todos:"]: if field not in fm: errors.append(f"Missing {field.rstrip(':')}") required = [ ("## Architecture Overview", "Architecture Overview"), ("## Core Components", "Core Components"), ("## Phase 1", "Phase section"), ("## Implementation Details", "Implementation Details"), ("## Success Criteria", "Success Criteria"), ] for pattern, name in required: if pattern not in content: errors.append(f"Missing {name}") return len(errors) == 0, errors