Blame for nps_agent.py

import asyncio
import os
import json
import logging
import sys
from typing import Any, Optional
from dotenv import load_dotenv
from pathlib import Path

# Ensure we can import nanobot if installed in venv
from playwright.async_api import async_playwright, BrowserContext, Page, Browser

try:
    from nanobot.agent.loop import AgentLoop
    from nanobot.agent.tools.base import Tool
    from nanobot.bus.queue import MessageBus
    from nanobot.providers.litellm_provider import LiteLLMProvider
    from nanobot.config.schema import ExecToolConfig
except ImportError:
    print("Error: 'nanobot' not found. Please install 'nanobot-ai'.")
    # We don't exit here so it can be imported safely, but it will likely fail on run
    pass

# Load environment variables
load_dotenv()

# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger("nps_nanobot")

class PlaywrightTool(Tool):
    """
    A tool to control a web browser using Playwright.
    Supports navigation, clicking, typing, and extracting text.
    """
    name = "browser"
    description = (
        "Control a web browser to navigate, interact, and extract information. "
        "Use this tool to visit websites, click elements, fill forms, and read content."
    )
    parameters = {
        "type": "object",
        "properties": {
            "action": {
                "type": "string",
                "enum": ["navigate", "click", "type", "press", "scroll", "get_text", "get_links", "get_dom", "screenshot"],
                "description": "The action to perform."
            },
            "url": {"type": "string", "description": "URL to navigate to (required for 'navigate')"},
            "selector": {"type": "string", "description": "CSS selector or text to interact with (required for 'click', 'type', 'get_text')"},
            "text": {"type": "string", "description": "Text to type (required for 'type')"},
            "key": {"type": "string", "description": "Key to press (required for 'press', e.g., 'Enter')"},
            "direction": {"type": "string", "enum": ["up", "down"], "description": "Scroll direction (required for 'scroll')"},
            "amount": {"type": "integer", "description": "Scroll amount in pixels (optional for 'scroll')"},
        },
        "required": ["action"]
    }

    def __init__(self):
        self._playwright = None
        self._browser = None
        self._context: Optional[BrowserContext] = None
        self._page: Optional[Page] = None
        self._initialized = False

    async def _ensure_initialized(self):
        if not self._initialized:
            self._playwright = await async_playwright().start()
            
            # Auto-detect HF Spaces or allow HEADLESS env var override
            is_spaces = "SPACE_ID" in os.environ
            default_headless = "true" if is_spaces else "false"
            is_headless = os.getenv("HEADLESS", default_headless).lower() == "true"

            # Launch with anti-detection args
            self._browser = await self._playwright.chromium.launch(
                headless=is_headless,
                args=[
                    "--no-sandbox",
                    "--disable-blink-features=AutomationControlled",
                    "--disable-infobars"
                ],
                ignore_default_args=["--enable-automation"]
            )
            self._context = await self._browser.new_context(
                viewport={"width": 1280, "height": 720},
                user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
            )
            self._page = await self._context.new_page()
            self._page.set_default_timeout(15000) # 15s timeout
            
            # Auto-handle native dialogs
            self._page.on("dialog", lambda dialog: dialog.accept())
            
            self._initialized = True
            logger.info("Playwright browser initialized")

    async def _handle_popups(self):
        """Try to dismiss common cookie banners and popups."""
        try:
            # Common patterns for dismissal
            patterns = [
                "text=Accept", "text=Agree", "text=Allow", 
                "text=Reject", "text=Close", "text=No thanks", 
                "text=Not now", "text=OK", "text=I understand",
                "button[aria-label='Close']", ".close-button",
                "#dismiss-button", "div[aria-label='Close ad']",
                ".ri-close-line", ".close-icon"
            ]
            for selector in patterns:
                try:
                    # Check if visible and click. fast timeout.
                    elem = await self._page.query_selector(selector)
                    if elem and await elem.is_visible():
                        await elem.click(timeout=500)
                        logger.info(f"Dismissed popup using {selector}")
                        await asyncio.sleep(0.5) # Wait for animation
                except Exception:
                    pass
        except Exception as e:
            logger.warning(f"Popup dismissal error: {e}")

    async def execute(self, action: str, **kwargs: Any) -> str:
        try:
            await self._ensure_initialized()
            page = self._page
            
            if action == "navigate":
                url = kwargs.get("url")
                if not url: return "Error: 'url' is required for navigate"
                if not url.startswith("http"): url = "https://" + url
                try:
                    await page.goto(url, timeout=20000)
                    await page.wait_for_load_state("domcontentloaded", timeout=10000)
                    # Attempt to clean up view
                    await self._handle_popups()
                except Exception as e:
                    # Ignore timeout and assume page loaded partially
                    logger.warning(f"Navigation to {url} timed out: {e}")
                    await self._handle_popups() # Try finding popups even if timed out
                    return f"Navigated to {url}, but timed out waiting for full load. Content might be available."
                
                title = await page.title()
                return f"Navigated to {url}. Page title: {title}"

            elif action == "click":
                selector = kwargs.get("selector")
                if not selector: return "Error: 'selector' is required for click"
                # specialized selector handling: try text if loose selector
                try:
                    # If it looks like "link: <text>", extract
                    if selector.startswith("link: "):
                        selector = f"text={selector[6:]}"
                    await page.click(selector, timeout=5000)
                except Exception:
                    # Fallback to text match
                    await page.click(f"text={selector}")
                return f"Clicked element: {selector}"

            elif action == "type":
                selector = kwargs.get("selector")
                text = kwargs.get("text")
                if not selector or text is None: return "Error: 'selector' and 'text' are required for type"
                await page.fill(selector, text)
                return f"Typed '{text}' into {selector}"

            elif action == "press":
                key = kwargs.get("key")
                if not key: return "Error: 'key' is required for press"
                await page.keyboard.press(key)
                return f"Pressed key: {key}"

            elif action == "scroll":
                direction = kwargs.get("direction", "down")
                amount = kwargs.get("amount", 500)
                if direction == "down":
                    await page.evaluate(f"window.scrollBy(0, {amount})")
                else:
                    await page.evaluate(f"window.scrollBy(0, -{amount})")
                return f"Scrolled {direction} by {amount} pixels"

            elif action == "get_text":
                selector = kwargs.get("selector", "body")
                if selector == "body":
                    # Smart extraction: get text but also hints of links
                    # We'll stick to text for this action to keep it simple, 
                    # but maybe use Readability if we had it. For now, innerText is fine.
                    content = await page.evaluate("document.body.innerText")
                    return content[:4000] + "... (truncated)"
                else:
                    element = await page.query_selector(selector)
                    if element:
                        text = await element.inner_text()
                        return text
                    return f"Element not found: {selector}"
            
            elif action == "get_links":
                # new action to find clickable links
                links = await page.evaluate("""() => {
                    const params = [];
                    document.querySelectorAll('a').forEach(a => {
                        if (a.innerText.trim().length > 3 && a.offsetParent !== null) {
                            params.push({text: a.innerText.trim().slice(0, 50), href: a.href});
                        }
                    });
                    return params.slice(0, 20); // Top 20 relevant links
                }""")
                return json.dumps(links, indent=2)

            elif action == "get_dom":
                # Simplified interactive DOM tree for buttons and inputs
                dom = await page.evaluate("""() => {
                    const params = [];
                    const elements = document.querySelectorAll('button, input, select, textarea, [role="button"]');
                    elements.forEach(el => {
                        if (el.offsetParent !== null) { // visible
                            let label = el.innerText || el.placeholder || el.getAttribute('aria-label') || el.name || el.id || el.value;
                            label = label ? label.trim().slice(0, 50) : '[No Label]';
                            const tag = el.tagName.toLowerCase();
                            params.push(`${tag}: ${label}`);
                        }
                    });
                    return params.slice(0, 50).join('\\n'); // Top 50 interactive elements
                }""")
                return dom if dom else "No interactive elements found."

            elif action == "screenshot":
                path = "screenshot.png"
                await page.screenshot(path=path)
                return f"Screenshot saved to {path}"

            else:
                return f"Unknown action: {action}"

        except Exception as e:
            logger.error(f"Playwright error: {e}")
            return f"Error executing {action}: {str(e)}"

    async def close(self):
        if self._browser:
            await self._browser.close()
        if self._playwright:
            await self._playwright.stop()


async def run_nps_task(task: str) -> str:
    """
    Executes a task using the NPS Nanobot Agent.
    
    Args:
        task: The natural language task description.
        
    Returns:
        The result of the task execution.
    """
    print("\n" + "="*60)
    print("🐈  NPS OPERATIONAL AGENT (Nanobot + Playwright + Ollama)")
    print("="*60)

    # Initialize bus
    bus = MessageBus()
    
    # Configure LiteLLM for Mistral API (same key as the main app)
    model_name = "mistral/mistral-small-latest"
    
    mistral_api_key = os.getenv("MISTRAL_API_KEY")
    if not mistral_api_key:
        logger.warning("MISTRAL_API_KEY not found in environment variables. Agent may fail.")

    # Initialize LiteLLM provider for Mistral
    class MistralProvider(LiteLLMProvider):
        async def chat(self, messages, tools=None, model=None, max_tokens=4096, temperature=0.7):
            model = model or self.default_model
            
            kwargs = {
                "model": model,
                "messages": messages,
                "max_tokens": max_tokens,
                "temperature": temperature,
                "api_key": os.getenv("MISTRAL_API_KEY", ""),
            }
            if tools:
                kwargs["tools"] = tools
                kwargs["tool_choice"] = "auto"
                
            max_retries = 5
            for attempt in range(max_retries):
                try:
                    from litellm import acompletion
                    from nanobot.providers.base import LLMResponse, ToolCallRequest
                    import json
                    
                    response = await acompletion(**kwargs)
                    llm_response = self._parse_response(response)
                    
                    # Fallback: if no tool calls but content looks like JSON tool call, parse it
                    if not llm_response.tool_calls and llm_response.content:
                        import re
                        content = llm_response.content.strip()
                        json_match = re.search(r'(\[.*\]|\{.*\})', content, re.DOTALL)
                        
                        if json_match:
                            try:
                                json_str = json_match.group(1)
                                try:
                                    data = json.loads(json_str)
                                except json.JSONDecodeError:
                                    data = []
                                    for line in json_str.splitlines():
                                        line = line.strip()
                                        if line:
                                            try:
                                                obj = json.loads(line)
                                                data.append(obj)
                                            except json.JSONDecodeError:
                                                pass
                                    if not data: raise
                                
                                tool_calls = []
                                if isinstance(data, list):
                                    for item in data:
                                        if isinstance(item, dict) and "name" in item and "arguments" in item:
                                            args = item["arguments"]
                                            if isinstance(args, str):
                                                try: args = json.loads(args)
                                                except: pass
                                            import uuid
                                            tool_calls.append(ToolCallRequest(
                                                id=f"call_{uuid.uuid4().hex[:8]}",
                                                name=item["name"],
                                                arguments=args
                                            ))
                                        elif isinstance(item, dict) and "type" in item and item["type"] == "function":
                                            func = item["function"]
                                            args = func["arguments"]
                                            if isinstance(args, str):
                                                try: args = json.loads(args)
                                                except: pass
                                            import uuid
                                            tool_calls.append(ToolCallRequest(
                                                id=item.get("id", f"call_{uuid.uuid4().hex[:8]}"),
                                                name=func["name"],
                                                arguments=args
                                            ))
                                elif isinstance(data, dict):
                                    if "name" in data and "arguments" in data:
                                        args = data["arguments"]
                                        if isinstance(args, str):
                                            try: args = json.loads(args)
                                            except: pass
                                        import uuid
                                        tool_calls.append(ToolCallRequest(
                                            id=f"call_{uuid.uuid4().hex[:8]}",
                                            name=data["name"],
                                            arguments=args
                                        ))

                                if tool_calls:
                                    llm_response.tool_calls = tool_calls
                                    llm_response.content = None
                                    llm_response.finish_reason = "tool_calls"
                                    
                            except Exception as e:
                                logger.warning(f"Failed to parse JSON tool call fallback: {e}")
                    
                    return llm_response

                except Exception as e:
                    if attempt == max_retries - 1:
                        from nanobot.providers.base import LLMResponse
                        return LLMResponse(content=f"Error calling LLM after {max_retries} attempts: {str(e)}", finish_reason="error")
                    logger.warning(f"LLM call failed (attempt {attempt+1}/{max_retries}): {e}")
            
            return LLMResponse(content="Error: Max retries exceeded", finish_reason="error")

    provider = MistralProvider(
        default_model=model_name,
        api_key=mistral_api_key
    )

    # Initialize Agent Loop
    # We use current directory as workspace
    workspace_path = Path(os.getcwd())

    # Clear memory to ensure fresh start
    memory_path = workspace_path / "memory"
    if memory_path.exists():
        import shutil
        try:
            shutil.rmtree(memory_path)
            print("INFO: Cleared previous agent memory.")
        except Exception as e:
            print(f"WARN: Could not clear memory: {e}")

    agent = AgentLoop(
        bus=bus,
        provider=provider,
        workspace=workspace_path,
        model=model_name,
        max_iterations=15,
    )

    # Register our custom Playwright tool
    playwright_tool = PlaywrightTool()
    agent.tools.register(playwright_tool)

    print(f"\nINFO: Agent initialized with model: {model_name}")
    print("INFO: Browser tool (Playwright) registered")

    try:
        if not task:
            return "❌ No task provided."

        print(f"\n🚀 Starting task: {task}")
        
        # Append strong instruction to the task
        system_instruction = (
             "\n\nIMPORTANT INSTRUCTIONS:"
             "\n1. You are an autonomous browser agent. You MUST USE TOOLS to perform actions."
             "\n2. DO NOT narrate what you are going to do (e.g., 'I will click...'). JUST CALL THE TOOL."
             "\n3. If the task is not complete, your response MUST contain a tool call."
             "\n4. Only output text when you have fully completed the user's request and have the answer."
             "\n5. Start by navigating: browser(action='navigate', ...)."
             "\n6. Use 'get_links' to find clickable items."
             "\n7. Use 'get_dom' to find buttons and inputs on interactive pages."
             "\n8. If you see popups, 'OK' dialogs, or cookie banners that block you, click them to dismiss."
        )
        full_task = task + system_instruction
        
        # Use a unique session ID for each task execution to prevent history leakage
        import uuid
        session_id = f"task_{uuid.uuid4().hex[:8]}"
        
        # Run the agent with a unique session key
        response = await agent.process_direct(
            full_task,
            session_key=f"cli:{session_id}",
            channel="cli",
            chat_id=session_id
        )
        
        print("\n" + "="*60)
        print("✅ TASK COMPLETED")
        print("="*60)
        return response

    except KeyboardInterrupt:
        return "Task interrupted by user."
    finally:
        # Cleanup
        await playwright_tool.close()
        agent.stop()

if __name__ == "__main__":
    # Test block
    task = input("Enter test task: ")
    asyncio.run(run_nps_task(task))