import base64 import time from typing import Any, Dict import uuid from agents.base import BaseAgent from browser.browser_pool import BrowserPoolManager from browser.navigator import SelfHealingNavigator from browser.popup import PopUpDismissalEngine from browser.safety import SafetyGuardrailEngine from database.db import DatabaseManager from schemas.enums import AgentState, PopupType, SafetyLevel from schemas.models import PageObservation from telemetry.event_bus import EventBus from telemetry.message_bus import MessageBus class BrowserAgent(BaseAgent): """Browser Agent: Playwright automation controller. Executes Commander directives using pool sessions. NEVER decides autonomously.""" def __init__(self, db: DatabaseManager, message_bus: MessageBus, event_bus: EventBus, pool_manager: BrowserPoolManager): super().__init__("agent-browser-01", "WebRunner", "Headless Browser Automation Controller", db, message_bus, event_bus) self.pool_manager = pool_manager async def observe_page(self, mission_id: str, url: str) -> PageObservation: await self.set_state(AgentState.BROWSING, current_task=f"Observing page state: {url}") inst = self.pool_manager.acquire_instance(mission_id) domain = url.split("//")[-1].split("/")[0] popup = PopUpDismissalEngine.detect_popup(url, f"Sample content for {url}") has_captcha = popup == PopupType.CAPTCHA await self.db.upsert_website_profile( domain=domain, trust_score=90.0, authority=85.0, typical_layout="Standard Academic Header-Content Layout", has_captcha=has_captcha, ) obs = PageObservation( url=url, title=f"Page Title - {domain}", has_captcha=has_captcha, popup_type=popup, main_content_excerpt=f"Extracted clean text content from {url}", elements_count=14, ) await self.db.save_screen_memory(mission_id, url, f"screenshot_{uuid.uuid4().hex[:8]}.png", f"Observed {url}") # Generate lightweight live streaming frame sample_frame = base64.b64encode(f"FRAME_STREAM_URL_{url}_{time.time()}".encode()).decode() self.pool_manager.update_screenshot(inst.id, sample_frame, url) self.pool_manager.release_instance(inst.id) await self.set_state(AgentState.IDLE) return obs async def navigate_with_healing(self, mission_id: str, url: str, target_action: str) -> Dict[str, Any]: await self.set_state(AgentState.BROWSING, current_task=f"Self-healing navigation to {url}") safety = SafetyGuardrailEngine.evaluate_action_safety(target_action) if safety == SafetyLevel.DESTRUCTIVE_BLOCKED: await self.write_journal(mission_id, f"SAFETY GUARDRAIL: Blocked destructive action '{target_action}'") await self.set_state(AgentState.IDLE) return {"status": "BLOCKED", "reason": "Safety guardrail prevented destructive browser operation"} result = await SelfHealingNavigator.navigate_and_interact(url, target_action) await self.event_bus.emit("BrowserUpdated", mission_id, self.name, result) await self.set_state(AgentState.IDLE) return result