Pathfinder MCP Server

"""Plan phase tools.""" import re from typing import Any from fastmcp.server.context import Context from pathfinder_mcp.artifacts import ArtifactWriter from pathfinder_mcp.context import ContextMonitor from pathfinder_mcp.session import SessionManager from pathfinder_mcp.state import Phase, PhaseState async def start_plan( session_id: str, ctx: Context | None = None, *, session_manager: SessionManager, artifact_writer: ArtifactWriter, context_monitor: ContextMonitor, sessions: dict[str, PhaseState], ) -> dict[str, Any]: """Transition from research to plan phase. Requires research artifact to exist. Uses elicit to confirm transition. Args: session_id: Session ID ctx: FastMCP context for elicitation session_manager: Session manager instance artifact_writer: Artifact writer instance context_monitor: Context monitor instance sessions: Active sessions dict Returns: Status with plan template path """ # Validate session state = sessions.get(session_id) if not state: return {"error": "Session not found", "code": "SESSION_NOT_FOUND"} # Validate current phase if not state.is_research: return { "error": f"Cannot start plan from {state.current_phase.value} phase", "code": "INVALID_PHASE", } # Validate research artifact exists if not artifact_writer.artifact_exists(session_id, "research.md"): return { "error": "Research artifact not found. Complete research first.", "code": "MISSING_RESEARCH", } # Elicit confirmation if context available if ctx is not None: try: result = await ctx.elicit( ( "Ready to transition to planning phase? " "This will create a plan template." ), response_type=["yes", "no"], ) if result.action != "accept" or result.data == "no": return { "session_id": session_id, "phase": "research", "message": "Plan transition cancelled. Continue research.", "cancelled": True, } except Exception: # Elicit not supported by client, proceed without confirmation pass # Read research to extract task name research = artifact_writer.read_artifact(session_id, "research.md") or "" # Extract task from first heading or use default match = re.search(r"^# Research: (.+)$", research, re.MULTILINE) task_name = match.group(1) if match else "Implementation Plan" # Transition phase new_state = state.transition_to(Phase.PLAN) sessions[session_id] = new_state # Create plan template template = artifact_writer.get_plan_template( name=task_name, overview=f"Implementation plan for: {task_name}" ) artifact_path = artifact_writer.write_plan(session_id, template) # Update snapshot snapshot = session_manager.load_snapshot(session_id) if snapshot: snapshot.phase = Phase.PLAN 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 with your implementation plan.", "template_created": True, } def validate_plan_structure(content: str) -> tuple[bool, list[str]]: """Validate plan follows Cursor plan format. Returns: Tuple of (is_valid, list of errors) """ errors: list[str] = [] # Check YAML frontmatter if not content.startswith("---"): errors.append("Missing YAML frontmatter (must start with ---)") else: # Find frontmatter end parts = content.split("---", 2) if len(parts) < 3: errors.append("Invalid YAML frontmatter (missing closing ---)") else: frontmatter = parts[1] if "name:" not in frontmatter: errors.append("YAML frontmatter missing 'name' field") if "overview:" not in frontmatter: errors.append("YAML frontmatter missing 'overview' field") if "todos:" not in frontmatter: errors.append("YAML frontmatter missing 'todos' field") # Check required sections required_sections = [ ("## Architecture Overview", "Architecture Overview section"), ("## Core Components", "Core Components section"), ("## Phase 1", "At least one Phase section"), ("## Implementation Details", "Implementation Details section"), ("## Success Criteria", "Success Criteria section"), ] for pattern, name in required_sections: if pattern not in content: errors.append(f"Missing {name}") return len(errors) == 0, errors def save_plan( session_id: str, plan_content: str, *, session_manager: SessionManager, artifact_writer: ArtifactWriter, context_monitor: ContextMonitor, sessions: dict[str, PhaseState], ) -> dict[str, Any]: """Save implementation plan. Validates plan follows Cursor plan format. Args: session_id: Session ID plan_content: Plan content in Cursor plan format session_manager: Session manager instance artifact_writer: Artifact writer instance context_monitor: Context monitor instance sessions: Active sessions dict Returns: Status with validation results """ # Validate session state = sessions.get(session_id) if not state: return {"error": "Session not found", "code": "SESSION_NOT_FOUND"} # Validate current phase if not state.is_plan: return { "error": f"Cannot save plan in {state.current_phase.value} phase", "code": "INVALID_PHASE", } # Validate plan structure is_valid, validation_errors = validate_plan_structure(plan_content) if not is_valid: return { "error": "Plan validation failed", "code": "INVALID_PLAN_FORMAT", "validation_errors": validation_errors, "hint": ( "Plan must follow Cursor plan format with YAML frontmatter " "and required sections." ), } # Write plan artifact_path = artifact_writer.write_plan(session_id, plan_content) # Track tokens context_monitor.add_message(plan_content) # Update snapshot snapshot = session_manager.load_snapshot(session_id) if snapshot: snapshot.plan_summary = plan_content[:500] snapshot.context_tokens = context_monitor.current_tokens session_manager.save_snapshot(session_id, snapshot) return { "session_id": session_id, "artifact_path": str(artifact_path), "validated": True, "context": context_monitor.get_status(), "message": "Plan saved. Use implement_phase to begin implementation.", }