backend/utils/agent_loop.py

"""Agent loop orchestration - the core of Manus autonomous agent."""
from typing import Optional, AsyncGenerator
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import json
import asyncio

from backend.services import get_ai_gateway, get_e2b, get_supabase, ModelRole
from backend.utils.tools import create_tool_registry, ToolRegistry, BaseTool
from backend.models.schemas import TaskStatus


class AgentState(str, Enum):
    IDLE = "idle"
    PLANNING = "planning"
    EXECUTING = "executing"
    WAITING = "waiting"
    COMPLETE = "complete"
    FAILED = "failed"


@dataclass
class AgentThought:
    """Represents a thought in the agent's reasoning chain."""
    timestamp: datetime
    thinking: str
    state: AgentState
    tool_used: Optional[str] = None
    result: Optional[str] = None
    success: bool = True


@dataclass
class ToolResult:
    """Structured result from tool execution."""
    tool_name: str
    success: bool
    output: str
    error: Optional[str] = None
    execution_time_ms: float = 0
    retry_count: int = 0


class AutonomousAgent:
    """
    Manus - An autonomous AI agent capable of planning, executing, and self-correcting.
    
    Key features:
    - Multi-step task planning
    - Tool execution with retry logic
    - Self-correction on failures
    - Memory of thought process
    - Web search and research capabilities
    """
    
    MAX_ITERATIONS = 50
    MAX_RETRIES = 3
    RETRY_DELAY = 2  # seconds
    
    PLANNING_PROMPT = """You are Manus, an autonomous AI agent with access to powerful tools.

Given a task, break it down into specific, actionable steps. For each step:
1. Identify the appropriate tool to use
2. Determine the exact parameters
3. Define what success looks like

Available tools:
- terminal: Execute shell commands, run scripts, install packages
- file: Read, write, list, or delete files
- tavily_search: Search the web for current information
- tavily_extract: Extract content from specific URLs  
- tavily_crawl: Crawl entire websites
- python: Execute Python code (use terminal with python3)

Format your response as JSON:
{
    "steps": [
        {
            "description": "Clear description of the action",
            "tool": "tool_name",
            "params": {"param1": "value1"},
            "expected_output": "What success looks like"
        }
    ],
    "reasoning": "Why this approach will work"
}

Be specific and actionable. Avoid vague instructions."""
    
    def __init__(self, task_id: str, description: str, sandbox=None, tavily_api_key: str = None, enable_playwright: bool = False):
        self.task_id = task_id
        self.description = description
        self.sandbox = sandbox
        self.tavily_api_key = tavily_api_key
        self.enable_playwright = enable_playwright
        self.state = AgentState.IDLE
        self.thoughts: list[AgentThought] = []
        self.current_step = 0
        self.steps: list[dict] = []
        self.results: list[ToolResult] = []
        
        self.ai = get_ai_gateway()
        self.supabase = get_supabase()
        self.tools: Optional[ToolRegistry] = None
    
    async def initialize(self):
        """Initialize the agent with tools and sandbox."""
        google_creds = None
        self.tools = create_tool_registry(self.sandbox, google_creds, self.tavily_api_key, self.enable_playwright)
    
    async def think(self, prompt: str, role: ModelRole = ModelRole.PLANNING) -> str:
        """Generate thoughts using AI."""
        return await self.ai.complete(prompt, role=role)
    
    async def execute_tool(self, tool_name: str, params: dict, step_description: str) -> ToolResult:
        """Execute a single tool with retry logic."""
        start_time = datetime.utcnow()
        retry_count = 0
        
        for attempt in range(self.MAX_RETRIES):
            try:
                # Log attempt
                self.thoughts.append(AgentThought(
                    timestamp=datetime.utcnow(),
                    thinking=f"Executing: {step_description} (attempt {attempt + 1})",
                    state=AgentState.EXECUTING,
                    tool_used=tool_name
                ))
                
                # Execute tool
                result = await self.tools.execute(tool_name, **params)
                
                execution_time = (datetime.utcnow() - start_time).total_seconds() * 1000
                
                if result.get("success"):
                    output = self._format_tool_output(result)
                    self.thoughts.append(AgentThought(
                        timestamp=datetime.utcnow(),
                        thinking=f"Result: {output[:200]}",
                        state=AgentState.EXECUTING,
                        tool_used=tool_name,
                        result=output,
                        success=True
                    ))
                    
                    return ToolResult(
                        tool_name=tool_name,
                        success=True,
                        output=output,
                        execution_time_ms=execution_time,
                        retry_count=retry_count
                    )
                else:
                    error = result.get("error", "Unknown error")
                    if attempt < self.MAX_RETRIES - 1:
                        retry_count += 1
                        self.thoughts.append(AgentThought(
                            timestamp=datetime.utcnow(),
                            thinking=f"Failed: {error}. Retrying in {self.RETRY_DELAY}s...",
                            state=AgentState.WAITING,
                            tool_used=tool_name,
                            success=False
                        ))
                        await asyncio.sleep(self.RETRY_DELAY)
                        continue
                    else:
                        return ToolResult(
                            tool_name=tool_name,
                            success=False,
                            output="",
                            error=error,
                            execution_time_ms=execution_time,
                            retry_count=retry_count
                        )
                        
            except Exception as e:
                if attempt < self.MAX_RETRIES - 1:
                    retry_count += 1
                    await asyncio.sleep(self.RETRY_DELAY)
                    continue
                else:
                    execution_time = (datetime.utcnow() - start_time).total_seconds() * 1000
                    return ToolResult(
                        tool_name=tool_name,
                        success=False,
                        output="",
                        error=str(e),
                        execution_time_ms=execution_time,
                        retry_count=retry_count
                    )
        
        return ToolResult(
            tool_name=tool_name,
            success=False,
            output="",
            error="Max retries exceeded",
            retry_count=retry_count
        )
    
    def _format_tool_output(self, result: dict) -> str:
        """Format tool result into readable output."""
        if "stdout" in result:
            return result["stdout"]
        elif "content" in result:
            return result["content"]
        elif "results" in result:
            if isinstance(result["results"], list):
                return json.dumps(result["results"], indent=2)
            return str(result["results"])
        elif "answer" in result:
            return result["answer"]
        elif "output" in result:
            return result["output"]
        else:
            return json.dumps(result, indent=2)
    
    async def parse_steps(self, plan_response: str) -> list[dict]:
        """Parse steps from planning response."""
        try:
            plan_data = json.loads(plan_response)
            return plan_data.get("steps", [])
        except json.JSONDecodeError:
            return self._parse_text_steps(plan_response)
    
    def _parse_text_steps(self, text: str) -> list[dict]:
        """Parse steps from plain text format."""
        steps = []
        lines = text.split("\n")
        
        for i, line in enumerate(lines):
            line = line.strip()
            if not line or line.startswith("#") or line.startswith("//"):
                continue
            
            tool = None
            params = {}
            
            if "terminal" in line.lower() or "bash" in line.lower() or "command" in line.lower():
                tool = "terminal"
                params = {"command": line.split("```")[-1].strip() if "```" in line else line}
            elif "search" in line.lower() or "google" in line.lower() or "find" in line.lower():
                tool = "tavily_search"
                params = {"query": line.split("search")[-1].strip() if "search" in line.lower() else line}
            elif "file" in line.lower() or "write" in line.lower() or "create" in line.lower():
                tool = "file"
                params = {"action": "write", "path": "/tmp/output.txt", "content": line}
            else:
                tool = "terminal"
                params = {"command": line}
            
            steps.append({
                "description": line,
                "tool": tool,
                "params": params
            })
        
        return steps
    
    async def self_correct(self, failed_step: dict, error: str, context: list[ToolResult]) -> dict:
        """Attempt to self-correct after a failed step."""
        self.thoughts.append(AgentThought(
            timestamp=datetime.utcnow(),
            thinking=f"Analyzing failure: {error}",
            state=AgentState.PLANNING,
            tool_used=None
        ))
        
        context_summary = "\n".join([
            f"- {r.tool_name}: {'SUCCESS' if r.success else 'FAILED'}: {r.output[:100] if r.output else r.error}"
            for r in context[-5:]
        ])
        
        correction_prompt = f"""The following step failed:
Step: {failed_step.get('description', 'Unknown')}
Tool: {failed_step.get('tool', 'Unknown')}
Error: {error}

Previous context:
{context_summary}

Task: {self.description}

Suggest an alternative approach. Return JSON:
{{
    "analysis": "Why this failed",
    "alternative": "What to try instead",
    "new_tool": "tool_name or null",
    "new_params": {{}} or null
}}"""
        
        try:
            response = await self.think(correction_prompt, role=ModelRole.PLANNING)
            correction = json.loads(response)
            
            if correction.get("new_tool"):
                self.thoughts.append(AgentThought(
                    timestamp=datetime.utcnow(),
                    thinking=f"Correction: Using {correction['new_tool']} instead",
                    state=AgentState.PLANNING,
                    tool_used=correction["new_tool"]
                ))
                
                return {
                    "description": correction.get("alternative", failed_step.get("description")),
                    "tool": correction["new_tool"],
                    "params": correction.get("new_params", {})
                }
        except:
            pass
        
        return {
            "description": f"Retry: {failed_step.get('description', 'Unknown')}",
            "tool": failed_step.get("tool", "terminal"),
            "params": failed_step.get("params", {})
        }
    
    async def run(self) -> AsyncGenerator[dict, None]:
        """Run the complete agent loop with self-correction."""
        try:
            self.state = AgentState.PLANNING
            
            yield {
                "type": "status",
                "content": "🚀 Manus initializing...",
                "timestamp": datetime.utcnow().isoformat()
            }
            
            await self.initialize()
            
            yield {
                "type": "status", 
                "content": f"📦 Tools loaded: {[t['name'] for t in self.tools.list_tools()]}",
                "timestamp": datetime.utcnow().isoformat()
            }
            
            # Phase 1: Planning
            yield {
                "type": "thought",
                "content": "🧠 Planning task decomposition...",
                "timestamp": datetime.utcnow().isoformat()
            }
            
            plan_response = await self.think(
                f"{self.PLANNING_PROMPT}\n\nTask: {self.description}",
                role=ModelRole.PLANNING
            )
            
            yield {
                "type": "thought",
                "content": f"📋 Plan generated:\n{plan_response[:500]}...",
                "timestamp": datetime.utcnow().isoformat()
            }
            
            self.steps = await self.parse_steps(plan_response)
            
            if not self.steps:
                self.steps = [{
                    "description": f"Execute task: {self.description}",
                    "tool": "terminal",
                    "params": {"command": f"echo 'Task: {self.description}'"}
                }]
            
            yield {
                "type": "status",
                "content": f"📌 Executing {len(self.steps)} steps",
                "timestamp": datetime.utcnow().isoformat()
            }
            
            # Phase 2: Execution with self-correction
            self.state = AgentState.EXECUTING
            
            for i, step in enumerate(self.steps):
                if i >= self.MAX_ITERATIONS:
                    yield {
                        "type": "error",
                        "content": "Max iterations reached",
                        "timestamp": datetime.utcnow().isoformat()
                    }
                    break
                
                self.current_step = i + 1
                
                yield {
                    "type": "action",
                    "content": f"⚡ Step {self.current_step}/{len(self.steps)}: {step.get('description', 'Unknown')[:80]}",
                    "timestamp": datetime.utcnow().isoformat()
                }
                
                result = await self.execute_tool(
                    step.get("tool", "terminal"),
                    step.get("params", {}),
                    step.get("description", "Unknown step")
                )
                
                self.results.append(result)
                
                if result.success:
                    yield {
                        "type": "success",
                        "content": f"✅ {result.tool_name}: {result.output[:300]}",
                        "timestamp": datetime.utcnow().isoformat(),
                        "metadata": {
                            "tool": result.tool_name,
                            "execution_time_ms": result.execution_time_ms,
                            "retries": result.retry_count
                        }
                    }
                else:
                    yield {
                        "type": "warning",
                        "content": f"⚠️ {result.tool_name} failed: {result.error}",
                        "timestamp": datetime.utcnow().isoformat()
                    }
                    
                    if result.retry_count > 0:
                        corrected_step = await self.self_correct(step, result.error, self.results)
                        
                        yield {
                            "type": "thought",
                            "content": "🔧 Attempting self-correction...",
                            "timestamp": datetime.utcnow().isoformat()
                        }
                        
                        corrected_result = await self.execute_tool(
                            corrected_step.get("tool", "terminal"),
                            corrected_step.get("params", {}),
                            corrected_step.get("description", "Corrected step")
                        )
                        
                        if corrected_result.success:
                            yield {
                                "type": "success",
                                "content": f"✅ Self-corrected: {corrected_result.output[:300]}",
                                "timestamp": datetime.utcnow().isoformat()
                            }
                            self.results.append(corrected_result)
                        else:
                            yield {
                                "type": "error",
                                "content": f"❌ Self-correction failed: {corrected_result.error}",
                                "timestamp": datetime.utcnow().isoformat()
                            }
                
                await self.supabase.update_task(
                    self.task_id,
                    metadata={
                        "current_step": self.current_step,
                        "total_steps": len(self.steps),
                        "progress": self.current_step / len(self.steps),
                        "last_tool": result.tool_name,
                        "last_success": result.success
                    }
                )
            
            # Phase 3: Completion
            self.state = AgentState.COMPLETE
            
            success_count = sum(1 for r in self.results if r.success)
            total_time = sum(r.execution_time_ms for r in self.results)
            
            final_result = {
                "completed_steps": success_count,
                "total_steps": len(self.steps),
                "total_execution_time_ms": total_time,
                "results": [
                    {"tool": r.tool_name, "success": r.success, "output": r.output[:500], "error": r.error}
                    for r in self.results
                ],
                "thoughts": len(self.thoughts)
            }
            
            yield {
                "type": "result",
                "content": f"🎉 Task completed! {success_count}/{len(self.steps)} steps successful.",
                "timestamp": datetime.utcnow().isoformat(),
                "metadata": final_result
            }
            
            await self.supabase.update_task(
                self.task_id,
                status=TaskStatus.COMPLETED.value,
                result=final_result,
                completed_at=datetime.utcnow()
            )
            
        except Exception as e:
            self.state = AgentState.FAILED
            
            yield {
                "type": "error",
                "content": f"💥 Agent failed: {str(e)}",
                "timestamp": datetime.utcnow().isoformat()
            }
            
            await self.supabase.update_task(
                self.task_id,
                status=TaskStatus.FAILED.value,
                error=str(e),
                completed_at=datetime.utcnow()
            )
    
    def get_thought_history(self) -> list[dict]:
        """Get the history of agent thoughts."""
        return [
            {
                "timestamp": t.timestamp.isoformat(),
                "thinking": t.thinking,
                "state": t.state.value,
                "tool": t.tool_used,
                "success": t.success
            }
            for t in self.thoughts
        ]


# Keep legacy class name for compatibility
AgentLoop = AutonomousAgent