How to Build an AI Browser Recovery Harness with CapSolver

Business goal from agent
          ↓
Browser recovery harness
  ├─ target policy
  ├─ Playwright context
  ├─ state classifier
  ├─ checkpoint store
  ├─ CapSolver recovery
  ├─ retry budget
  ├─ trace + artifacts
  └─ cleanup
          ↓
Approved page action or operator review

from dataclasses import dataclass, field
from urllib.parse import urlparse

@dataclass(frozen=True)
class TargetPolicy:
    allowed_hosts: set[str]
    allowed_path_prefixes: tuple[str, ...]
    max_navigations: int = 20
    max_recovery_attempts: int = 1
    capture_screenshots: bool = True
    capture_html: bool = False
    allow_form_submission: bool = False

    def validate_url(self, url: str) -> None:
        parsed = urlparse(url)
        if parsed.scheme != "https":
            raise PermissionError("Only HTTPS targets are permitted")
        if parsed.hostname not in self.allowed_hosts:
            raise PermissionError("Host is outside the approved policy")
        if not parsed.path.startswith(self.allowed_path_prefixes):
            raise PermissionError("Path is outside the approved policy")

from enum import Enum

class BrowserState(str, Enum):
    EXPECTED_PAGE = "expected_page"
    SUPPORTED_CHALLENGE = "supported_challenge"
    UNKNOWN_PAGE = "unknown_page"
    RECOVERING = "recovering"
    RECOVERED = "recovered"
    REVIEW_REQUIRED = "review_required"
    FAILED = "failed"

ALLOWED_TRANSITIONS = {
    BrowserState.EXPECTED_PAGE: {
        BrowserState.EXPECTED_PAGE,
        BrowserState.SUPPORTED_CHALLENGE,
        BrowserState.UNKNOWN_PAGE,
    },
    BrowserState.SUPPORTED_CHALLENGE: {
        BrowserState.RECOVERING,
        BrowserState.REVIEW_REQUIRED,
    },
    BrowserState.RECOVERING: {
        BrowserState.RECOVERED,
        BrowserState.REVIEW_REQUIRED,
        BrowserState.FAILED,
    },
    BrowserState.RECOVERED: {
        BrowserState.EXPECTED_PAGE,
        BrowserState.REVIEW_REQUIRED,
    },
}

from dataclasses import dataclass
from datetime import datetime, timezone

@dataclass
class BrowserCheckpoint:
    url: str
    title: str
    expected_selector: str | None
    navigation_index: int
    recovery_attempts: int
    observed_at: str

async def checkpoint(page, expected_selector, nav_index, attempts):
    return BrowserCheckpoint(
        url=page.url,
        title=await page.title(),
        expected_selector=expected_selector,
        navigation_index=nav_index,
        recovery_attempts=attempts,
        observed_at=datetime.now(timezone.utc).isoformat(),
    )

async def classify_page(page, expected_selector: str) -> BrowserState:
    if await page.locator(expected_selector).count():
        return BrowserState.EXPECTED_PAGE

    title = (await page.title()).strip().lower()
    html = (await page.content()).lower()

    challenge_markers = (
        "just a moment...",
        "challenge-platform",
        "cf-chl-",
    )
    if any(marker in title or marker in html for marker in challenge_markers):
        return BrowserState.SUPPORTED_CHALLENGE

    return BrowserState.UNKNOWN_PAGE

import os
from capsolver_core import create_capsolver


def create_recovery_client():
    return create_capsolver(
        api_key=os.environ["CAPSOLVER_API_KEY"],
        default_timeout=120,
        polling_interval=5,
        request_timeout_ms=30000,
        source="ai-browser-recovery-harness",
        version="1.0.0",
    )

from capsolver_core import SolveOnPageOptions

async def recover_supported_challenge(
    cap,
    page,
    policy: TargetPolicy,
    recovery_attempts: int,
) -> dict:
    policy.validate_url(page.url)

    if recovery_attempts >= policy.max_recovery_attempts:
        return {
            "success": False,
            "state": BrowserState.REVIEW_REQUIRED,
            "reason": "recovery budget exhausted",
        }

    detected = await cap.detect(page)
    if not detected:
        return {
            "success": False,
            "state": BrowserState.REVIEW_REQUIRED,
            "reason": "no supported challenge detected",
        }

    infos = await cap.get_captcha_info(page)
    results = await cap.solve_on_page(
        page,
        options=SolveOnPageOptions(
            autofill=True,
            throw_on_error=False,
            timeout=120,
            polling_interval=5,
        ),
    )

    errors = [item.error for item in results if item.error]
    filled = bool(results) and all(item.filled for item in results)

    return {
        "success": filled and not errors,
        "state": (
            BrowserState.RECOVERED
            if filled and not errors
            else BrowserState.REVIEW_REQUIRED
        ),
        "detected_count": len(detected),
        "info_count": len(infos),
        "result_count": len(results),
        "errors": errors,
    }

async def verify_recovery(
    page,
    expected_selector: str,
    timeout_ms: int = 15000,
) -> bool:
    try:
        await page.locator(expected_selector).wait_for(
            state="visible",
            timeout=timeout_ms,
        )
        return True
    except Exception:
        return False

async def recover_and_verify(cap, page, policy, expected_selector, attempts):
    result = await recover_supported_challenge(
        cap=cap,
        page=page,
        policy=policy,
        recovery_attempts=attempts,
    )
    if not result["success"]:
        return result

    if not await verify_recovery(page, expected_selector):
        return {
            "success": False,
            "state": BrowserState.REVIEW_REQUIRED,
            "reason": "expected page did not return after recovery",
        }

    return {
        "success": True,
        "state": BrowserState.EXPECTED_PAGE,
        "reason": "page recovered and verified",
    }

from contextlib import asynccontextmanager
from playwright.async_api import async_playwright

@dataclass
class BrowserHarness:
    policy: TargetPolicy
    playwright: object
    browser: object
    context: object
    page: object
    capsolver: object
    navigation_count: int = 0
    recovery_attempts: int = 0

@asynccontextmanager
async def browser_recovery_harness(policy: TargetPolicy):
    async with async_playwright() as playwright:
        browser = await playwright.chromium.launch(headless=True)
        context = await browser.new_context()
        page = await context.new_page()

        async with create_recovery_client() as cap:
            harness = BrowserHarness(
                policy=policy,
                playwright=playwright,
                browser=browser,
                context=context,
                page=page,
                capsolver=cap,
            )
            try:
                yield harness
            finally:
                await context.close()
                await browser.close()

async def safe_navigate(
    harness: BrowserHarness,
    url: str,
    expected_selector: str,
) -> dict:
    harness.policy.validate_url(url)

    if harness.navigation_count >= harness.policy.max_navigations:
        return {
            "success": False,
            "state": BrowserState.REVIEW_REQUIRED,
            "reason": "navigation budget exhausted",
        }

    harness.navigation_count += 1
    await harness.page.goto(url, wait_until="domcontentloaded")

    state = await classify_page(harness.page, expected_selector)
    if state == BrowserState.EXPECTED_PAGE:
        return {"success": True, "state": state}

    if state == BrowserState.SUPPORTED_CHALLENGE:
        result = await recover_and_verify(
            cap=harness.capsolver,
            page=harness.page,
            policy=harness.policy,
            expected_selector=expected_selector,
            attempts=harness.recovery_attempts,
        )
        harness.recovery_attempts += 1
        return result

    return {
        "success": False,
        "state": BrowserState.REVIEW_REQUIRED,
        "reason": "unknown page state",
    }

from opentelemetry import trace

tracer = trace.get_tracer("capsolver.browser_harness")

async def traced_safe_navigate(harness, url, expected_selector):
    with tracer.start_as_current_span("browser.safe_navigate") as span:
        span.set_attribute("browser.target_host", url.split("/")[2])
        span.set_attribute("browser.navigation_index", harness.navigation_count + 1)
        span.set_attribute("browser.recovery_attempts", harness.recovery_attempts)

        result = await safe_navigate(harness, url, expected_selector)

        span.set_attribute("browser.outcome", str(result.get("state")))
        span.set_attribute("browser.success", bool(result.get("success")))
        return result

from pathlib import Path
import secrets

async def capture_failure_artifacts(harness, directory: Path) -> dict:
    artifact_id = secrets.token_hex(12)
    screenshot = directory / f"{artifact_id}.png"

    await harness.page.screenshot(
        path=str(screenshot),
        full_page=False,
    )

    return {
        "artifact_id": artifact_id,
        "screenshot_path": str(screenshot),
        "url": harness.page.url,
        "title": await harness.page.title(),
    }

import pytest

@pytest.mark.asyncio
async def test_unknown_host_is_rejected():
    policy = TargetPolicy(
        allowed_hosts={"staging.example.com"},
        allowed_path_prefixes=("/qa/",),
    )

    with pytest.raises(PermissionError):
        policy.validate_url("https://other.example.net/qa/test")

@pytest.mark.asyncio
async def test_recovery_budget_is_bounded(fake_cap, fake_page):
    policy = TargetPolicy(
        allowed_hosts={"staging.example.com"},
        allowed_path_prefixes=("/qa/",),
        max_recovery_attempts=1,
    )

    result = await recover_supported_challenge(
        cap=fake_cap,
        page=fake_page,
        policy=policy,
        recovery_attempts=1,
    )

    assert result["state"] == BrowserState.REVIEW_REQUIRED
    assert result["reason"] == "recovery budget exhausted"