333 lines
11 KiB
Python
333 lines
11 KiB
Python
"""
|
|
Planning Agent - Analyzes project and generates task lists.
|
|
"""
|
|
import sys
|
|
import json
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
from typing import Dict, Any, List
|
|
|
|
from config import AgentConfig
|
|
from cursor_cli_wrapper import CursorCLI, CursorResultStatus
|
|
|
|
|
|
class PlanningAgent:
|
|
"""Planning agent that generates task lists."""
|
|
|
|
def __init__(self, config: AgentConfig):
|
|
self.config = config
|
|
self.cursor_cli = CursorCLI(
|
|
project_root=config.project_root,
|
|
api_key=config.cursor_api_key,
|
|
model=config.cursor_model,
|
|
verbose=True
|
|
)
|
|
|
|
def run(self) -> int:
|
|
"""
|
|
Generate task list for component.
|
|
Returns: 0 on success, 1 on error
|
|
"""
|
|
print(f"🎯 Planning Agent for component: {self.config.component}")
|
|
print(f"📁 Project root: {self.config.project_root}")
|
|
|
|
try:
|
|
# Build planning prompt
|
|
prompt = self._build_planning_prompt()
|
|
|
|
# Run cursor agent to analyze and plan
|
|
print("\n🤖 Running Cursor Planning Agent...")
|
|
result = self.cursor_cli.run_agent(
|
|
task_description=prompt,
|
|
force=True, # Allow writing task_list.json
|
|
max_iterations=3
|
|
)
|
|
|
|
if result.status != CursorResultStatus.SUCCESS:
|
|
print(f"❌ Planning agent failed: {result.error}")
|
|
return 1
|
|
|
|
# Verify task list was generated
|
|
if not self.config.task_list_path.exists():
|
|
print(f"❌ Task list not generated: {self.config.task_list_path}")
|
|
return 1
|
|
|
|
# Validate and summarize task list
|
|
task_list = self._load_and_validate_task_list()
|
|
|
|
if not task_list:
|
|
print("❌ Invalid task list generated")
|
|
return 1
|
|
|
|
# Print summary
|
|
self._print_summary(task_list)
|
|
|
|
# Create summary commit
|
|
self._commit_task_list()
|
|
|
|
print("\n✅ Planning completed successfully")
|
|
return 0
|
|
|
|
except Exception as e:
|
|
print(f"\n❌ Planning error: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return 1
|
|
|
|
def _build_planning_prompt(self) -> str:
|
|
"""Build comprehensive planning prompt."""
|
|
# Load planning prompt template
|
|
prompt_file = self.config.prompts_dir / "planning_prompt.md"
|
|
|
|
if prompt_file.exists():
|
|
with open(prompt_file, 'r') as f:
|
|
template = f.read()
|
|
else:
|
|
template = "You are an AI planning agent. Analyze the project and create a task list.\n\n"
|
|
|
|
# Add context from existing files
|
|
context = self._gather_context()
|
|
|
|
prompt = f"""{template}
|
|
|
|
## PLANNING REQUEST
|
|
|
|
Generate a task list for component: **{self.config.component}**
|
|
|
|
### Current Context:
|
|
|
|
{context}
|
|
|
|
### Task List Requirements:
|
|
|
|
1. Analyze the current state from tasks.md and workflow_state.md
|
|
2. Review what's already completed in agent_state.json
|
|
3. Create new tasks or update existing ones
|
|
4. Ensure proper dependencies and priorities
|
|
5. Write the complete task list to: `{self.config.task_list_path}`
|
|
|
|
### Output Format:
|
|
|
|
Write a JSON file at `{self.config.task_list_path}` with this structure:
|
|
|
|
```json
|
|
{{
|
|
"project": "mnemo_cards_{self.config.component}",
|
|
"component": "{self.config.component}",
|
|
"version": "1.0",
|
|
"generated_at": "{datetime.now(timezone.utc).isoformat()}",
|
|
"generated_by": "planning_agent",
|
|
"tasks": [
|
|
{{
|
|
"id": "TASK-001",
|
|
"title": "Clear title",
|
|
"priority": "high|medium|low",
|
|
"status": "pending",
|
|
"estimated_hours": 4.0,
|
|
"description": "Detailed description",
|
|
"acceptance_criteria": ["Criterion 1", "Criterion 2"],
|
|
"dependencies": [],
|
|
"files_to_modify": ["path/to/file.dart"],
|
|
"component": "{self.config.component}"
|
|
}}
|
|
]
|
|
}}
|
|
```
|
|
|
|
### Guidelines:
|
|
|
|
- Create 5-10 actionable tasks
|
|
- Prioritize based on business value and dependencies
|
|
- Each task should be 2-8 hours of work
|
|
- Include clear acceptance criteria
|
|
- Specify files to modify
|
|
- Set proper dependencies
|
|
|
|
Start planning now!
|
|
"""
|
|
|
|
return prompt
|
|
|
|
def _gather_context(self) -> str:
|
|
"""Gather context from existing files."""
|
|
context = ""
|
|
|
|
# Read tasks.md if exists
|
|
tasks_md = self.config.component_root / "tasks.md"
|
|
if not tasks_md.exists():
|
|
tasks_md = self.config.project_root / "mnemo_cards_web_v2" / "tasks.md"
|
|
|
|
if tasks_md.exists():
|
|
with open(tasks_md, 'r') as f:
|
|
content = f.read()
|
|
# Truncate if too long
|
|
if len(content) > 10000:
|
|
content = content[:10000] + "\n... (truncated)"
|
|
context += f"### tasks.md:\n```\n{content}\n```\n\n"
|
|
|
|
# Read workflow_state.md if exists
|
|
workflow_state = self.config.component_root / "workflow_state.md"
|
|
if not workflow_state.exists():
|
|
workflow_state = self.config.project_root / "mnemo_cards_web_v2" / "workflow_state.md"
|
|
|
|
if workflow_state.exists():
|
|
with open(workflow_state, 'r') as f:
|
|
content = f.read()
|
|
if len(content) > 5000:
|
|
content = content[:5000] + "\n... (truncated)"
|
|
context += f"### workflow_state.md:\n```\n{content}\n```\n\n"
|
|
|
|
# Read current task list if exists
|
|
if self.config.task_list_path.exists():
|
|
with open(self.config.task_list_path, 'r') as f:
|
|
current_tasks = json.load(f)
|
|
context += f"### Current task_list.json:\n```json\n{json.dumps(current_tasks, indent=2)}\n```\n\n"
|
|
|
|
# Read agent state
|
|
if self.config.agent_state_path.exists():
|
|
with open(self.config.agent_state_path, 'r') as f:
|
|
state = json.load(f)
|
|
context += f"### Current agent_state.json:\n```json\n{json.dumps(state, indent=2)}\n```\n\n"
|
|
|
|
if not context:
|
|
context = "No existing context files found. Create fresh task list based on project structure.\n"
|
|
|
|
return context
|
|
|
|
def _load_and_validate_task_list(self) -> Dict[str, Any]:
|
|
"""Load and validate generated task list."""
|
|
try:
|
|
with open(self.config.task_list_path, 'r') as f:
|
|
task_list = json.load(f)
|
|
|
|
# Validate structure
|
|
if "tasks" not in task_list:
|
|
print("❌ Task list missing 'tasks' field")
|
|
return None
|
|
|
|
tasks = task_list["tasks"]
|
|
|
|
if not isinstance(tasks, list):
|
|
print("❌ 'tasks' must be a list")
|
|
return None
|
|
|
|
# Validate each task
|
|
required_fields = [
|
|
"id", "title", "priority", "status",
|
|
"estimated_hours", "description",
|
|
"acceptance_criteria", "dependencies",
|
|
"files_to_modify", "component"
|
|
]
|
|
|
|
for i, task in enumerate(tasks):
|
|
for field in required_fields:
|
|
if field not in task:
|
|
print(f"❌ Task {i} missing required field: {field}")
|
|
return None
|
|
|
|
return task_list
|
|
|
|
except json.JSONDecodeError as e:
|
|
print(f"❌ Invalid JSON in task list: {e}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"❌ Error loading task list: {e}")
|
|
return None
|
|
|
|
def _print_summary(self, task_list: Dict[str, Any]) -> None:
|
|
"""Print task list summary."""
|
|
tasks = task_list.get("tasks", [])
|
|
|
|
print(f"\n{'='*80}")
|
|
print(f"📋 Task List Summary for {self.config.component}")
|
|
print(f"{'='*80}")
|
|
print(f"Total tasks: {len(tasks)}")
|
|
|
|
# Count by priority
|
|
high_count = sum(1 for t in tasks if t.get("priority") == "high")
|
|
medium_count = sum(1 for t in tasks if t.get("priority") == "medium")
|
|
low_count = sum(1 for t in tasks if t.get("priority") == "low")
|
|
|
|
print(f"\nPriority breakdown:")
|
|
print(f" 🔴 HIGH: {high_count}")
|
|
print(f" 🟡 MEDIUM: {medium_count}")
|
|
print(f" 🟢 LOW: {low_count}")
|
|
|
|
# Total estimated hours
|
|
total_hours = sum(t.get("estimated_hours", 0) for t in tasks)
|
|
print(f"\nTotal estimated: {total_hours:.1f} hours")
|
|
|
|
# List high priority tasks
|
|
print(f"\n🔴 High Priority Tasks:")
|
|
for task in tasks:
|
|
if task.get("priority") == "high":
|
|
print(f" - {task['id']}: {task['title']} ({task['estimated_hours']}h)")
|
|
|
|
print(f"\n{'='*80}")
|
|
|
|
def _commit_task_list(self) -> None:
|
|
"""Commit the generated task list."""
|
|
import subprocess
|
|
|
|
# Add task list file
|
|
subprocess.run(
|
|
["git", "add", str(self.config.task_list_path)],
|
|
cwd=self.config.project_root
|
|
)
|
|
|
|
# Commit
|
|
commit_message = f"""plan({self.config.component}): Update task list
|
|
|
|
Generated by: Planning Agent
|
|
Timestamp: {datetime.now(timezone.utc).isoformat()}
|
|
Component: {self.config.component}
|
|
"""
|
|
|
|
result = subprocess.run(
|
|
["git", "commit", "-m", commit_message],
|
|
cwd=self.config.project_root,
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
print("✅ Task list committed")
|
|
|
|
# Push
|
|
subprocess.run(
|
|
["git", "push"],
|
|
cwd=self.config.project_root
|
|
)
|
|
print("✅ Pushed to remote")
|
|
else:
|
|
print("⚠️ No changes to commit (task list unchanged)")
|
|
|
|
|
|
def main():
|
|
"""Main entry point."""
|
|
if len(sys.argv) < 2:
|
|
print("Usage: python planning_agent.py <component>")
|
|
print("Components: web_v2, backend, common")
|
|
sys.exit(1)
|
|
|
|
component = sys.argv[1]
|
|
|
|
# Load configuration
|
|
config = AgentConfig.from_env(component)
|
|
|
|
try:
|
|
config.validate()
|
|
except ValueError as e:
|
|
print(f"❌ Configuration error: {e}")
|
|
sys.exit(1)
|
|
|
|
# Create planning agent and run
|
|
agent = PlanningAgent(config)
|
|
exit_code = agent.run()
|
|
|
|
sys.exit(exit_code)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|