Spaces:
Sleeping
Sleeping
| 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"<html><body>Sample content for {url}</body></html>") | |
| 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 | |