feat: implement multi-agent system with coordinator

backend/app/agents/__init__.py

@@ -0,0 +1,38 @@
+"""
+Agents module for ScrapeRL.
+
+This module contains specialized agents for web scraping with RL:
+- BaseAgent: Abstract base class for all agents
+- PlannerAgent: Goal decomposition and task planning
+- NavigatorAgent: URL prioritization and page navigation
+- ExtractorAgent: Data extraction with selectors
+- VerifierAgent: Cross-source verification
+- MemoryAgent: Memory operations and knowledge management
+- AgentCoordinator: Orchestrates multiple agents with message passing
+"""
+
+from .base import BaseAgent
+from .coordinator import AgentCoordinator, AgentRole, Message
+from .extractor import ExtractorAgent
+from .memory_agent import MemoryAgent, MemoryEntry
+from .navigator import NavigatorAgent
+from .planner import PlannerAgent
+from .verifier import VerificationResult, VerifierAgent
+
+__all__ = [
+    # Base
+    "BaseAgent",
+    # Agents
+    "PlannerAgent",
+    "NavigatorAgent",
+    "ExtractorAgent",
+    "VerifierAgent",
+    "MemoryAgent",
+    # Coordinator
+    "AgentCoordinator",
+    "AgentRole",
+    "Message",
+    # Data classes
+    "VerificationResult",
+    "MemoryEntry",
+]

backend/app/agents/base.py

@@ -0,0 +1,127 @@
+"""Base agent abstract class for ScrapeRL agents."""
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+from app.core.action import Action
+from app.core.observation import Observation
+
+
+class BaseAgent(ABC):
+    """
+    Abstract base class for all agents in the ScrapeRL system.
+    
+    Each agent specializes in a specific aspect of the scraping workflow:
+    - Planning and goal decomposition
+    - Navigation and URL prioritization
+    - Data extraction
+    - Verification and validation
+    - Memory operations
+    
+    Agents communicate through message passing and coordinate via
+    the AgentCoordinator.
+    """
+
+    def __init__(self, agent_id: str, config: dict[str, Any] | None = None):
+        """
+        Initialize the agent.
+        
+        Args:
+            agent_id: Unique identifier for this agent instance.
+            config: Optional configuration dictionary for the agent.
+        """
+        self.agent_id = agent_id
+        self.config = config or {}
+        self._message_queue: list[dict[str, Any]] = []
+        self._action_history: list[Action] = []
+
+    @abstractmethod
+    async def act(self, observation: Observation) -> Action:
+        """
+        Select an action based on the current observation.
+        
+        This is the main decision-making method. The agent analyzes
+        the observation and returns the best action to take.
+        
+        Args:
+            observation: The current state observation from the environment.
+            
+        Returns:
+            The action to execute.
+        """
+        pass
+
+    @abstractmethod
+    async def plan(self, observation: Observation) -> list[Action]:
+        """
+        Create a plan of actions based on the current observation.
+        
+        Unlike act() which returns a single action, plan() creates
+        a sequence of actions to achieve a goal.
+        
+        Args:
+            observation: The current state observation from the environment.
+            
+        Returns:
+            A list of planned actions in execution order.
+        """
+        pass
+
+    async def explain(self, action: Action) -> str:
+        """
+        Explain why this action was chosen.
+        
+        Args:
+            action: The action to explain.
+            
+        Returns:
+            A human-readable explanation of the action choice.
+        """
+        return action.reasoning or "No explanation provided"
+
+    def receive_message(self, message: dict[str, Any]) -> None:
+        """
+        Receive a message from another agent.
+        
+        Args:
+            message: The message dictionary containing sender, type, and content.
+        """
+        self._message_queue.append(message)
+
+    def get_pending_messages(self) -> list[dict[str, Any]]:
+        """
+        Get all pending messages and clear the queue.
+        
+        Returns:
+            List of pending messages.
+        """
+        messages = self._message_queue.copy()
+        self._message_queue.clear()
+        return messages
+
+    def record_action(self, action: Action) -> None:
+        """
+        Record an action in the agent's history.
+        
+        Args:
+            action: The action that was executed.
+        """
+        self._action_history.append(action)
+
+    def get_action_history(self) -> list[Action]:
+        """
+        Get the history of actions taken by this agent.
+        
+        Returns:
+            List of past actions.
+        """
+        return self._action_history.copy()
+
+    def reset(self) -> None:
+        """Reset the agent state for a new episode."""
+        self._message_queue.clear()
+        self._action_history.clear()
+
+    def __repr__(self) -> str:
+        """String representation of the agent."""
+        return f"{self.__class__.__name__}(agent_id={self.agent_id!r})"

backend/app/agents/coordinator.py

@@ -0,0 +1,387 @@
+"""Agent coordinator for orchestrating multiple agents with message passing."""
+
+import asyncio
+from datetime import datetime
+from enum import Enum
+from typing import Any
+
+from app.core.action import Action, ActionType
+from app.core.observation import Observation
+
+from .base import BaseAgent
+from .extractor import ExtractorAgent
+from .memory_agent import MemoryAgent
+from .navigator import NavigatorAgent
+from .planner import PlannerAgent
+from .verifier import VerifierAgent
+
+
+class AgentRole(str, Enum):
+    """Roles that agents can fulfill."""
+
+    PLANNER = "planner"
+    NAVIGATOR = "navigator"
+    EXTRACTOR = "extractor"
+    VERIFIER = "verifier"
+    MEMORY = "memory"
+
+
+class Message:
+    """A message between agents."""
+
+    def __init__(
+        self,
+        sender: str,
+        recipient: str,
+        message_type: str,
+        content: dict[str, Any],
+        priority: int = 0,
+    ):
+        """Initialize a message."""
+        self.sender = sender
+        self.recipient = recipient
+        self.message_type = message_type
+        self.content = content
+        self.priority = priority
+        self.timestamp = datetime.utcnow()
+
+    def to_dict(self) -> dict[str, Any]:
+        """Convert to dictionary."""
+        return {
+            "sender": self.sender,
+            "recipient": self.recipient,
+            "message_type": self.message_type,
+            "content": self.content,
+            "priority": self.priority,
+            "timestamp": self.timestamp.isoformat(),
+        }
+
+
+class AgentCoordinator:
+    """
+    Orchestrator for multiple specialized agents.
+    
+    The AgentCoordinator manages:
+    - Agent lifecycle and initialization
+    - Message passing between agents
+    - Action selection and routing
+    - Coordination of multi-agent workflows
+    - Error handling and recovery
+    """
+
+    def __init__(
+        self,
+        config: dict[str, Any] | None = None,
+    ):
+        """
+        Initialize the AgentCoordinator.
+        
+        Args:
+            config: Optional configuration with keys:
+                - enable_parallel: Allow parallel agent execution (default: False)
+                - max_messages_per_step: Max messages per step (default: 10)
+                - default_timeout: Default timeout for agent actions (default: 30)
+        """
+        self.config = config or {}
+        self.enable_parallel = self.config.get("enable_parallel", False)
+        self.max_messages_per_step = self.config.get("max_messages_per_step", 10)
+        self.default_timeout = self.config.get("default_timeout", 30)
+
+        # Initialize agents
+        self._agents: dict[str, BaseAgent] = {}
+        self._message_queue: list[Message] = []
+        self._action_history: list[tuple[str, Action]] = []
+        self._current_lead: str | None = None
+
+        # Initialize default agents
+        self._initialize_default_agents()
+
+    def _initialize_default_agents(self) -> None:
+        """Initialize the default set of agents."""
+        self._agents = {
+            AgentRole.PLANNER: PlannerAgent(
+                agent_id="planner",
+                config=self.config.get("planner_config"),
+            ),
+            AgentRole.NAVIGATOR: NavigatorAgent(
+                agent_id="navigator",
+                config=self.config.get("navigator_config"),
+            ),
+            AgentRole.EXTRACTOR: ExtractorAgent(
+                agent_id="extractor",
+                config=self.config.get("extractor_config"),
+            ),
+            AgentRole.VERIFIER: VerifierAgent(
+                agent_id="verifier",
+                config=self.config.get("verifier_config"),
+            ),
+            AgentRole.MEMORY: MemoryAgent(
+                agent_id="memory",
+                config=self.config.get("memory_config"),
+            ),
+        }
+
+    def register_agent(self, role: str, agent: BaseAgent) -> None:
+        """
+        Register an agent for a specific role.
+        
+        Args:
+            role: The role this agent fulfills.
+            agent: The agent instance.
+        """
+        self._agents[role] = agent
+
+    def get_agent(self, role: str) -> BaseAgent | None:
+        """
+        Get an agent by role.
+        
+        Args:
+            role: The role to look up.
+            
+        Returns:
+            The agent if found, None otherwise.
+        """
+        return self._agents.get(role)
+
+    async def step(self, observation: Observation) -> Action:
+        """
+        Perform one coordination step.
+        
+        Determines which agent should act, processes messages,
+        and returns the selected action.
+        
+        Args:
+            observation: The current state observation.
+            
+        Returns:
+            The action to execute.
+        """
+        try:
+            # Process pending messages
+            await self._process_messages()
+
+            # Determine lead agent based on state
+            lead_role = self._determine_lead_agent(observation)
+            self._current_lead = lead_role
+
+            # Get action from lead agent
+            lead_agent = self._agents.get(lead_role)
+            if not lead_agent:
+                return self._create_error_action(f"No agent for role: {lead_role}")
+
+            # Get action from the lead agent
+            action = await lead_agent.act(observation)
+            action.agent_id = lead_agent.agent_id
+
+            # Record action
+            self._action_history.append((lead_role, action))
+            lead_agent.record_action(action)
+
+            # Handle inter-agent communication actions
+            if action.action_type == ActionType.SEND_MESSAGE:
+                self._handle_send_message(action)
+
+            return action
+
+        except Exception as e:
+            return self._create_error_action(f"Coordination error: {e}")
+
+    async def plan(self, observation: Observation) -> list[Action]:
+        """
+        Create a coordinated plan using multiple agents.
+        
+        The planner agent creates the high-level plan, which is then
+        refined by other agents.
+        
+        Args:
+            observation: The current state observation.
+            
+        Returns:
+            A coordinated list of actions.
+        """
+        try:
+            # Get plan from planner
+            planner = self._agents.get(AgentRole.PLANNER)
+            if not planner:
+                return []
+
+            plan = await planner.plan(observation)
+
+            # Refine with navigator for navigation steps
+            navigator = self._agents.get(AgentRole.NAVIGATOR)
+            if navigator:
+                nav_plan = await navigator.plan(observation)
+                # Insert navigation at the beginning if needed
+                if nav_plan and not observation.current_url:
+                    plan = nav_plan + plan
+
+            return plan
+
+        except Exception as e:
+            return [self._create_error_action(f"Planning error: {e}")]
+
+    def send_message(
+        self,
+        sender: str,
+        recipient: str,
+        message_type: str,
+        content: dict[str, Any],
+        priority: int = 0,
+    ) -> None:
+        """
+        Send a message between agents.
+        
+        Args:
+            sender: ID of the sending agent.
+            recipient: ID of the receiving agent.
+            message_type: Type of the message.
+            content: Message content.
+            priority: Message priority (higher = more urgent).
+        """
+        message = Message(
+            sender=sender,
+            recipient=recipient,
+            message_type=message_type,
+            content=content,
+            priority=priority,
+        )
+        self._message_queue.append(message)
+
+    async def _process_messages(self) -> None:
+        """Process queued messages and deliver to agents."""
+        # Sort by priority (highest first)
+        self._message_queue.sort(key=lambda m: -m.priority)
+
+        # Process up to max messages
+        messages_processed = 0
+        while self._message_queue and messages_processed < self.max_messages_per_step:
+            message = self._message_queue.pop(0)
+
+            # Find recipient agent
+            recipient = None
+            for role, agent in self._agents.items():
+                if agent.agent_id == message.recipient or role == message.recipient:
+                    recipient = agent
+                    break
+
+            if recipient:
+                recipient.receive_message(message.to_dict())
+                messages_processed += 1
+
+    def _determine_lead_agent(self, observation: Observation) -> str:
+        """
+        Determine which agent should lead based on state.
+        
+        Args:
+            observation: Current observation.
+            
+        Returns:
+            The role of the agent that should lead.
+        """
+        # If no URL, navigator should lead
+        if not observation.current_url:
+            return AgentRole.NAVIGATOR
+
+        # If there are unverified fields, verifier should lead
+        unverified = [f for f in observation.extracted_so_far if not f.verified]
+        if unverified and observation.extraction_progress > 0.5:
+            return AgentRole.VERIFIER
+
+        # If there are remaining fields to extract, extractor should lead
+        if observation.fields_remaining:
+            return AgentRole.EXTRACTOR
+
+        # If we have errors, planner should re-plan
+        if observation.consecutive_errors > 0:
+            return AgentRole.PLANNER
+
+        # Default to planner
+        return AgentRole.PLANNER
+
+    def _handle_send_message(self, action: Action) -> None:
+        """Handle a send_message action from an agent."""
+        params = action.parameters
+        self.send_message(
+            sender=action.agent_id or "unknown",
+            recipient=params.get("target_agent", ""),
+            message_type=params.get("message_type", "generic"),
+            content=params.get("content", {}),
+        )
+
+    def _create_error_action(self, error: str) -> Action:
+        """Create a fail action for errors."""
+        return Action(
+            action_type=ActionType.FAIL,
+            parameters={"success": False, "message": error},
+            reasoning=error,
+            confidence=1.0,
+            agent_id="coordinator",
+        )
+
+    async def run_parallel_agents(
+        self,
+        observation: Observation,
+        roles: list[str],
+    ) -> dict[str, Action]:
+        """
+        Run multiple agents in parallel.
+        
+        Args:
+            observation: Current observation.
+            roles: List of agent roles to run.
+            
+        Returns:
+            Dictionary mapping role to action.
+        """
+        if not self.enable_parallel:
+            # Fallback to sequential
+            results = {}
+            for role in roles:
+                agent = self._agents.get(role)
+                if agent:
+                    results[role] = await agent.act(observation)
+            return results
+
+        # Run agents in parallel
+        async def run_agent(role: str) -> tuple[str, Action]:
+            agent = self._agents.get(role)
+            if agent:
+                action = await agent.act(observation)
+                return (role, action)
+            return (role, self._create_error_action(f"No agent for role: {role}"))
+
+        tasks = [run_agent(role) for role in roles]
+        results = await asyncio.gather(*tasks)
+
+        return dict(results)
+
+    def get_action_history(self) -> list[tuple[str, Action]]:
+        """Get the history of actions with their agent roles."""
+        return self._action_history.copy()
+
+    def get_current_lead(self) -> str | None:
+        """Get the current lead agent role."""
+        return self._current_lead
+
+    def get_message_queue_length(self) -> int:
+        """Get the number of pending messages."""
+        return len(self._message_queue)
+
+    def reset(self) -> None:
+        """Reset all agents and coordinator state."""
+        for agent in self._agents.values():
+            agent.reset()
+
+        self._message_queue.clear()
+        self._action_history.clear()
+        self._current_lead = None
+
+    def get_stats(self) -> dict[str, Any]:
+        """Get coordinator statistics."""
+        return {
+            "agents": list(self._agents.keys()),
+            "current_lead": self._current_lead,
+            "pending_messages": len(self._message_queue),
+            "action_count": len(self._action_history),
+            "enable_parallel": self.enable_parallel,
+        }

backend/app/agents/extractor.py

@@ -0,0 +1,489 @@
+"""Extractor agent for data extraction with selectors."""
+
+import re
+from typing import Any
+
+from app.core.action import Action, ActionType
+from app.core.observation import Observation, PageElement
+
+from .base import BaseAgent
+
+
+class ExtractorAgent(BaseAgent):
+    """
+    Agent responsible for extracting structured data from pages.
+    
+    The ExtractorAgent handles:
+    - Identifying data elements using CSS/XPath selectors
+    - Extracting text, attributes, and structured content
+    - Handling tables and lists
+    - Post-processing extracted values
+    - Confidence scoring for extractions
+    """
+
+    def __init__(
+        self,
+        agent_id: str = "extractor",
+        config: dict[str, Any] | None = None,
+    ):
+        """
+        Initialize the ExtractorAgent.
+        
+        Args:
+            agent_id: Unique identifier for this agent.
+            config: Optional configuration with keys:
+                - min_confidence: Minimum confidence to accept extraction
+                - extraction_timeout: Timeout for extraction operations
+                - enable_fuzzy_matching: Enable fuzzy text matching
+        """
+        super().__init__(agent_id, config)
+        self.min_confidence = self.config.get("min_confidence", 0.5)
+        self.extraction_timeout = self.config.get("extraction_timeout", 5000)
+        self.enable_fuzzy_matching = self.config.get("enable_fuzzy_matching", True)
+        self._extraction_cache: dict[str, Any] = {}
+        self._selector_patterns: dict[str, list[str]] = self._init_selector_patterns()
+
+    def _init_selector_patterns(self) -> dict[str, list[str]]:
+        """Initialize common selector patterns for different field types."""
+        return {
+            "price": [
+                "[class*='price']",
+                "[id*='price']",
+                "[itemprop='price']",
+                ".product-price",
+                ".item-price",
+                "span[data-price]",
+            ],
+            "title": [
+                "h1",
+                "[class*='title']",
+                "[itemprop='name']",
+                ".product-title",
+                ".item-title",
+            ],
+            "description": [
+                "[class*='description']",
+                "[itemprop='description']",
+                ".product-description",
+                "article p",
+                ".content p",
+            ],
+            "image": [
+                "[class*='product-image'] img",
+                "[itemprop='image']",
+                ".main-image img",
+                "figure img",
+            ],
+            "date": [
+                "time",
+                "[datetime]",
+                "[class*='date']",
+                "[itemprop='datePublished']",
+            ],
+            "author": [
+                "[class*='author']",
+                "[itemprop='author']",
+                "[rel='author']",
+                ".byline",
+            ],
+        }
+
+    async def act(self, observation: Observation) -> Action:
+        """
+        Select the best extraction action based on observation.
+        
+        Analyzes the page and decides what data to extract next.
+        
+        Args:
+            observation: The current state observation.
+            
+        Returns:
+            The extraction action to execute.
+        """
+        try:
+            # Get remaining fields to extract
+            remaining_fields = observation.fields_remaining
+
+            if not remaining_fields:
+                return Action(
+                    action_type=ActionType.DONE,
+                    parameters={"success": True, "message": "All fields extracted"},
+                    reasoning="No more fields to extract",
+                    confidence=1.0,
+                    agent_id=self.agent_id,
+                )
+
+            # Pick the next field to extract
+            field_name = remaining_fields[0]
+
+            # Find best selector for the field
+            selector, confidence = await self._find_selector_for_field(
+                field_name,
+                observation,
+            )
+
+            if selector and confidence >= self.min_confidence:
+                return self._create_extraction_action(
+                    field_name,
+                    selector,
+                    confidence,
+                )
+
+            # Try alternative extraction methods
+            alt_action = await self._try_alternative_extraction(
+                field_name,
+                observation,
+            )
+            if alt_action:
+                return alt_action
+
+            # Cannot extract this field
+            return Action(
+                action_type=ActionType.EXTRACT_FIELD,
+                parameters={
+                    "field_name": field_name,
+                    "selector": None,
+                    "extraction_method": "llm",
+                },
+                reasoning=f"No selector found, using LLM extraction for {field_name}",
+                confidence=0.4,
+                agent_id=self.agent_id,
+            )
+
+        except Exception as e:
+            return Action(
+                action_type=ActionType.FAIL,
+                parameters={"success": False, "message": str(e)},
+                reasoning=f"Extraction error: {e}",
+                confidence=1.0,
+                agent_id=self.agent_id,
+            )
+
+    async def plan(self, observation: Observation) -> list[Action]:
+        """
+        Create an extraction plan for all remaining fields.
+        
+        Analyzes the page structure and plans the optimal
+        extraction sequence.
+        
+        Args:
+            observation: The current state observation.
+            
+        Returns:
+            A list of planned extraction actions.
+        """
+        try:
+            actions: list[Action] = []
+            remaining_fields = observation.fields_remaining
+
+            for field_name in remaining_fields:
+                selector, confidence = await self._find_selector_for_field(
+                    field_name,
+                    observation,
+                )
+
+                if selector:
+                    actions.append(
+                        self._create_extraction_action(
+                            field_name,
+                            selector,
+                            confidence,
+                        )
+                    )
+                else:
+                    # Plan LLM-based extraction as fallback
+                    actions.append(
+                        Action(
+                            action_type=ActionType.EXTRACT_FIELD,
+                            parameters={
+                                "field_name": field_name,
+                                "extraction_method": "llm",
+                            },
+                            reasoning=f"Planning LLM extraction for {field_name}",
+                            confidence=0.5,
+                            agent_id=self.agent_id,
+                        )
+                    )
+
+            return actions
+
+        except Exception as e:
+            return [
+                Action(
+                    action_type=ActionType.FAIL,
+                    parameters={"message": f"Extraction planning failed: {e}"},
+                    reasoning=str(e),
+                    confidence=1.0,
+                    agent_id=self.agent_id,
+                )
+            ]
+
+    async def _find_selector_for_field(
+        self,
+        field_name: str,
+        observation: Observation,
+    ) -> tuple[str | None, float]:
+        """
+        Find the best selector for a field.
+        
+        Args:
+            field_name: Name of the field to extract.
+            observation: Current observation.
+            
+        Returns:
+            Tuple of (selector, confidence).
+        """
+        best_selector: str | None = None
+        best_confidence = 0.0
+
+        # Check predefined patterns first
+        patterns = self._get_patterns_for_field(field_name)
+        for pattern in patterns:
+            element = self._find_element_by_selector(
+                pattern,
+                observation.page_elements,
+            )
+            if element:
+                confidence = self._calculate_confidence(element, field_name)
+                if confidence > best_confidence:
+                    best_selector = element.selector
+                    best_confidence = confidence
+
+        # Search by text content if fuzzy matching enabled
+        if self.enable_fuzzy_matching and best_confidence < 0.7:
+            element, confidence = self._find_element_by_text(
+                field_name,
+                observation.page_elements,
+            )
+            if element and confidence > best_confidence:
+                best_selector = element.selector
+                best_confidence = confidence
+
+        return best_selector, best_confidence
+
+    def _get_patterns_for_field(self, field_name: str) -> list[str]:
+        """Get selector patterns for a field type."""
+        field_lower = field_name.lower()
+
+        # Direct match
+        if field_lower in self._selector_patterns:
+            return self._selector_patterns[field_lower]
+
+        # Partial match
+        for key, patterns in self._selector_patterns.items():
+            if key in field_lower or field_lower in key:
+                return patterns
+
+        # Generate generic patterns
+        return [
+            f"[class*='{field_lower}']",
+            f"[id*='{field_lower}']",
+            f"[data-{field_lower}]",
+            f".{field_lower}",
+            f"#{field_lower}",
+        ]
+
+    def _find_element_by_selector(
+        self,
+        selector: str,
+        elements: list[PageElement],
+    ) -> PageElement | None:
+        """Find an element matching a selector pattern."""
+        selector_lower = selector.lower()
+
+        for element in elements:
+            element_selector = element.selector.lower()
+            if selector_lower in element_selector:
+                return element
+
+            # Check class and id attributes
+            classes = element.attributes.get("class", "").lower()
+            element_id = element.attributes.get("id", "").lower()
+
+            if selector_lower.strip(".[#]") in classes:
+                return element
+            if selector_lower.strip(".[#]") in element_id:
+                return element
+
+        return None
+
+    def _find_element_by_text(
+        self,
+        field_name: str,
+        elements: list[PageElement],
+    ) -> tuple[PageElement | None, float]:
+        """Find an element by text content matching."""
+        field_lower = field_name.lower().replace("_", " ")
+        best_element: PageElement | None = None
+        best_score = 0.0
+
+        for element in elements:
+            if not element.text:
+                continue
+
+            text_lower = element.text.lower()
+
+            # Check for label-like patterns
+            if f"{field_lower}:" in text_lower or f"{field_lower} :" in text_lower:
+                score = 0.9
+            elif field_lower in text_lower:
+                # Calculate similarity score
+                score = len(field_lower) / max(len(text_lower), 1) * 0.8
+            else:
+                continue
+
+            if score > best_score:
+                best_element = element
+                best_score = score
+
+        return best_element, best_score
+
+    def _calculate_confidence(self, element: PageElement, field_name: str) -> float:
+        """Calculate extraction confidence for an element."""
+        confidence = 0.5
+
+        # Boost for visible elements
+        if element.is_visible:
+            confidence += 0.1
+
+        # Boost for semantic attributes
+        if element.attributes.get("itemprop"):
+            confidence += 0.2
+        if element.attributes.get("data-field"):
+            confidence += 0.15
+
+        # Boost if text contains field name
+        if element.text and field_name.lower() in element.text.lower():
+            confidence += 0.1
+
+        # Penalty for very long text (likely not a single field)
+        if element.text and len(element.text) > 500:
+            confidence -= 0.2
+
+        return min(1.0, max(0.0, confidence))
+
+    async def _try_alternative_extraction(
+        self,
+        field_name: str,
+        observation: Observation,
+    ) -> Action | None:
+        """Try alternative extraction methods."""
+        # Check for table data
+        for element in observation.page_elements:
+            if element.tag in ("table", "tbody"):
+                return Action(
+                    action_type=ActionType.EXTRACT_TABLE,
+                    parameters={
+                        "table_selector": element.selector,
+                        "target_field": field_name,
+                    },
+                    reasoning=f"Extracting {field_name} from table",
+                    confidence=0.6,
+                    agent_id=self.agent_id,
+                )
+
+        # Check for list data
+        for element in observation.page_elements:
+            if element.tag in ("ul", "ol", "dl"):
+                return Action(
+                    action_type=ActionType.EXTRACT_LIST,
+                    parameters={
+                        "container_selector": element.selector,
+                        "item_selector": "li",
+                        "field_selectors": {field_name: "text"},
+                    },
+                    reasoning=f"Extracting {field_name} from list",
+                    confidence=0.55,
+                    agent_id=self.agent_id,
+                )
+
+        return None
+
+    def _create_extraction_action(
+        self,
+        field_name: str,
+        selector: str,
+        confidence: float,
+    ) -> Action:
+        """Create an extraction action."""
+        return Action(
+            action_type=ActionType.EXTRACT_FIELD,
+            parameters={
+                "field_name": field_name,
+                "selector": selector,
+                "extraction_method": "text",
+            },
+            reasoning=f"Extracting {field_name} using selector: {selector}",
+            confidence=confidence,
+            agent_id=self.agent_id,
+        )
+
+    def extract_with_regex(
+        self,
+        text: str,
+        pattern: str,
+        group: int = 0,
+    ) -> str | None:
+        """
+        Extract text using a regex pattern.
+        
+        Args:
+            text: The text to search in.
+            pattern: Regex pattern.
+            group: Capture group to return.
+            
+        Returns:
+            Extracted text or None.
+        """
+        try:
+            match = re.search(pattern, text)
+            if match:
+                return match.group(group)
+            return None
+        except re.error:
+            return None
+
+    def post_process_value(
+        self,
+        value: Any,
+        field_name: str,
+    ) -> Any:
+        """
+        Post-process an extracted value based on field type.
+        
+        Args:
+            value: The raw extracted value.
+            field_name: Name of the field (used to infer type).
+            
+        Returns:
+            Processed value.
+        """
+        if value is None:
+            return None
+
+        value_str = str(value).strip()
+        field_lower = field_name.lower()
+
+        # Price processing
+        if "price" in field_lower:
+            # Remove currency symbols but keep numbers and decimal
+            price_match = re.search(r"[\d,]+\.?\d*", value_str.replace(",", ""))
+            if price_match:
+                return float(price_match.group().replace(",", ""))
+
+        # Date processing
+        if "date" in field_lower:
+            return value_str  # Return as-is, let caller parse
+
+        # Number processing
+        if any(x in field_lower for x in ["count", "quantity", "number"]):
+            num_match = re.search(r"\d+", value_str)
+            if num_match:
+                return int(num_match.group())
+
+        return value_str
+
+    def reset(self) -> None:
+        """Reset the extractor state."""
+        super().reset()
+        self._extraction_cache.clear()

backend/app/agents/memory_agent.py

@@ -0,0 +1,474 @@
+"""Memory agent for memory operations and knowledge management."""
+
+from datetime import datetime
+from typing import Any
+
+from app.core.action import Action, ActionType
+from app.core.observation import Observation
+
+from .base import BaseAgent
+
+
+class MemoryEntry:
+    """A single memory entry."""
+
+    def __init__(
+        self,
+        key: str,
+        value: Any,
+        memory_type: str = "working",
+        ttl_seconds: int | None = None,
+        metadata: dict[str, Any] | None = None,
+    ):
+        """Initialize memory entry."""
+        self.key = key
+        self.value = value
+        self.memory_type = memory_type
+        self.ttl_seconds = ttl_seconds
+        self.metadata = metadata or {}
+        self.created_at = datetime.utcnow()
+        self.accessed_at = datetime.utcnow()
+        self.access_count = 0
+
+    def is_expired(self) -> bool:
+        """Check if the memory entry has expired."""
+        if self.ttl_seconds is None:
+            return False
+        elapsed = (datetime.utcnow() - self.created_at).total_seconds()
+        return elapsed > self.ttl_seconds
+
+    def access(self) -> Any:
+        """Access the memory and update metadata."""
+        self.accessed_at = datetime.utcnow()
+        self.access_count += 1
+        return self.value
+
+    def to_dict(self) -> dict[str, Any]:
+        """Convert to dictionary."""
+        return {
+            "key": self.key,
+            "value": self.value,
+            "memory_type": self.memory_type,
+            "ttl_seconds": self.ttl_seconds,
+            "metadata": self.metadata,
+            "created_at": self.created_at.isoformat(),
+            "accessed_at": self.accessed_at.isoformat(),
+            "access_count": self.access_count,
+        }
+
+
+class MemoryAgent(BaseAgent):
+    """
+    Agent responsible for memory operations and knowledge management.
+    
+    The MemoryAgent handles:
+    - Storing and retrieving memories across different layers
+    - Managing short-term, working, and long-term memory
+    - Memory consolidation and cleanup
+    - Relevance-based memory retrieval
+    - Sharing knowledge between episodes
+    """
+
+    def __init__(
+        self,
+        agent_id: str = "memory",
+        config: dict[str, Any] | None = None,
+    ):
+        """
+        Initialize the MemoryAgent.
+        
+        Args:
+            agent_id: Unique identifier for this agent.
+            config: Optional configuration with keys:
+                - max_short_term: Max short-term memory entries (default: 100)
+                - max_working: Max working memory entries (default: 50)
+                - consolidation_threshold: Accesses before long-term (default: 3)
+                - enable_auto_cleanup: Auto cleanup expired entries (default: True)
+        """
+        super().__init__(agent_id, config)
+        self.max_short_term = self.config.get("max_short_term", 100)
+        self.max_working = self.config.get("max_working", 50)
+        self.consolidation_threshold = self.config.get("consolidation_threshold", 3)
+        self.enable_auto_cleanup = self.config.get("enable_auto_cleanup", True)
+
+        # Memory stores
+        self._short_term: dict[str, MemoryEntry] = {}
+        self._working: dict[str, MemoryEntry] = {}
+        self._pending_operations: list[dict[str, Any]] = []
+
+    async def act(self, observation: Observation) -> Action:
+        """
+        Select the best memory action based on observation.
+        
+        Analyzes the current state and determines if any memory
+        operations are needed.
+        
+        Args:
+            observation: The current state observation.
+            
+        Returns:
+            The memory action to execute.
+        """
+        try:
+            # Process any pending messages requesting memory operations
+            messages = self.get_pending_messages()
+            for msg in messages:
+                if msg.get("message_type") == "memory_request":
+                    return self._process_memory_request(msg)
+
+            # Auto cleanup if enabled
+            if self.enable_auto_cleanup:
+                self._cleanup_expired()
+
+            # Check if we should store new information
+            store_action = self._check_for_storage(observation)
+            if store_action:
+                return store_action
+
+            # Check if any memories need consolidation
+            consolidation_action = self._check_for_consolidation()
+            if consolidation_action:
+                return consolidation_action
+
+            # No memory operations needed
+            return Action(
+                action_type=ActionType.WAIT,
+                parameters={"duration_ms": 100},
+                reasoning="No memory operations required",
+                confidence=1.0,
+                agent_id=self.agent_id,
+            )
+
+        except Exception as e:
+            return Action(
+                action_type=ActionType.FAIL,
+                parameters={"success": False, "message": str(e)},
+                reasoning=f"Memory operation error: {e}",
+                confidence=1.0,
+                agent_id=self.agent_id,
+            )
+
+    async def plan(self, observation: Observation) -> list[Action]:
+        """
+        Create a plan of memory operations.
+        
+        Plans memory operations needed based on the current state
+        and extracted data.
+        
+        Args:
+            observation: The current state observation.
+            
+        Returns:
+            A list of planned memory actions.
+        """
+        try:
+            actions: list[Action] = []
+
+            # Plan to store extracted fields
+            for field in observation.extracted_so_far:
+                if field.verified and field.confidence > 0.8:
+                    actions.append(
+                        Action(
+                            action_type=ActionType.STORE_MEMORY,
+                            parameters={
+                                "key": f"extracted:{field.field_name}",
+                                "value": field.value,
+                                "memory_type": "working",
+                                "metadata": {
+                                    "source": observation.current_url,
+                                    "confidence": field.confidence,
+                                },
+                            },
+                            reasoning=f"Storing verified field: {field.field_name}",
+                            confidence=0.9,
+                            agent_id=self.agent_id,
+                        )
+                    )
+
+            # Plan to recall relevant memories for current task
+            if observation.task_context:
+                for target in observation.task_context.target_fields:
+                    actions.append(
+                        Action(
+                            action_type=ActionType.RECALL_MEMORY,
+                            parameters={
+                                "key": f"pattern:{target}",
+                                "memory_type": "long_term",
+                            },
+                            reasoning=f"Recalling patterns for field: {target}",
+                            confidence=0.7,
+                            agent_id=self.agent_id,
+                        )
+                    )
+
+            return actions
+
+        except Exception as e:
+            return [
+                Action(
+                    action_type=ActionType.FAIL,
+                    parameters={"message": f"Memory planning failed: {e}"},
+                    reasoning=str(e),
+                    confidence=1.0,
+                    agent_id=self.agent_id,
+                )
+            ]
+
+    def store(
+        self,
+        key: str,
+        value: Any,
+        memory_type: str = "working",
+        ttl_seconds: int | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> bool:
+        """
+        Store a value in memory.
+        
+        Args:
+            key: The key to store under.
+            value: The value to store.
+            memory_type: Type of memory (short_term, working).
+            ttl_seconds: Optional time-to-live.
+            metadata: Optional metadata.
+            
+        Returns:
+            True if stored successfully.
+        """
+        entry = MemoryEntry(
+            key=key,
+            value=value,
+            memory_type=memory_type,
+            ttl_seconds=ttl_seconds,
+            metadata=metadata,
+        )
+
+        if memory_type == "short_term":
+            self._enforce_limit(self._short_term, self.max_short_term)
+            self._short_term[key] = entry
+        elif memory_type == "working":
+            self._enforce_limit(self._working, self.max_working)
+            self._working[key] = entry
+        else:
+            return False
+
+        return True
+
+    def recall(
+        self,
+        key: str,
+        memory_type: str | None = None,
+    ) -> Any | None:
+        """
+        Recall a value from memory.
+        
+        Args:
+            key: The key to recall.
+            memory_type: Optional specific memory type to search.
+            
+        Returns:
+            The value if found, None otherwise.
+        """
+        # Search in order of specificity
+        stores = []
+        if memory_type == "working" or memory_type is None:
+            stores.append(self._working)
+        if memory_type == "short_term" or memory_type is None:
+            stores.append(self._short_term)
+
+        for store in stores:
+            if key in store:
+                entry = store[key]
+                if not entry.is_expired():
+                    return entry.access()
+                else:
+                    # Clean up expired entry
+                    del store[key]
+
+        return None
+
+    def search(
+        self,
+        query: str,
+        memory_type: str | None = None,
+        limit: int = 10,
+    ) -> list[dict[str, Any]]:
+        """
+        Search memories by key prefix or content.
+        
+        Args:
+            query: Search query (matches key prefix).
+            memory_type: Optional specific memory type.
+            limit: Maximum results to return.
+            
+        Returns:
+            List of matching memories.
+        """
+        results: list[dict[str, Any]] = []
+        query_lower = query.lower()
+
+        stores = []
+        if memory_type in ("working", None):
+            stores.append(("working", self._working))
+        if memory_type in ("short_term", None):
+            stores.append(("short_term", self._short_term))
+
+        for store_name, store in stores:
+            for key, entry in store.items():
+                if entry.is_expired():
+                    continue
+
+                # Match by key prefix or value content
+                if (
+                    key.lower().startswith(query_lower)
+                    or query_lower in str(entry.value).lower()
+                ):
+                    results.append({
+                        **entry.to_dict(),
+                        "store": store_name,
+                    })
+
+                if len(results) >= limit:
+                    break
+
+        return results[:limit]
+
+    def _process_memory_request(self, message: dict[str, Any]) -> Action:
+        """Process a memory request from another agent."""
+        content = message.get("content", {})
+        operation = content.get("operation", "recall")
+        key = content.get("key", "")
+
+        if operation == "store":
+            success = self.store(
+                key=key,
+                value=content.get("value"),
+                memory_type=content.get("memory_type", "working"),
+                ttl_seconds=content.get("ttl_seconds"),
+                metadata=content.get("metadata"),
+            )
+            return Action(
+                action_type=ActionType.STORE_MEMORY,
+                parameters={"key": key, "success": success},
+                reasoning=f"Processed store request for key: {key}",
+                confidence=1.0 if success else 0.5,
+                agent_id=self.agent_id,
+            )
+
+        elif operation == "recall":
+            value = self.recall(key, content.get("memory_type"))
+            return Action(
+                action_type=ActionType.RECALL_MEMORY,
+                parameters={"key": key, "value": value, "found": value is not None},
+                reasoning=f"Processed recall request for key: {key}",
+                confidence=1.0 if value else 0.3,
+                agent_id=self.agent_id,
+            )
+
+        else:
+            return Action(
+                action_type=ActionType.FAIL,
+                parameters={"message": f"Unknown memory operation: {operation}"},
+                reasoning=f"Invalid memory request",
+                confidence=1.0,
+                agent_id=self.agent_id,
+            )
+
+    def _check_for_storage(self, observation: Observation) -> Action | None:
+        """Check if any new information should be stored."""
+        # Store newly extracted, verified fields
+        for field in observation.extracted_so_far:
+            key = f"field:{field.field_name}"
+            if key not in self._working and field.verified:
+                return Action(
+                    action_type=ActionType.STORE_MEMORY,
+                    parameters={
+                        "key": key,
+                        "value": {
+                            "field_name": field.field_name,
+                            "value": field.value,
+                            "confidence": field.confidence,
+                            "source": observation.current_url,
+                        },
+                        "memory_type": "working",
+                    },
+                    reasoning=f"Storing verified extraction: {field.field_name}",
+                    confidence=0.85,
+                    agent_id=self.agent_id,
+                )
+
+        return None
+
+    def _check_for_consolidation(self) -> Action | None:
+        """Check if any memories should be consolidated to long-term."""
+        for key, entry in self._working.items():
+            if entry.access_count >= self.consolidation_threshold:
+                return Action(
+                    action_type=ActionType.STORE_MEMORY,
+                    parameters={
+                        "key": key,
+                        "value": entry.value,
+                        "memory_type": "long_term",
+                        "metadata": {
+                            "access_count": entry.access_count,
+                            "consolidated_from": "working",
+                        },
+                    },
+                    reasoning=f"Consolidating frequently accessed memory: {key}",
+                    confidence=0.8,
+                    agent_id=self.agent_id,
+                )
+
+        return None
+
+    def _cleanup_expired(self) -> int:
+        """Clean up expired memory entries."""
+        cleaned = 0
+
+        for store in [self._short_term, self._working]:
+            expired_keys = [
+                k for k, v in store.items()
+                if v.is_expired()
+            ]
+            for key in expired_keys:
+                del store[key]
+                cleaned += 1
+
+        return cleaned
+
+    def _enforce_limit(
+        self,
+        store: dict[str, MemoryEntry],
+        limit: int,
+    ) -> None:
+        """Enforce memory limit by removing least accessed entries."""
+        if len(store) < limit:
+            return
+
+        # Sort by access count and last access time
+        sorted_entries = sorted(
+            store.items(),
+            key=lambda x: (x[1].access_count, x[1].accessed_at),
+        )
+
+        # Remove oldest/least accessed entries
+        to_remove = len(store) - limit + 1
+        for key, _ in sorted_entries[:to_remove]:
+            del store[key]
+
+    def get_memory_stats(self) -> dict[str, Any]:
+        """Get statistics about memory usage."""
+        return {
+            "short_term_count": len(self._short_term),
+            "short_term_limit": self.max_short_term,
+            "working_count": len(self._working),
+            "working_limit": self.max_working,
+            "total_entries": len(self._short_term) + len(self._working),
+        }
+
+    def reset(self) -> None:
+        """Reset the memory agent state."""
+        super().reset()
+        self._short_term.clear()
+        self._working.clear()
+        self._pending_operations.clear()

backend/app/agents/navigator.py

@@ -0,0 +1,368 @@
+"""Navigator agent for URL prioritization and page navigation."""
+
+from typing import Any
+from urllib.parse import urljoin, urlparse
+
+from app.core.action import Action, ActionType
+from app.core.observation import Observation, PageElement
+
+from .base import BaseAgent
+
+
+class NavigatorAgent(BaseAgent):
+    """
+    Agent responsible for intelligent page navigation.
+    
+    The NavigatorAgent handles:
+    - URL prioritization based on relevance to task
+    - Link discovery and scoring
+    - Navigation decision making
+    - Handling pagination and multi-page content
+    - Avoiding irrelevant or harmful URLs
+    """
+
+    def __init__(
+        self,
+        agent_id: str = "navigator",
+        config: dict[str, Any] | None = None,
+    ):
+        """
+        Initialize the NavigatorAgent.
+        
+        Args:
+            agent_id: Unique identifier for this agent.
+            config: Optional configuration with keys:
+                - max_depth: Maximum navigation depth (default: 5)
+                - allowed_domains: List of allowed domains to visit
+                - blocked_patterns: URL patterns to avoid
+                - prioritize_https: Prefer HTTPS URLs (default: True)
+        """
+        super().__init__(agent_id, config)
+        self.max_depth = self.config.get("max_depth", 5)
+        self.allowed_domains = self.config.get("allowed_domains", [])
+        self.blocked_patterns = self.config.get("blocked_patterns", [
+            "logout", "signout", "delete", "remove", "unsubscribe",
+        ])
+        self.prioritize_https = self.config.get("prioritize_https", True)
+        self._visited_urls: set[str] = set()
+        self._url_scores: dict[str, float] = {}
+
+    async def act(self, observation: Observation) -> Action:
+        """
+        Select the best navigation action based on observation.
+        
+        Analyzes available links and decides whether to:
+        - Navigate to a new page
+        - Go back to a previous page
+        - Click an element to reveal more content
+        
+        Args:
+            observation: The current state observation.
+            
+        Returns:
+            The navigation action to execute.
+        """
+        try:
+            # Track current URL
+            if observation.current_url:
+                self._visited_urls.add(observation.current_url)
+
+            # Check if we've reached max depth
+            nav_depth = len(observation.navigation_history)
+            if nav_depth >= self.max_depth:
+                return self._create_go_back_action(
+                    "Reached maximum navigation depth"
+                )
+
+            # Find best link to follow
+            best_link = await self._find_best_link(observation)
+
+            if best_link:
+                return self._create_navigate_action(best_link, observation)
+
+            # Check for pagination
+            pagination_action = self._find_pagination(observation)
+            if pagination_action:
+                return pagination_action
+
+            # No good links, consider going back
+            if observation.can_go_back and nav_depth > 1:
+                return self._create_go_back_action(
+                    "No relevant links found, going back"
+                )
+
+            # Nothing to navigate to
+            return Action(
+                action_type=ActionType.WAIT,
+                parameters={"duration_ms": 500},
+                reasoning="No navigation targets found",
+                confidence=0.5,
+                agent_id=self.agent_id,
+            )
+
+        except Exception as e:
+            return Action(
+                action_type=ActionType.FAIL,
+                parameters={"success": False, "message": str(e)},
+                reasoning=f"Navigation error: {e}",
+                confidence=1.0,
+                agent_id=self.agent_id,
+            )
+
+    async def plan(self, observation: Observation) -> list[Action]:
+        """
+        Create a navigation plan based on task requirements.
+        
+        Plans a sequence of navigation actions to reach content
+        relevant to the task.
+        
+        Args:
+            observation: The current state observation.
+            
+        Returns:
+            A list of planned navigation actions.
+        """
+        try:
+            actions: list[Action] = []
+            task_context = observation.task_context
+
+            if not task_context:
+                return []
+
+            # Analyze task hints for navigation targets
+            target_urls = self._extract_urls_from_hints(task_context.hints)
+
+            for url in target_urls[:3]:  # Limit to top 3 URLs
+                if url not in self._visited_urls:
+                    actions.append(
+                        Action(
+                            action_type=ActionType.NAVIGATE,
+                            parameters={"url": url, "timeout_ms": 30000},
+                            reasoning=f"Navigating to task-relevant URL: {url}",
+                            confidence=0.85,
+                            agent_id=self.agent_id,
+                        )
+                    )
+
+            # If no URLs from hints, plan a search
+            if not actions:
+                search_query = self._build_search_query(task_context)
+                actions.append(
+                    Action(
+                        action_type=ActionType.SEARCH_ENGINE,
+                        parameters={"query": search_query, "engine": "google"},
+                        reasoning=f"Searching for: {search_query}",
+                        confidence=0.7,
+                        agent_id=self.agent_id,
+                    )
+                )
+
+            return actions
+
+        except Exception as e:
+            return [
+                Action(
+                    action_type=ActionType.FAIL,
+                    parameters={"message": f"Navigation planning failed: {e}"},
+                    reasoning=str(e),
+                    confidence=1.0,
+                    agent_id=self.agent_id,
+                )
+            ]
+
+    async def _find_best_link(self, observation: Observation) -> str | None:
+        """Find the best link to follow based on task relevance."""
+        if not observation.task_context:
+            return None
+
+        target_fields = observation.task_context.target_fields
+        remaining_fields = observation.fields_remaining
+
+        # Score all links on the page
+        link_scores: list[tuple[str, float]] = []
+
+        for element in observation.page_elements:
+            if not element.is_interactive:
+                continue
+
+            href = element.attributes.get("href", "")
+            if not href or href.startswith("#") or href.startswith("javascript:"):
+                continue
+
+            # Resolve relative URLs
+            full_url = self._resolve_url(href, observation.current_url)
+            if not full_url:
+                continue
+
+            # Skip already visited URLs
+            if full_url in self._visited_urls:
+                continue
+
+            # Skip blocked patterns
+            if self._is_blocked_url(full_url):
+                continue
+
+            # Check domain restrictions
+            if not self._is_allowed_domain(full_url):
+                continue
+
+            # Score the link
+            score = self._score_link(element, full_url, remaining_fields)
+            if score > 0:
+                link_scores.append((full_url, score))
+
+        # Return highest scoring link
+        if link_scores:
+            link_scores.sort(key=lambda x: x[1], reverse=True)
+            return link_scores[0][0]
+
+        return None
+
+    def _score_link(
+        self,
+        element: PageElement,
+        url: str,
+        target_fields: list[str],
+    ) -> float:
+        """Score a link based on relevance to task fields."""
+        score = 0.0
+        text = (element.text or "").lower()
+        url_lower = url.lower()
+
+        # Check if link text contains target field names
+        for field in target_fields:
+            field_lower = field.lower()
+            if field_lower in text:
+                score += 0.4
+            if field_lower in url_lower:
+                score += 0.3
+
+        # Prefer HTTPS
+        if self.prioritize_https and url.startswith("https://"):
+            score += 0.1
+
+        # Boost content-like URLs
+        content_indicators = ["detail", "view", "info", "about", "product", "page"]
+        for indicator in content_indicators:
+            if indicator in url_lower:
+                score += 0.2
+                break
+
+        # Penalize non-content URLs
+        noise_indicators = ["login", "cart", "checkout", "share", "print"]
+        for indicator in noise_indicators:
+            if indicator in url_lower:
+                score -= 0.3
+                break
+
+        return max(0.0, score)
+
+    def _resolve_url(self, href: str, base_url: str | None) -> str | None:
+        """Resolve a relative URL to an absolute URL."""
+        if not href:
+            return None
+
+        if href.startswith(("http://", "https://")):
+            return href
+
+        if not base_url:
+            return None
+
+        try:
+            return urljoin(base_url, href)
+        except Exception:
+            return None
+
+    def _is_blocked_url(self, url: str) -> bool:
+        """Check if URL matches any blocked patterns."""
+        url_lower = url.lower()
+        for pattern in self.blocked_patterns:
+            if pattern.lower() in url_lower:
+                return True
+        return False
+
+    def _is_allowed_domain(self, url: str) -> bool:
+        """Check if URL domain is allowed."""
+        if not self.allowed_domains:
+            return True
+
+        try:
+            parsed = urlparse(url)
+            domain = parsed.netloc.lower()
+            for allowed in self.allowed_domains:
+                if domain == allowed.lower() or domain.endswith("." + allowed.lower()):
+                    return True
+            return False
+        except Exception:
+            return False
+
+    def _find_pagination(self, observation: Observation) -> Action | None:
+        """Find and create action for pagination elements."""
+        pagination_selectors = [
+            "[aria-label*='next']",
+            "[aria-label*='Next']",
+            "a.next",
+            "button.next",
+            "[rel='next']",
+        ]
+
+        for element in observation.page_elements:
+            text = (element.text or "").lower()
+            if element.is_interactive and ("next" in text or "more" in text):
+                return Action(
+                    action_type=ActionType.CLICK,
+                    parameters={"selector": element.selector},
+                    reasoning="Clicking pagination to load more content",
+                    confidence=0.7,
+                    agent_id=self.agent_id,
+                )
+
+        return None
+
+    def _extract_urls_from_hints(self, hints: list[str]) -> list[str]:
+        """Extract URLs from task hints."""
+        urls = []
+        for hint in hints:
+            if hint.startswith(("http://", "https://")):
+                urls.append(hint)
+            elif "://" not in hint and "." in hint:
+                # Might be a domain without protocol
+                urls.append(f"https://{hint}")
+        return urls
+
+    def _build_search_query(self, task_context: Any) -> str:
+        """Build a search query from task context."""
+        parts = [task_context.task_name]
+        if task_context.target_fields:
+            parts.extend(task_context.target_fields[:2])
+        return " ".join(parts)
+
+    def _create_navigate_action(self, url: str, observation: Observation) -> Action:
+        """Create a navigate action for the given URL."""
+        return Action(
+            action_type=ActionType.NAVIGATE,
+            parameters={"url": url, "timeout_ms": 30000},
+            reasoning=f"Navigating to relevant URL: {url}",
+            confidence=0.75,
+            agent_id=self.agent_id,
+        )
+
+    def _create_go_back_action(self, reason: str) -> Action:
+        """Create a go back action."""
+        return Action(
+            action_type=ActionType.GO_BACK,
+            parameters={},
+            reasoning=reason,
+            confidence=0.8,
+            agent_id=self.agent_id,
+        )
+
+    def get_visited_urls(self) -> set[str]:
+        """Get the set of visited URLs."""
+        return self._visited_urls.copy()
+
+    def reset(self) -> None:
+        """Reset the navigator state."""
+        super().reset()
+        self._visited_urls.clear()
+        self._url_scores.clear()

backend/app/agents/planner.py

@@ -0,0 +1,242 @@
+"""Planner agent for goal decomposition and task planning."""
+
+from typing import Any
+
+from app.core.action import Action, ActionType
+from app.core.observation import Observation
+
+from .base import BaseAgent
+
+
+class PlannerAgent(BaseAgent):
+    """
+    Agent responsible for high-level planning and goal decomposition.
+    
+    The PlannerAgent analyzes the task requirements and creates
+    structured plans that other agents can execute. It handles:
+    - Breaking down complex tasks into subtasks
+    - Determining the optimal sequence of actions
+    - Adapting plans based on execution results
+    - Coordinating multi-step extraction workflows
+    """
+
+    def __init__(
+        self,
+        agent_id: str = "planner",
+        config: dict[str, Any] | None = None,
+    ):
+        """
+        Initialize the PlannerAgent.
+        
+        Args:
+            agent_id: Unique identifier for this agent.
+            config: Optional configuration with keys:
+                - max_plan_depth: Maximum depth of nested plans (default: 5)
+                - replan_threshold: Error count before replanning (default: 2)
+                - planning_model: LLM model to use for planning
+        """
+        super().__init__(agent_id, config)
+        self.max_plan_depth = self.config.get("max_plan_depth", 5)
+        self.replan_threshold = self.config.get("replan_threshold", 2)
+        self._current_plan: list[Action] | None = None
+        self._plan_step: int = 0
+
+    async def act(self, observation: Observation) -> Action:
+        """
+        Select the next action based on the current plan or create a new one.
+        
+        If no plan exists or the current plan has failed, creates a new plan.
+        Otherwise, returns the next action in the current plan.
+        
+        Args:
+            observation: The current state observation.
+            
+        Returns:
+            The next action to execute.
+        """
+        try:
+            # Check if we need to replan due to errors
+            if observation.consecutive_errors >= self.replan_threshold:
+                self._current_plan = None
+                self._plan_step = 0
+
+            # Create plan if none exists
+            if self._current_plan is None or self._plan_step >= len(self._current_plan):
+                self._current_plan = await self.plan(observation)
+                self._plan_step = 0
+
+            if not self._current_plan:
+                return self._create_done_action("No actions planned")
+
+            # Get next action from plan
+            action = self._current_plan[self._plan_step]
+            action.plan_step = self._plan_step
+            action.agent_id = self.agent_id
+            self._plan_step += 1
+
+            return action
+
+        except Exception as e:
+            return self._create_error_action(f"Planning error: {e}")
+
+    async def plan(self, observation: Observation) -> list[Action]:
+        """
+        Create a plan of actions to achieve the task goals.
+        
+        Analyzes the observation to determine:
+        - What fields still need to be extracted
+        - What navigation may be required
+        - What verification steps are needed
+        
+        Args:
+            observation: The current state observation.
+            
+        Returns:
+            A list of planned actions in execution order.
+        """
+        try:
+            actions: list[Action] = []
+            task_context = observation.task_context
+
+            if not task_context:
+                return [self._create_done_action("No task context provided")]
+
+            # Determine remaining fields to extract
+            remaining_fields = observation.fields_remaining
+            extracted_fields = [f.field_name for f in observation.extracted_so_far]
+
+            # If no URL loaded, plan navigation first
+            if not observation.current_url:
+                search_action = self._plan_initial_navigation(task_context)
+                if search_action:
+                    actions.append(search_action)
+
+            # Plan extraction for remaining fields
+            for field in remaining_fields:
+                extraction_action = self._plan_field_extraction(
+                    field,
+                    observation,
+                )
+                actions.append(extraction_action)
+
+            # Plan verification if fields have been extracted
+            if extracted_fields:
+                verify_action = self._plan_verification(extracted_fields)
+                actions.append(verify_action)
+
+            # Add completion action
+            actions.append(
+                Action(
+                    action_type=ActionType.DONE,
+                    parameters={"success": True, "message": "Plan completed"},
+                    reasoning="All planned steps completed",
+                    confidence=0.9,
+                    agent_id=self.agent_id,
+                )
+            )
+
+            return actions
+
+        except Exception as e:
+            return [self._create_error_action(f"Plan creation failed: {e}")]
+
+    def _plan_initial_navigation(self, task_context: Any) -> Action | None:
+        """Plan initial navigation based on task context."""
+        if task_context.hints:
+            # Use hints for navigation
+            for hint in task_context.hints:
+                if hint.startswith("http"):
+                    return Action(
+                        action_type=ActionType.NAVIGATE,
+                        parameters={"url": hint},
+                        reasoning=f"Navigating to hinted URL: {hint}",
+                        confidence=0.85,
+                        agent_id=self.agent_id,
+                    )
+
+        # Default to search
+        search_query = f"{task_context.task_name} site information"
+        return Action(
+            action_type=ActionType.SEARCH_ENGINE,
+            parameters={"query": search_query, "engine": "google"},
+            reasoning=f"Searching for: {search_query}",
+            confidence=0.7,
+            agent_id=self.agent_id,
+        )
+
+    def _plan_field_extraction(
+        self,
+        field_name: str,
+        observation: Observation,
+    ) -> Action:
+        """Plan extraction for a specific field."""
+        # Check if we have page elements that might contain the field
+        selector = None
+        confidence = 0.6
+
+        for element in observation.page_elements:
+            element_text = (element.text or "").lower()
+            if field_name.lower() in element_text:
+                selector = element.selector
+                confidence = 0.8
+                break
+
+        return Action(
+            action_type=ActionType.EXTRACT_FIELD,
+            parameters={
+                "field_name": field_name,
+                "selector": selector,
+                "extraction_method": "text",
+            },
+            reasoning=f"Extracting field: {field_name}",
+            confidence=confidence,
+            agent_id=self.agent_id,
+        )
+
+    def _plan_verification(self, fields: list[str]) -> Action:
+        """Plan verification for extracted fields."""
+        return Action(
+            action_type=ActionType.VERIFY_FIELD,
+            parameters={
+                "field_name": fields[0] if fields else "unknown",
+                "validation_rules": ["not_empty", "format_check"],
+            },
+            reasoning=f"Verifying extracted fields: {fields}",
+            confidence=0.75,
+            agent_id=self.agent_id,
+        )
+
+    def _create_done_action(self, message: str) -> Action:
+        """Create a done action."""
+        return Action(
+            action_type=ActionType.DONE,
+            parameters={"success": True, "message": message},
+            reasoning=message,
+            confidence=1.0,
+            agent_id=self.agent_id,
+        )
+
+    def _create_error_action(self, error: str) -> Action:
+        """Create a fail action for errors."""
+        return Action(
+            action_type=ActionType.FAIL,
+            parameters={"success": False, "message": error},
+            reasoning=error,
+            confidence=1.0,
+            agent_id=self.agent_id,
+        )
+
+    def get_current_plan(self) -> list[Action] | None:
+        """Get the current plan."""
+        return self._current_plan
+
+    def get_plan_progress(self) -> tuple[int, int]:
+        """Get current plan progress as (current_step, total_steps)."""
+        total = len(self._current_plan) if self._current_plan else 0
+        return (self._plan_step, total)
+
+    def reset(self) -> None:
+        """Reset the planner state."""
+        super().reset()
+        self._current_plan = None
+        self._plan_step = 0

backend/app/agents/verifier.py

@@ -0,0 +1,468 @@
+"""Verifier agent for cross-source verification."""
+
+import re
+from typing import Any
+
+from app.core.action import Action, ActionType
+from app.core.observation import ExtractedField, Observation
+
+from .base import BaseAgent
+
+
+class VerificationResult:
+    """Result of a verification check."""
+
+    def __init__(
+        self,
+        field_name: str,
+        is_valid: bool,
+        confidence: float,
+        issues: list[str] | None = None,
+        sources_checked: int = 0,
+    ):
+        """Initialize verification result."""
+        self.field_name = field_name
+        self.is_valid = is_valid
+        self.confidence = confidence
+        self.issues = issues or []
+        self.sources_checked = sources_checked
+
+    def to_dict(self) -> dict[str, Any]:
+        """Convert to dictionary."""
+        return {
+            "field_name": self.field_name,
+            "is_valid": self.is_valid,
+            "confidence": self.confidence,
+            "issues": self.issues,
+            "sources_checked": self.sources_checked,
+        }
+
+
+class VerifierAgent(BaseAgent):
+    """
+    Agent responsible for verifying extracted data.
+    
+    The VerifierAgent handles:
+    - Format validation (emails, URLs, dates, etc.)
+    - Cross-source verification
+    - Consistency checks across fields
+    - Confidence scoring for verified data
+    - Flagging suspicious or inconsistent data
+    """
+
+    def __init__(
+        self,
+        agent_id: str = "verifier",
+        config: dict[str, Any] | None = None,
+    ):
+        """
+        Initialize the VerifierAgent.
+        
+        Args:
+            agent_id: Unique identifier for this agent.
+            config: Optional configuration with keys:
+                - min_confidence: Minimum confidence to accept (default: 0.7)
+                - require_cross_validation: Require multiple sources (default: False)
+                - strict_mode: Apply stricter validation rules (default: False)
+        """
+        super().__init__(agent_id, config)
+        self.min_confidence = self.config.get("min_confidence", 0.7)
+        self.require_cross_validation = self.config.get("require_cross_validation", False)
+        self.strict_mode = self.config.get("strict_mode", False)
+        self._validation_rules = self._init_validation_rules()
+        self._verification_history: list[VerificationResult] = []
+
+    def _init_validation_rules(self) -> dict[str, list[dict[str, Any]]]:
+        """Initialize validation rules for common field types."""
+        return {
+            "email": [
+                {
+                    "type": "regex",
+                    "pattern": r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
+                    "error": "Invalid email format",
+                },
+            ],
+            "url": [
+                {
+                    "type": "regex",
+                    "pattern": r"^https?://[^\s]+$",
+                    "error": "Invalid URL format",
+                },
+            ],
+            "phone": [
+                {
+                    "type": "regex",
+                    "pattern": r"[\d\s\-\(\)\+]{7,}",
+                    "error": "Invalid phone format",
+                },
+            ],
+            "price": [
+                {
+                    "type": "range",
+                    "min": 0,
+                    "max": 1000000,
+                    "error": "Price out of reasonable range",
+                },
+            ],
+            "date": [
+                {
+                    "type": "regex",
+                    "pattern": r"\d{1,4}[-/]\d{1,2}[-/]\d{1,4}",
+                    "error": "Invalid date format",
+                },
+            ],
+            "rating": [
+                {
+                    "type": "range",
+                    "min": 0,
+                    "max": 5,
+                    "error": "Rating out of range",
+                },
+            ],
+        }
+
+    async def act(self, observation: Observation) -> Action:
+        """
+        Select the best verification action based on observation.
+        
+        Determines which extracted fields need verification and
+        selects the appropriate verification method.
+        
+        Args:
+            observation: The current state observation.
+            
+        Returns:
+            The verification action to execute.
+        """
+        try:
+            # Find unverified fields
+            unverified = [
+                f for f in observation.extracted_so_far
+                if not f.verified
+            ]
+
+            if not unverified:
+                return Action(
+                    action_type=ActionType.DONE,
+                    parameters={"success": True, "message": "All fields verified"},
+                    reasoning="No unverified fields remaining",
+                    confidence=1.0,
+                    agent_id=self.agent_id,
+                )
+
+            # Verify the first unverified field
+            field = unverified[0]
+            result = await self._verify_field(field, observation)
+
+            if result.is_valid and result.confidence >= self.min_confidence:
+                return Action(
+                    action_type=ActionType.VERIFY_FIELD,
+                    parameters={
+                        "field_name": field.field_name,
+                        "verified": True,
+                        "confidence": result.confidence,
+                        "issues": result.issues,
+                    },
+                    reasoning=f"Field {field.field_name} verified with confidence {result.confidence:.2f}",
+                    confidence=result.confidence,
+                    agent_id=self.agent_id,
+                )
+            else:
+                # Verification failed - may need re-extraction
+                return self._create_reverify_action(field, result)
+
+        except Exception as e:
+            return Action(
+                action_type=ActionType.FAIL,
+                parameters={"success": False, "message": str(e)},
+                reasoning=f"Verification error: {e}",
+                confidence=1.0,
+                agent_id=self.agent_id,
+            )
+
+    async def plan(self, observation: Observation) -> list[Action]:
+        """
+        Create a verification plan for all extracted fields.
+        
+        Args:
+            observation: The current state observation.
+            
+        Returns:
+            A list of planned verification actions.
+        """
+        try:
+            actions: list[Action] = []
+
+            # Plan verification for each unverified field
+            for field in observation.extracted_so_far:
+                if field.verified:
+                    continue
+
+                # Basic format verification
+                actions.append(
+                    Action(
+                        action_type=ActionType.VERIFY_FIELD,
+                        parameters={
+                            "field_name": field.field_name,
+                            "expected_type": self._infer_field_type(field.field_name),
+                        },
+                        reasoning=f"Verify format of {field.field_name}",
+                        confidence=0.8,
+                        agent_id=self.agent_id,
+                    )
+                )
+
+                # Cross-source verification if required
+                if self.require_cross_validation:
+                    actions.append(
+                        Action(
+                            action_type=ActionType.VERIFY_FACT,
+                            parameters={
+                                "claim": f"{field.field_name}: {field.value}",
+                                "confidence_threshold": self.min_confidence,
+                            },
+                            reasoning=f"Cross-validate {field.field_name} with other sources",
+                            confidence=0.7,
+                            agent_id=self.agent_id,
+                        )
+                    )
+
+            return actions
+
+        except Exception as e:
+            return [
+                Action(
+                    action_type=ActionType.FAIL,
+                    parameters={"message": f"Verification planning failed: {e}"},
+                    reasoning=str(e),
+                    confidence=1.0,
+                    agent_id=self.agent_id,
+                )
+            ]
+
+    async def _verify_field(
+        self,
+        field: ExtractedField,
+        observation: Observation,
+    ) -> VerificationResult:
+        """
+        Verify a single field.
+        
+        Args:
+            field: The field to verify.
+            observation: Current observation context.
+            
+        Returns:
+            Verification result.
+        """
+        issues: list[str] = []
+        confidence = field.confidence
+        sources_checked = 1
+
+        # Apply validation rules
+        field_type = self._infer_field_type(field.field_name)
+        format_valid, format_issues = self._validate_format(
+            field.value,
+            field_type,
+        )
+
+        if not format_valid:
+            issues.extend(format_issues)
+            confidence *= 0.5
+
+        # Check for empty or null values
+        if field.value is None or (
+            isinstance(field.value, str) and not field.value.strip()
+        ):
+            issues.append("Empty value")
+            confidence = 0.0
+
+        # Check against memory context for consistency
+        consistency_issues = self._check_consistency(field, observation)
+        if consistency_issues:
+            issues.extend(consistency_issues)
+            confidence *= 0.8
+
+        # Create result
+        result = VerificationResult(
+            field_name=field.field_name,
+            is_valid=len(issues) == 0,
+            confidence=confidence,
+            issues=issues,
+            sources_checked=sources_checked,
+        )
+
+        self._verification_history.append(result)
+        return result
+
+    def _validate_format(
+        self,
+        value: Any,
+        field_type: str,
+    ) -> tuple[bool, list[str]]:
+        """
+        Validate value format against rules.
+        
+        Args:
+            value: The value to validate.
+            field_type: The expected field type.
+            
+        Returns:
+            Tuple of (is_valid, list of issues).
+        """
+        if value is None:
+            return False, ["Value is None"]
+
+        issues: list[str] = []
+        rules = self._validation_rules.get(field_type, [])
+
+        value_str = str(value)
+
+        for rule in rules:
+            rule_type = rule.get("type")
+
+            if rule_type == "regex":
+                pattern = rule.get("pattern", "")
+                if not re.match(pattern, value_str):
+                    issues.append(rule.get("error", "Format validation failed"))
+
+            elif rule_type == "range":
+                try:
+                    num_value = float(value_str.replace(",", "").replace("$", ""))
+                    min_val = rule.get("min", float("-inf"))
+                    max_val = rule.get("max", float("inf"))
+                    if not (min_val <= num_value <= max_val):
+                        issues.append(rule.get("error", "Value out of range"))
+                except ValueError:
+                    issues.append("Cannot convert to number for range check")
+
+            elif rule_type == "length":
+                min_len = rule.get("min", 0)
+                max_len = rule.get("max", float("inf"))
+                if not (min_len <= len(value_str) <= max_len):
+                    issues.append(rule.get("error", "Length validation failed"))
+
+        return len(issues) == 0, issues
+
+    def _check_consistency(
+        self,
+        field: ExtractedField,
+        observation: Observation,
+    ) -> list[str]:
+        """
+        Check field consistency with other data.
+        
+        Args:
+            field: The field to check.
+            observation: Current observation.
+            
+        Returns:
+            List of consistency issues.
+        """
+        issues: list[str] = []
+
+        # Check against other extracted fields
+        for other in observation.extracted_so_far:
+            if other.field_name == field.field_name:
+                continue
+
+            # Example: price should be less than total_price
+            if field.field_name == "price" and other.field_name == "total_price":
+                try:
+                    price = float(str(field.value).replace("$", "").replace(",", ""))
+                    total = float(str(other.value).replace("$", "").replace(",", ""))
+                    if price > total:
+                        issues.append("Price exceeds total_price")
+                except (ValueError, TypeError):
+                    pass
+
+        # Check against memory for historical consistency
+        memory = observation.memory_context
+        if memory.long_term_relevant:
+            for mem in memory.long_term_relevant:
+                if mem.get("field") == field.field_name:
+                    historical_value = mem.get("value")
+                    if historical_value and historical_value != field.value:
+                        # Different from historical - flag for review
+                        issues.append(
+                            f"Value differs from historical: {historical_value}"
+                        )
+
+        return issues
+
+    def _infer_field_type(self, field_name: str) -> str:
+        """Infer the field type from its name."""
+        field_lower = field_name.lower()
+
+        type_keywords = {
+            "email": ["email", "mail"],
+            "url": ["url", "link", "href", "website"],
+            "phone": ["phone", "tel", "mobile", "fax"],
+            "price": ["price", "cost", "amount", "total", "fee"],
+            "date": ["date", "time", "created", "updated", "published"],
+            "rating": ["rating", "score", "stars"],
+        }
+
+        for field_type, keywords in type_keywords.items():
+            for keyword in keywords:
+                if keyword in field_lower:
+                    return field_type
+
+        return "text"
+
+    def _create_reverify_action(
+        self,
+        field: ExtractedField,
+        result: VerificationResult,
+    ) -> Action:
+        """Create an action to handle failed verification."""
+        if result.confidence < 0.3:
+            # Very low confidence - suggest re-extraction
+            return Action(
+                action_type=ActionType.EXTRACT_FIELD,
+                parameters={
+                    "field_name": field.field_name,
+                    "reason": "Re-extracting due to verification failure",
+                },
+                reasoning=f"Verification failed with issues: {result.issues}",
+                confidence=0.6,
+                agent_id=self.agent_id,
+            )
+        else:
+            # Moderate confidence - try cross-validation
+            return Action(
+                action_type=ActionType.VERIFY_FACT,
+                parameters={
+                    "claim": f"{field.field_name}: {field.value}",
+                    "sources": None,
+                    "confidence_threshold": self.min_confidence,
+                },
+                reasoning=f"Attempting cross-validation for {field.field_name}",
+                confidence=0.5,
+                agent_id=self.agent_id,
+            )
+
+    def add_validation_rule(
+        self,
+        field_type: str,
+        rule: dict[str, Any],
+    ) -> None:
+        """
+        Add a custom validation rule.
+        
+        Args:
+            field_type: The field type this rule applies to.
+            rule: The validation rule dictionary.
+        """
+        if field_type not in self._validation_rules:
+            self._validation_rules[field_type] = []
+        self._validation_rules[field_type].append(rule)
+
+    def get_verification_history(self) -> list[dict[str, Any]]:
+        """Get verification history as dictionaries."""
+        return [r.to_dict() for r in self._verification_history]
+
+    def reset(self) -> None:
+        """Reset the verifier state."""
+        super().reset()
+        self._verification_history.clear()