""" Plan Manager for CodeAct Agent. Handles plan creation, updates, and progress tracking. """ import re from typing import Optional, Dict class PlanManager: """Manages plan creation, updates, and progress tracking.""" @staticmethod def extract_plan_from_content(content: str) -> Optional[str]: """Extract plan from agent content.""" plan_pattern = r'\d+\.\s*\[[^\]]*\]\s*[^\n]+(?:\n\d+\.\s*\[[^\]]*\]\s*[^\n]+)*' matches = re.findall(plan_pattern, content) # Return the last (most recent) plan if multiple plans exist return matches[-1] if matches else None @staticmethod def update_plan_for_solution(plan_text: str) -> str: """Update plan to mark all remaining steps as completed when providing final solution.""" if not plan_text: return plan_text lines = plan_text.split('\n') updated_lines = [] for line in lines: # Mark any unchecked or failed steps as completed since we're providing final solution if '[ ]' in line or '[✗]' in line: updated_line = re.sub(r'\[\s*[^\]]*\]', '[✓]', line) updated_lines.append(updated_line) else: updated_lines.append(line) return '\n'.join(updated_lines) @staticmethod def mark_step_completed(plan_text: str, step_description: str) -> str: """Mark a specific step as completed based on its description.""" if not plan_text or not step_description: return plan_text lines = plan_text.split('\n') updated_lines = [] for line in lines: # Check if this line contains the step description and is unchecked if step_description.lower() in line.lower() and ('[ ]' in line or '[✗]' in line): updated_line = re.sub(r'\[\s*[^\]]*\]', '[✓]', line) updated_lines.append(updated_line) else: updated_lines.append(line) return '\n'.join(updated_lines) @staticmethod def get_plan_progress(plan_text: str) -> Dict[str, int]: """Get plan progress statistics.""" if not plan_text: return {"total": 0, "completed": 0, "pending": 0, "failed": 0} lines = plan_text.split('\n') stats = {"total": 0, "completed": 0, "pending": 0, "failed": 0} for line in lines: if re.search(r'\d+\.\s*\[', line): stats["total"] += 1 if '[✓]' in line: stats["completed"] += 1 elif '[ ]' in line: stats["pending"] += 1 elif '[✗]' in line: stats["failed"] += 1 return stats