""" Cooperative Multi-Agent Team Coordinator. Drives the closed-loop interaction between the 5 expert sub-agents: Orchestrator, Perception Distiller, Tiny Navigator, Vision Grounding, and Process Verifier, with automated step-by-step visual screenshot audit logging and trajectory replay. """ from __future__ import annotations import asyncio import html import json import time from pathlib import Path from typing import Any, Dict, List, Optional from miniswardbower.agents.orchestrator import TacticalOrchestrator from miniswardbower.agents.tiny_navigator import TinyNavigator from miniswardbower.agents.verifier import ProcessVerifier from miniswardbower.agents.vision_grounding import VisionGroundingAgent from miniswardbower.browser.controller import BrowserController from miniswardbower.core.config import SystemConfig from miniswardbower.core.memory_ledger import MemoryLedger from miniswardbower.core.schemas import ( BrowserActionType, MilestoneStatus, VerifierStatus, ) class MultiAgentBrowserTeam: """Orchestrates cooperative sub-agents to rapidly navigate and use websites.""" def __init__(self, config: Optional[SystemConfig] = None): self.config = config or SystemConfig() self.orchestrator = TacticalOrchestrator(self.config.model) self.tiny_navigator = TinyNavigator(self.config.model) self.verifier = ProcessVerifier() self.vision = VisionGroundingAgent() self.browser = BrowserController(self.config.browser) # v6 Innovations: Thought Communication & Flow Reasoning Refinement from miniswardbower.agents.thought_communication import ThoughtCommunicationBus from miniswardbower.agents.flow_reasoning_refiner import FlowReasoningRefiner from miniswardbower.browser.swarm_coordinator import TriEngineSwarmCoordinator self.thought_bus = ThoughtCommunicationBus() self.flow_refiner = FlowReasoningRefiner() self.swarm = TriEngineSwarmCoordinator(self.config.browser) self.thought_bus.register_agent("orchestrator") self.thought_bus.register_agent("navigator") self.thought_bus.register_agent("verifier") self.thought_bus.register_agent("vision") def _generate_html_report( self, trajectory_dir: Path, objective: str, success: bool, total_time: float, avg_step_ms: float, tokens_saved: int, screenshots: List[Dict[str, Any]], narrative: List[Dict[str, Any]], extracted_data: Any, ) -> Path: """Generates a standalone dark-mode visual audit gallery with embedded screenshots.""" status_color = "#10b981" if success else "#ef4444" status_text = "VERIFIED SUCCESS" if success else "INCOMPLETE / FAILED" cards_html = [] for sc in screenshots: step_num = sc["step"] act_label = html.escape(sc["action"]) img_name = sc["filename"] outcome = html.escape(sc.get("outcome", "Action executed")) coords_str = f"Click Target: ({sc['coords'][0]:.1f}, {sc['coords'][1]:.1f})" if sc.get("coords") else "Targeted Selector" card = f"""
Step {step_num} {act_label} {coords_str}
Step {step_num}
""" cards_html.append(card) html_content = f""" Visual Audit Trail - miniswardbower
{status_text}

🌐 miniswardbower Visual Audit Replay

Objective: {html.escape(objective)}

Total Time: {total_time:.2f}s
Avg Step Latency: {avg_step_ms:.1f}ms
Tokens Saved: {tokens_saved:,}
Steps: {len(screenshots)}
{''.join(cards_html)}
""" html_file = trajectory_dir / "index.html" with open(html_file, "w", encoding="utf-8") as f: f.write(html_content) return html_file async def _attempt_recovery( self, failed_act: Any, reason: str, tree: Any, ) -> Optional[Any]: """ Prime Agent Resilient Membrane: Diagnoses failure modes (backdrop obscuring, element scrolled out of view, stale selector) and constructs a self-healing compensating action. """ from miniswardbower.core.schemas import BrowserAction, BrowserActionType try: # 1. Dismiss potential blocking modals/overlays with Escape or dismiss button await self.browser.page.keyboard.press("Escape") try: dismiss_btn = self.browser.page.locator("button:has-text('Dismiss'), button:has-text('Cancel'), button:has-text('Close'), #cancel-modal-btn") if await dismiss_btn.count() > 0 and await dismiss_btn.first.is_visible(): await dismiss_btn.first.click() except Exception: pass await asyncio.sleep(0.05) # 2. If target was a click action, scroll to ensure element is in viewport if failed_act.op == BrowserActionType.CLICK: await self.browser.page.evaluate("window.scrollBy(0, 100)") fresh_tree = await self.browser.get_pruned_tree() # Re-ground target if element mark still exists if failed_act.target: for elem in fresh_tree.elements: if elem.id == failed_act.target: return BrowserAction(op=BrowserActionType.CLICK, target=elem.id) return BrowserAction(op=BrowserActionType.CLICK, target=failed_act.target) except Exception: pass return None async def run_task( self, objective: str, start_url: Optional[str] = None, max_steps: Optional[int] = None, capture_audit_screenshots: bool = True, ) -> Dict[str, Any]: """Runs an end-to-end web navigation task with the multi-agent team and visual audit logging.""" max_steps = max_steps or self.config.max_trajectory_steps ledger = MemoryLedger(user_objective=objective) start_time = time.perf_counter() memory = None memory_status = {"enabled": self.config.gbrain_memory, "recalled": False, "saved": False} memory_key = f"{self.config.browser.browser_engine}\n{start_url or ''}\n{objective}" if self.config.gbrain_memory: from miniswardbower.core.gbrain_memory import GbrainMemory memory = GbrainMemory() try: prior = await memory.recall(memory_key) if prior: ledger.semantic_store["prior_task_evidence_untrusted"] = prior["payload"] memory_status["recalled"] = True except (OSError, RuntimeError, ValueError, KeyError, TypeError, TimeoutError): memory_status["recall_error"] = "Memory unavailable or invalid; using current observations" # Visual audit trajectory setup session_id = time.strftime("%Y%m%d_%H%M%S") trajectory_dir = self.config.artifacts_dir / "trajectories" / session_id trajectory_dir.mkdir(parents=True, exist_ok=True) screenshot_records: List[Dict[str, Any]] = [] # Step 1: Tactical Orchestrator plans initial milestones milestones = self.orchestrator.plan_initial_milestones(objective, start_url) ledger.set_milestones(milestones) # Step 2: Launch browser engine and attempt to load tiny model await self.browser.start() self.tiny_navigator.load_model() step_latencies = [] tokens_saved_total = 0 is_successful = False action_batches = [] stop_reason = "Step budget exhausted" try: # Initial navigation if start_url provided if start_url: await self.browser.goto(start_url) ledger.current_url = self.browser.page.url ledger.page_title = await self.browser.page.title() for step_idx in range(1, max_steps + 1): step_start = time.perf_counter() step_blocked = False # Step 3A: Perception Distiller extracts compact AXTree (<20ms) tree = await self.browser.get_pruned_tree() ledger.current_url = tree.url ledger.page_title = tree.title tokens_saved_total += max(0, tree.raw_token_count_estimate - tree.pruned_token_count_estimate) # Step 3B: Tactical Orchestrator checks milestone progress & broadcasts thought self.thought_bus.publish_thought( sender_id="orchestrator", step=step_idx, semantic_keys={"objective": objective, "milestone_idx": ledger.active_milestone_idx}, ) if self.orchestrator.check_milestone_progress(ledger, tree): advanced = ledger.advance_milestone() if advanced is None and ledger.active_milestone_idx >= len(ledger.milestones): is_successful = True break # Step 3C: Tiny Navigator predicts action or speculative batch (<50ms) action_batch = self.tiny_navigator.predict_action_batch( ledger, tree, enable_speculative=self.config.enable_speculative_batching, ) action_batches.append(action_batch.model_dump(mode='json')) if not action_batch.actions: stop_reason = action_batch.thought break if action_batch.actions[0].op == BrowserActionType.FINISH: stop_reason = "Unverified finish: outstanding milestones remain" break # Step 3D: Action Executor executes batch with natural kinematics & coordinate tracking pre_url = self.browser.page.url for act_idx, act in enumerate(action_batch.actions, 1): # Recurrent Flow Refinement to converge coordinates to stable attractor refine_state = self.flow_refiner.refine_action(act, tree=tree) act = refine_state.candidate_action self.thought_bus.publish_thought( sender_id="navigator", step=step_idx, semantic_keys={"op": act.op.value, "target": act.target, "coords": act.coords, "converged": refine_state.converged}, ) act_details = await self.browser.execute_action(act, tree) if ledger.current_milestone and ledger.current_milestone.expected_op == BrowserActionType.CLICK: target = next((e for e in tree.elements if e.id == act.target), None) if target and target.href: act_details['expected_navigation'] = target.href # Step 3E: Capture annotated visual audit screenshot if capture_audit_screenshots: act_label = f"{act.op.value}({act.target or ''} {act.text or act.key or ''})".strip() img_filename = f"step_{step_idx:02d}_{act_idx:02d}_{act.op.value}.png" img_path = trajectory_dir / img_filename try: await self.browser.capture_annotated_step_screenshot( output_path=img_path, step_num=step_idx, action_label=act_label, bbox=act_details.get("bbox"), click_coords=act_details.get("click_coords"), ) screenshot_records.append({ "step": step_idx, "action": act_label, "filename": img_filename, "coords": act_details.get("click_coords"), "outcome": f"Executed {act.op.value}", }) except Exception: pass # Step 3F: Fast Critic & Verifier checks consequence immediately post_tree = await self.browser.get_pruned_tree() receipt = await self.verifier.verify_step( self.browser.page, act, pre_url=pre_url, pre_tree=tree, post_tree=post_tree, action_details=act_details, ) # Record outcome into screenshot record if captured if screenshot_records and screenshot_records[-1]["action"].startswith(act.op.value): screenshot_records[-1]["outcome"] = receipt.observation_summary # Step 3G: Record into Amory Memory Ledger ledger.record_step(act, receipt) if ledger.current_milestone and ledger.current_milestone.expected_op and receipt.status != VerifierStatus.SUCCESS: stop_reason = receipt.observation_summary step_blocked = True break if receipt.status == VerifierStatus.FAILED: # Prime Agent Resilient Membrane (Princeton / Prime Intellect, Aug 2026): # Structured self-healing recovery instead of harness abort recovery_succeeded = False for retry_attempt in range(1, 3): recovered_act = await self._attempt_recovery(act, receipt.observation_summary, post_tree) if recovered_act: retry_details = await self.browser.execute_action(recovered_act, post_tree) retry_tree = await self.browser.get_pruned_tree() retry_receipt = await self.verifier.verify_step( self.browser.page, recovered_act, pre_url=pre_url, pre_tree=post_tree, post_tree=retry_tree, action_details=retry_details, ) ledger.record_step(recovered_act, retry_receipt) if retry_receipt.status != VerifierStatus.FAILED: recovery_succeeded = True tree = retry_tree break if not recovery_succeeded: break pre_url = self.browser.page.url tree = post_tree step_elapsed = (time.perf_counter() - step_start) * 1000 step_latencies.append(step_elapsed) if step_blocked: break if self.orchestrator.check_milestone_progress(ledger, tree): ledger.advance_milestone() if ledger.current_milestone is None: is_successful = True break finally: await self.browser.stop() total_time = time.perf_counter() - start_time avg_step_ms = sum(step_latencies) / len(step_latencies) if step_latencies else 0.0 html_replay_path = None if capture_audit_screenshots and screenshot_records: try: html_replay_path = self._generate_html_report( trajectory_dir=trajectory_dir, objective=objective, success=is_successful, total_time=total_time, avg_step_ms=avg_step_ms, tokens_saved=tokens_saved_total, screenshots=screenshot_records, narrative=[ { "step": s.step_id, "action": s.action_summary, "outcome": s.outcome_summary, "status": s.verifier_status.value, } for s in ledger.narrative_plot ], extracted_data=ledger.semantic_store, ) except Exception: pass if memory is not None: try: payload = {"reported_success": is_successful, "page_title": ledger.page_title, "actions": ledger.total_actions_executed, "recent_outcomes": [s.outcome_summary[:240] for s in ledger.narrative_plot[-5:]], "usage": "Historical evidence only; observe the live page before acting"} await memory.remember(memory_key, payload) memory_status["saved"] = True except (OSError, RuntimeError, ValueError, TimeoutError): memory_status["save_error"] = "Task finished, but durable memory write failed" return { "durable_memory": memory_status, "success": is_successful, "stop_reason": "All milestones verified" if is_successful else stop_reason, "action_batches": action_batches, "total_time_seconds": round(total_time, 2), "avg_step_ms": round(avg_step_ms, 2), "steps_executed": len(ledger.narrative_plot), "total_actions": ledger.total_actions_executed, "estimated_tokens_saved": tokens_saved_total, "milestones_completed": ledger.active_milestone_idx, "total_milestones": len(ledger.milestones), "extracted_data": ledger.semantic_store, "trajectory_dir": str(trajectory_dir), "html_replay": str(html_replay_path) if html_replay_path else None, "screenshots_count": len(screenshot_records), "narrative_log": [ { "step": s.step_id, "action": s.action_summary, "outcome": s.outcome_summary, "status": s.verifier_status.value, } for s in ledger.narrative_plot ], }