3.2Robust Error Handling

A production ReAct agent must handle three categories of errors:

Category 1: Agent Errors (the agent makes a bad decision)

class AgentDecisionError(Exception):
    """Agent produced an unparseable or invalid action."""
    pass

def parse_decision(self, raw_response: str) -> Action:
    """Parse the LLM's response into a structured action."""
    raw = raw_response.strip()

    # Handle malformed responses
    if not raw:
        raise AgentDecisionError("Empty response from LLM")

    # Try to parse known action formats
    for prefix in ["NAVIGATE", "CLICK", "TYPE", "ASSERT", "DONE"]:
        if raw.upper().startswith(prefix):
            return Action(type=prefix, payload=raw[len(prefix):].strip())

    # If no known action matches, ask the LLM to retry
    retry_prompt = f"""
    Your previous response was not a valid action: "{raw}"
    Please respond with exactly one of:
    - NAVIGATE <url>
    - CLICK <selector>
    - TYPE <selector> <text>
    - ASSERT <condition>
    - DONE <pass|fail> <reason>
    """
    retry_response = self.llm.generate(retry_prompt)
    return self.parse_decision(retry_response)  # One retry only

Category 2: Environment Errors (the system under test fails)

def execute_with_recovery(self, action: Action) -> ActionResult:
    """Execute an action with environment error recovery."""
    try:
        return self.execute(action)
    except ElementNotFoundError as e:
        # Try alternative selectors
        alternatives = self.browser.find_similar(action.selector)
        if alternatives:
            return ActionResult(
                status="recovered",
                message=f"Original selector '{action.selector}' not found. "
                        f"Found alternatives: {alternatives}",
                alternatives=alternatives
            )
        return ActionResult(status="failed", message=str(e))
    except TimeoutError:
        # Take a screenshot for debugging
        screenshot = self.browser.screenshot(f"timeout_step_{self.step_count}.png")
        return ActionResult(
            status="timeout",
            message="Action timed out",
            screenshot=screenshot
        )
    except NavigationError as e:
        # Page failed to load
        return ActionResult(
            status="nav_error",
            message=f"Navigation failed: {e}",
            url=self.browser.current_url()
        )

Category 3: Infrastructure Errors (the agent infrastructure fails)

def run_with_infrastructure_safety(self, objective: str) -> TestResult:
    """Run a test with infrastructure-level safety nets."""
    try:
        return self.run(objective)
    except LLMRateLimitError:
        return TestResult(status="ABORTED", reason="LLM rate limit exceeded")
    except LLMTimeoutError:
        return TestResult(status="ABORTED", reason="LLM response timeout")
    except BrowserCrashError:
        return TestResult(status="ABORTED", reason="Browser process crashed")
    except Exception as e:
        # Catch-all for unexpected errors
        return TestResult(
            status="ERROR",
            reason=f"Unexpected infrastructure error: {type(e).__name__}: {e}",
            screenshot=self.safe_screenshot()
        )