"""Browser session abstraction with a zero-dependency simulated backend. The SimulatedBrowser models the bundled demo "vendor portal" (an order form and a pending-orders list) so the agentic loop, tool calls, page-state sanitization, and metrics are all exercised end-to-end without Playwright or a network. A real PlaywrightSession (same interface) is used automatically when Playwright is installed — swapping is a one-line change. """ from __future__ import annotations import importlib.util import uuid from dataclasses import dataclass, field def _has(mod: str) -> bool: return importlib.util.find_spec(mod) is not None # --- demo portal data --------------------------------------------------------- PENDING_ORDERS = [ {"order_id": "PO-100481", "supplier": "Acme Industrial", "amount": 12450.00, "status": "PENDING"}, {"order_id": "PO-100482", "supplier": "GlobalParts GmbH", "amount": 3820.50, "status": "PENDING"}, {"order_id": "PO-100483", "supplier": "Initech Supplies", "amount": 980.00, "status": "PENDING"}, ] ORDER_FORM_FIELDS = [ {"selector": "#vendor-id", "label": "Vendor ID", "value": ""}, {"selector": "#sku", "label": "SKU", "value": ""}, {"selector": "#qty", "label": "Quantity", "value": ""}, {"selector": "#delivery-date", "label": "Delivery Date", "value": ""}, ] # Complex order the ERP "Create Order" modal exposes (read by the agent). COMPLEX_ORDER = { "po_number": "PO-2026-88440", "vendor": "Meridian Industrial Components Ltd", "contact": "j.harlow@meridian.example", "incoterms": "DAP", "payment_terms": "Net 45", "priority": "Expedited", "ship_to": {"facility": "Aperture DC-West (Bay 14)", "address": "4400 Logistics Pkwy, Reno NV 89506, USA", "need_by": "2026-07-09"}, "line_items": [ {"sku": "HX-220", "description": "Hydraulic pump HX-220", "qty": 4, "unit_price": 1250.00, "amount": 5000.00}, {"sku": "SK-12", "description": "Seal kit (set of 12)", "qty": 10, "unit_price": 85.50, "amount": 855.00}, {"sku": "PG-300", "description": "Pressure gauge 0-300psi", "qty": 6, "unit_price": 142.75, "amount": 856.50}, {"sku": "LAB-1", "description": "Installation labor", "qty": 1, "unit_price": 2400.00, "amount": 2400.00}, ], "subtotal": 9111.50, "tax": 751.70, "freight": 180.00, "total": 10043.20, "currency": "USD", "approver": "Priya Anand (Procurement Lead)", } @dataclass class ActionRecord: tool: str args: dict ok: bool note: str = "" @dataclass class SimulatedBrowser: """In-memory model of the demo portal.""" url: str = "about:blank" fields: list = field(default_factory=lambda: [dict(f) for f in ORDER_FORM_FIELDS]) submitted: bool = False confirmation_id: str | None = None screenshots: int = 0 log: list = field(default_factory=list) backend: str = "simulated" erp_stage: str | None = None # None | dashboard | procurement | form # --- tool primitives --- def navigate(self, url: str) -> ActionRecord: self.url = url note = f"loaded {url}" if "/erp" in url: self.erp_stage = "dashboard" note = f"loaded ERP dashboard ({url})" rec = ActionRecord("navigate", {"url": url}, True, note) self.log.append(rec) return rec def click(self, selector: str) -> ActionRecord: ok = True sel = selector.lower() note = f"clicked {selector}" # --- multi-step ERP flow: dashboard → procurement → order modal --- if self.erp_stage is not None: if "procurement" in sel: self.erp_stage = "procurement" note = "opened Procurement module" elif "create-order" in sel or "create order" in sel: self.erp_stage = "form" note = "clicked '+ Create Order' → order form modal opened" elif "submit" in sel: self.submitted = True self.confirmation_id = f"PO-CONF-{uuid.uuid4().hex[:6].upper()}" note = f"submitted order for approval → {self.confirmation_id}" rec = ActionRecord("click", {"selector": selector}, ok, note) self.log.append(rec) return rec if "submit" in sel: self.submitted = True self.confirmation_id = f"CONF-{uuid.uuid4().hex[:6].upper()}" note = f"submitted form → confirmation {self.confirmation_id}" rec = ActionRecord("click", {"selector": selector}, ok, note) self.log.append(rec) return rec def fill(self, selector: str, text: str) -> ActionRecord: ok = False for f in self.fields: if f["selector"] == selector or selector.lstrip("#") in f["label"].lower().replace(" ", "-"): f["value"] = text ok = True break rec = ActionRecord("fill", {"selector": selector, "text": text}, ok, f"filled {selector}='{text}'" if ok else f"no field {selector}") self.log.append(rec) return rec def extract(self) -> dict: if self.erp_stage == "form": self.log.append(ActionRecord("extract", {}, True, "read complex order form fields")) return dict(COMPLEX_ORDER) if "orders" in self.url and "new" not in self.url: data = {"orders": PENDING_ORDERS} elif self.submitted: data = {"confirmation_id": self.confirmation_id, "order": {f["label"]: f["value"] for f in self.fields}} else: data = {f["label"]: f["value"] for f in self.fields} self.log.append(ActionRecord("extract", {}, True, "extracted page data")) return data def screenshot(self) -> str: self.screenshots += 1 self.log.append(ActionRecord("screenshot", {}, True, f"screenshot #{self.screenshots}")) return f"screenshot-{self.screenshots}.png" # --- sanitized page state for the LLM (the 'compact markdown DOM') --- def get_state(self) -> str: # multi-step ERP flow states if self.erp_stage == "dashboard": return ("# Acme ERP — Dashboard\nURL: " + self.url + "\n\nModules (tiles):\n" "- Procurement (link `#tile-procurement` → procurement/) — create & manage POs\n" "- Invoices `#tile-invoices` · Suppliers `#tile-suppliers` · Reports `#tile-reports`\n" "Goal hint: open Procurement to create/read an order.") if self.erp_stage == "procurement": return ("# Acme ERP — Procurement\nURL: " + self.url + "\n\nA table of purchase orders is shown.\n" "Button: '+ Create Order' (selector `#create-order`) — opens the new-order form modal.") if self.erp_stage == "form": o = COMPLEX_ORDER lines = "\n".join(f" - {li['sku']} {li['description']} x{li['qty']} @ ${li['unit_price']:.2f} = ${li['amount']:.2f}" for li in o["line_items"]) return (f"# New Purchase Order (modal)\nURL: {self.url}\n\n" f"PO: {o['po_number']} · Vendor: {o['vendor']} · Terms: {o['payment_terms']} · Priority: {o['priority']}\n" f"Ship to: {o['ship_to']['facility']}, {o['ship_to']['address']} (need-by {o['ship_to']['need_by']})\n" f"Line items:\n{lines}\n" f"Subtotal ${o['subtotal']:.2f} · Tax ${o['tax']:.2f} · Freight ${o['freight']:.2f} · " f"Total ${o['total']:.2f} {o['currency']} · Approver {o['approver']}\n" f"Buttons: 'Submit for approval' (`#submit-order`). Use extract() to read the fields.") if self.submitted: return (f"# Order Portal — Confirmation\nURL: {self.url}\n\n" f"Order submitted. Confirmation ID: **{self.confirmation_id}**\n" f"Fields: " + ", ".join(f'{f["label"]}={f["value"]}' for f in self.fields)) if "orders" in self.url and "new" not in self.url: rows = "\n".join( f"| {o['order_id']} | {o['supplier']} | {o['amount']} | {o['status']} |" for o in PENDING_ORDERS ) return ("# Order Portal — Pending Orders\nURL: " + self.url + "\n\n| Order | Supplier | Amount | Status |\n|---|---|---|---|\n" + rows) # default: the new-order form fields_md = "\n".join( f"- {f['label']} (selector `{f['selector']}`): " f"'{f['value']}'" if f["value"] else f"- {f['label']} (selector `{f['selector']}`): empty" for f in self.fields ) return (f"# Order Portal — New Order Form\nURL: {self.url}\n\n" f"Form fields:\n{fields_md}\n\nButton: 'Submit Order' (selector `#submit-order`)") class PlaywrightSession: """Real browser. Same interface as SimulatedBrowser. Lazily imports Playwright.""" backend = "playwright" def __init__(self, headless: bool = True) -> None: from playwright.sync_api import sync_playwright self._pw = sync_playwright().start() self._browser = self._pw.chromium.launch(headless=headless) self._ctx = self._browser.new_context(viewport={"width": 1280, "height": 800}) self.page = self._ctx.new_page() self.url = "about:blank" self.log: list = [] self.screenshots = 0 def navigate(self, url: str) -> ActionRecord: self.page.goto(url, wait_until="networkidle") self.url = url rec = ActionRecord("navigate", {"url": url}, True, f"loaded {url}") self.log.append(rec) return rec def click(self, selector: str) -> ActionRecord: try: self.page.click(selector, timeout=5000) ok, note = True, f"clicked {selector}" except Exception as e: ok, note = False, str(e) rec = ActionRecord("click", {"selector": selector}, ok, note) self.log.append(rec) return rec def fill(self, selector: str, text: str) -> ActionRecord: try: self.page.fill(selector, text, timeout=5000) ok, note = True, f"filled {selector}" except Exception as e: ok, note = False, str(e) rec = ActionRecord("fill", {"selector": selector, "text": text}, ok, note) self.log.append(rec) return rec def extract(self) -> dict: text = self.page.inner_text("body") self.log.append(ActionRecord("extract", {}, True, "extracted body text")) return {"text": text[:4000]} def screenshot(self) -> str: self.screenshots += 1 name = f"screenshot-{self.screenshots}.png" try: self.page.screenshot(path=name) except Exception: pass self.log.append(ActionRecord("screenshot", {}, True, name)) return name def get_state(self) -> str: try: from bs4 import BeautifulSoup # optional sanitizer html = self.page.content() soup = BeautifulSoup(html, "html.parser") for el in soup(["script", "style", "noscript", "svg", "img", "iframe"]): el.decompose() return soup.get_text(" ", strip=True)[:4000] except Exception: return self.page.inner_text("body")[:4000] def close(self) -> None: try: self._browser.close() self._pw.stop() except Exception: pass def get_session(headless: bool = True, prefer_simulated: bool | None = None): """Return a REAL Playwright session by default; the simulated session is used only in prototype mode (APERTURE_MODE=prototype) or when Playwright is absent. This keeps production web automation real while letting tests / the serverless demo run deterministically against the SimulatedBrowser (prototype/offline).""" if prefer_simulated is None: try: from ..config import get_settings prefer_simulated = get_settings().prototype_mode except Exception: prefer_simulated = False if not prefer_simulated and _has("playwright"): try: return PlaywrightSession(headless=headless) except Exception: pass return SimulatedBrowser()