diff --git a/docs/probabilistic-policy.md b/docs/probabilistic-policy.md index 955bd5b..8ca547b 100644 --- a/docs/probabilistic-policy.md +++ b/docs/probabilistic-policy.md @@ -787,3 +787,51 @@ This work is a specific instance of a more general claim about LLM-driven system > Most performance engineering today uses hand-tuned heuristics where a real probabilistic policy would do better. The constraint that justified hand-tuning (microsecond budgets, no idle time) doesn't bind in LLM-driven workflows because the LLM itself is the slow part. Everything below the LLM has spare budget for inference that previous-generation systems couldn't afford. unbrowser is one place to demonstrate this pattern. If it works here, the same shape — frequentist hot path, Bayesian policy layer, persistent priors per workload class — applies to many systems-level decisions in LLM-adjacent infrastructure. + +--- + +## 13. Tool-invocation routing (the same frame, one layer up) + +The posteriors above decide what *unbrowser* does internally (run scripts? +settle? call an API?). The identical math governs the other half of the +system: which tool the *agent* reaches for next. Every agent decision is + +``` +P(call T next | evidence) +``` + +and each layer of the stack is one update: + +| Update | Mechanism | Owner | +|---|---|---| +| Prior | tool name + description (training-data semantics) | us, at design time | +| Prior shift | MCP `instructions` field at handshake | server | +| Evidence | `derive_tool_likelihoods()` — page features → per-tool scores | binary | +| Posterior | `next_tools[]` with confidences | smart layer | +| MAP estimate | `micro_hint` | smart layer | +| Suppression | `avoid[]` — hard-absence evidence zeroes mass | smart layer | +| Ambiguity | `tool_entropy.h` — flat distribution ⇒ "gather info", not argmax | smart layer | +| New label | `report_outcome` binds success/failure to navigation_id | driver | + +Design rules that fall out (all shipped in the smart layer as of 0.0.20-dev): + +1. **Calibration beats correctness.** A hint that fires when it shouldn't + trains agents to ignore hints. Every advisory branch gates on positive + evidence (a table is data only with ≥8 `` cells; a layout table on a + docs page must not route to `extract_table`). +2. **Negative advice saves more than positive advice.** Each avoided call is + a full round-trip plus failed-parse cost. `avoid[]` emits only on hard + structural absence (no JSON scripts, no tables, no forms), never + speculation. +3. **Suppress argmax under ambiguity.** When the next_tools distribution is + flat (normalized entropy > 0.85), `micro_hint` is withheld and the bundle + says so — argmax over noise is how hints lose trust. +4. **Phase B closes the loop.** `report_outcome` labels feed a Beta-Bernoulli + per `(page_shape_bucket × tool)` pair; hand-tuned likelihood weights + become learned ones. Same sample-efficiency argument as Appendix B: two + inspectable parameters per cell, no RL. + +The name/description prior is why tool naming is a probabilistic decision, +not an aesthetic one (`open` routes intent better than `navigate_auto`) — +but priors are the *smallest* lever we control. The evidence, suppression, +and calibration layers are where the accuracy actually comes from. diff --git a/python/pyproject.toml b/python/pyproject.toml index 8a62007..1588fe4 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -46,9 +46,13 @@ solver = ["unchainedsky-cli"] # `pip install pyunbrowser` installs the primary `unbrowser` script and a # `pyunbrowser` compatibility alias. Both dispatch to the same CLI; the alias # makes the Registry-derived `uvx pyunbrowser --mcp` command executable. +# `unbrowser-smart` is the stdio MCP server for the minimal-3 smart surface +# (search/open/help) — configured in MCP hosts as `"command": ["unbrowser-smart"]`. [project.scripts] unbrowser = "unbrowser._cli:main" pyunbrowser = "unbrowser._cli:main" +unbrowser-smart = "unbrowser.smart_mcp:main" +pyunbrowser-smart = "unbrowser.smart_mcp:main" # The wheel ships the platform-specific native binary inside unbrowser/_bin/. # CI builds the binary first (cargo build --release for each target), copies diff --git a/python/unbrowser/__init__.py b/python/unbrowser/__init__.py index 5db93c2..6e03b91 100644 --- a/python/unbrowser/__init__.py +++ b/python/unbrowser/__init__.py @@ -15,6 +15,12 @@ For the `extract` / auto-strategy command, watchdog-bounded `exec_scripts`, the cookie handoff for bot-walled sites, and the BlockMap shape: see the project README at https://github.com/protostatis/unbrowser. + +Smart wrapper (minimal 3-tool progressive discovery): `SmartClient` adds +`search(query)` (Brave → DDG fallback), `navigate_auto(url, goal=)` +(open + bounded discover/cards + `escalation`/`micro_hint`/`next_tools`), +and `help(topic)` (grouped catalog). Runs as an MCP server via the +`unbrowser-smart` console script or `python -m unbrowser.smart_mcp`. """ from __future__ import annotations @@ -247,6 +253,10 @@ def search(self, query: str, engine: str = "ddg") -> dict: bing — Bing search. Tracker links in results are auto-decoded on click (the binary detects bing.com/ck/a?u=... URLs and follows to the real destination). + brave — Brave Search HTML via unbrowser. Prefer the SmartClient + wrapper (``from unbrowser.smart import SmartClient``) for + a parsed ``[{title,url,snippet}]`` result; this base + method returns the raw navigate result for brave as well. Google is intentionally NOT supported via the cheap path — Google's search page returns ~no useful HTML without JS, so it would silently @@ -260,9 +270,11 @@ def search(self, query: str, engine: str = "ddg") -> dict: url = "https://duckduckgo.com/html/?q=" + quote_plus(query) elif engine == "bing": url = "https://www.bing.com/search?q=" + quote_plus(query) + elif engine == "brave": + url = "https://search.brave.com/search?q=" + quote_plus(query) + "&source=web" else: raise UnbrowserError( - f"unknown search engine '{engine}'. Supported: ddg, bing. " + f"unknown search engine '{engine}'. Supported: ddg, bing, brave. " "Google is intentionally unsupported via the cheap path." ) return self.navigate(url) @@ -439,3 +451,20 @@ def navigate(url: str, exec_scripts: bool = False, shim_mode: str | None = None) """ with Client(shim_mode=shim_mode) as ub: return ub.navigate(url, exec_scripts=exec_scripts) + + +# Lazy re-export of SmartClient (guard against circular import: smart.py imports +# Client/UnbrowserError from this module, so the import has to come last). +try: # pragma: no cover - import guard + from .smart import SmartClient as SmartClient + + __all__ = [ + "Client", + "UnbrowserError", + "SmartClient", + "find_binary", + "navigate", + "__version__", + ] +except ImportError: # pragma: no cover - smart.py missing (source checkout/old wheel) + pass diff --git a/python/unbrowser/_cli.py b/python/unbrowser/_cli.py index e6005ea..54be87e 100644 --- a/python/unbrowser/_cli.py +++ b/python/unbrowser/_cli.py @@ -5,12 +5,17 @@ agents and MCP hosts can use directly (e.g. `command: "unbrowser"` in .mcp.json). -The wrapper keeps the native binary as the execution engine and exposes a -useful `--help` surface. Invocations are passed through to the binary. +The wrapper keeps the native binary as the execution engine. Help follows +progressive-disclosure conventions (clig.dev): `--help` shows the core path +plus grouped tool families; `unbrowser help ` drills into any family +or tool; unknown commands get did-you-mean suggestions on stderr with +exit code 2. """ from __future__ import annotations +import difflib +import json import os import subprocess import sys @@ -19,42 +24,137 @@ from . import find_binary +# Grouped tool families — mirrors HELP_CATALOG in unbrowser/smart.py and the +# Rust MCP surface. Kept as plain data so `--help` renders without importing +# the smart layer. +TOOL_FAMILIES: dict[str, list[str]] = { + "reading": ["text", "text_main", "text_clean", "blockmap", "body"], + "query": ["query", "query_debug", "query_text", "find_text", "text_around"], + "extraction": ["extract", "extract_table", "extract_list", "extract_cards", "table_to_json"], + "discovery": ["discover", "route_discover", "page_model", "network_extract", "network_stores"], + "interaction": ["click", "type", "submit", "activate", "settle", "eval"], + "session": ["cookies_set", "cookies_get", "cookies_clear", "report_outcome"], +} + +_KNOWN_COMMANDS = [ + "navigate", "search", "open", "help", "exec", "session", + "router", "cookie-service", "policy-check", "--mcp", "--version", + "--list-profiles", "--prefit-info", +] + + def _usage() -> None: + fams = "\n".join(f" {fam:<12} {' '.join(tools)}" for fam, tools in TOOL_FAMILIES.items()) print( - """unbrowser + f"""unbrowser — web access for LLM agents. One static binary. No Chrome. -Usage: - unbrowser session start [--id ] [--profile ] [--policy=blocklist] [--shims stable|enhanced] - unbrowser session exec [--pretty] [params-json | shorthand args] +START HERE + unbrowser navigate [--exec-scripts] fetch a page -> low-token BlockMap + unbrowser search "" [--count N] web search (Brave->DDG) -> [{{title,url,snippet}}] + unbrowser open [--goal G] fetch + auto-discover + next-step hints + unbrowser --mcp MCP server mode for agent hosts + +MULTI-STEP SESSIONS (cookies + last page persist) + unbrowser session start [--id ] [--profile ] [--policy=blocklist] unbrowser exec [--pretty] [params-json | shorthand args] - unbrowser session stop - unbrowser session list - unbrowser session prune - unbrowser navigate [--exec-scripts] [--json] [--events] [--shims stable|enhanced] - unbrowser router [--cookie-service ] [--allow-remote-cookie-service] [--no-auto-cookie-service] - unbrowser cookie-service [--headless|--no-headless] [--port ] [--allow-host ] [--allow-remote-bind] + unbrowser session stop | session list | session prune + +TOOLS — call via `unbrowser exec '{{...}}'`, or over MCP +{fams} + + unbrowser help details + examples (e.g. `unbrowser help extraction`) + +MORE + unbrowser router bot-wall cookie handoff via local Chrome + unbrowser cookie-service [--headless] local solver service (needs [solver] extra) unbrowser policy-check [...] - unbrowser --list-profiles - unbrowser --prefit-info - unbrowser [--profile ] [--policy=blocklist] [--shims stable|enhanced] [--mcp] - unbrowser --version - -Examples: - unbrowser session start --id demo - unbrowser exec demo navigate https://news.ycombinator.com - unbrowser exec --pretty demo blockmap - unbrowser session stop demo - unbrowser navigate https://news.ycombinator.com --json - unbrowser cookie-service --headless --profile unbrowser-cookie-service - unbrowser router https://example.com/protected - unbrowser policy-check https://www.bbc.com/news - printf '{\"id\":1,\"method\":\"navigate\",\"params\":{\"url\":\"https://news.ycombinator.com\"}}\n' | unbrowser - -`navigate` delegates to the native binary; output is always the binary's JSON. + unbrowser --list-profiles | --prefit-info | --version + +Every result carries routing hints: micro_hint (the next concrete step), +next_tools (ranked candidates), avoid (tools with nothing to act on). """ ) +def _help_topic(topic: str | None) -> int: + """Render the grouped catalog, one family, or one tool. Exit 0.""" + try: + from .smart import HELP_CATALOG + except ImportError: + print("help catalog unavailable in this install", file=sys.stderr) + return 1 + if not topic: + for fam, tools in HELP_CATALOG.items(): + print(f"{fam}:") + for name, info in tools.items(): + print(f" {name:<16} {info.get('when', '')}") + print("\nDrill in: unbrowser help e.g. unbrowser help extract_table") + return 0 + t = topic.lower() + for fam, tools in HELP_CATALOG.items(): + if t == fam: + print(f"{fam}:") + for name, info in tools.items(): + print(f"\n {name}\n {info.get('when', '')}") + if info.get("example"): + print(f" e.g. {info['example']}") + return 0 + if t in tools: + info = tools[t] + print(f"{t} ({fam})\n {info.get('when', '')}") + if info.get("example"): + print(f" e.g. {info['example']}") + return 0 + # fuzzy fallback + matches = difflib.get_close_matches(t, [n for f_ in HELP_CATALOG.values() for n in f_], n=3) + if matches: + print(f"unknown topic '{topic}'. Did you mean: {', '.join(matches)}?") + else: + print(f"unknown topic '{topic}'") + return 1 + + +def _suggest_and_exit(bad: str) -> None: + matches = difflib.get_close_matches(bad, _KNOWN_COMMANDS + [n for f_ in TOOL_FAMILIES.values() for n in f_], n=3) + hint = f" Did you mean: {', '.join(matches)}?" if matches else "" + print(f"unbrowser: unknown command '{bad}'.{hint}\nRun `unbrowser --help` to see what's available.", file=sys.stderr) + raise SystemExit(2) + + +def _cmd_search(args: list[str]) -> None: + count = 5 + if "--count" in args: + i = args.index("--count") + count = int(args[i + 1]) + del args[i : i + 2] + query = " ".join(a for a in args if not a.startswith("-")) + if not query: + print("usage: unbrowser search \"\" [--count N]", file=sys.stderr) + raise SystemExit(2) + from .smart import SmartClient + + with SmartClient() as ub: + hits = ub.search(query, count=count) + print(json.dumps(hits, indent=2)) + + +def _cmd_open(args: list[str]) -> None: + goal = None + if "--goal" in args: + i = args.index("--goal") + goal = args[i + 1] + del args[i : i + 2] + url = next((a for a in args if not a.startswith("-")), None) + if not url: + print("usage: unbrowser open [--goal G]", file=sys.stderr) + raise SystemExit(2) + from .smart import SmartClient + + with SmartClient() as ub: + bundle = ub.navigate_auto(url, goal=goal) + print(json.dumps(bundle, indent=2)) + + def _is_help_flag(arg: str) -> bool: return arg in {"-h", "--help"} @@ -102,6 +202,17 @@ def main() -> None: _usage() return + if argv[0] == "help": + raise SystemExit(_help_topic(argv[1] if len(argv) > 1 else None)) + + if argv[0] == "search": + _cmd_search(argv[1:]) + return + + if argv[0] == "open": + _cmd_open(argv[1:]) + return + if argv[0] == "navigate": _navigate(argv[1:]) return @@ -114,9 +225,12 @@ def main() -> None: _router(argv[1:]) return - binary = find_binary() - # Preserve the native binary behavior for every other command. - os.execv(binary, ["unbrowser", *argv]) + # Pass through known binary commands and flags; anything else gets a + # did-you-mean instead of a cryptic binary error. + if argv[0].startswith("-") or argv[0] in {"session", "exec", "policy-check"}: + binary = find_binary() + os.execv(binary, ["unbrowser", *argv]) + _suggest_and_exit(argv[0]) if __name__ == "__main__": diff --git a/python/unbrowser/smart.py b/python/unbrowser/smart.py new file mode 100644 index 0000000..fdf092d --- /dev/null +++ b/python/unbrowser/smart.py @@ -0,0 +1,784 @@ +"""Smart wrapper: infers search vs navigate. + +Two-entry inference: + 1. search: -> Brave HTML search via unbrowser (fallback to Brave API if BRAVE_API_KEY set) + 2. navigate: -> navigate + auto discover (discover + cards + page_model) + +Also exposes `run(task)` which infers from raw string: URL-looking -> navigate_auto, +otherwise -> brave search. + +Example: + from unbrowser.smart import SmartClient + + with SmartClient() as ub: + # entry 1: search + hits = ub.search("Pixel 11 review", engine="brave") # -> [{title,url,snippet,display_url}] + # entry 2: navigate + auto discover + bundle = ub.navigate_auto(hits[0]["url"], goal="Pixel 11 review") + # bundle = {navigate, blockmap, discover, cards, page_model, extract} + + # or infer + bundle = ub.run("https://www.engadget.com/2240042/google-pixel-11-review/") + hits = ub.run("Pixel 11 review") + +Requires UNBROWSER_BIN to point to a recent build that exposes discover/extract_cards/page_model +for the auto bundle to be rich; otherwise it gracefully degrades. +""" + +from __future__ import annotations + +import json +import os +import re +from typing import Any +from urllib.parse import quote_plus, urlparse +from urllib.request import Request, urlopen + +from . import Client, UnbrowserError + +BRAVE_SEARCH_HTML = "https://search.brave.com/search?q={q}&source=web" +BRAVE_API_ENDPOINT = "https://api.search.brave.com/res/v1/web/search" + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + +_URL_RE = re.compile(r"^https?://", re.I) + + +def is_url(s: str) -> bool: + s = s.strip() + if _URL_RE.match(s): + return True + # bare domain like engadget.com/foo -> treat as url + try: + p = urlparse(s if "://" in s else "https://" + s) + return bool(p.netloc and "." in p.netloc and " " not in s) + except Exception: + return False + + +def _norm_url(u: str, base: str | None = None) -> str: + u = u.strip() + if not u: + return u + # absolute URL + if _URL_RE.match(u): + return u + # protocol-relative //cdn.example.com/foo + if u.startswith("//"): + return "https:" + u + # root-relative or relative: join with base (last_url) if available + if base and (u.startswith("/") or not _URL_RE.match(u)): + try: + from urllib.parse import urljoin + + return urljoin(base, u) + except Exception: + pass + return "https://" + u.lstrip("/") + + +# --------------------------------------------------------------------------- +# Brave HTML extraction +# --------------------------------------------------------------------------- + +_BRAVE_SNIPPET_JS = r""" +(function(q){ + const snippets = document.querySelectorAll('.snippet[data-type="web"]'); + const out = []; + const limit = q.limit || 10; + for (let i=0;i 40 && !t.startsWith("http")){ + snippet = t.slice(0,500); + break; + } + } + if (!snippet){ + // fallback: whole snippet text minus title + const all = el.textContent.replace(/\s+/g,' ').trim(); + snippet = all.slice(title.length, title.length+400).trim(); + } + const cite = el.querySelector('cite'); + const display_url = cite ? cite.textContent.trim().replace(/\s+/g,' ').slice(0,120) : href; + // filter out non-web snippets that slipped through + if (!href || href.includes("search.brave.com")) continue; + out.push({title, url: href, snippet, display_url}); + } + return JSON.stringify(out); +}) +""" + + +def _ddg_html_extract(client: Client, count: int = 10) -> list[dict]: + """Fallback DDG HTML extraction (used when Brave 429s).""" + seen: set[str] = set() + out: list[dict] = [] + for a in client.query("a.result__a"): + href = (a.get("attrs", {}).get("href") or "").strip() + # DDG wraps as //duckduckgo.com/l/?uddg=... + if "uddg=" in href: + try: + from urllib.parse import parse_qs, urlparse as _up + href = parse_qs(_up(href).query).get("uddg", [href])[0] + except Exception: + pass + if href.startswith("//"): + href = "https:" + href + if not href.startswith("http") or "duckduckgo.com" in href: + continue + if href in seen: + continue + seen.add(href) + txt = (a.get("text") or "").strip()[:200] + if len(txt) < 8: + continue + out.append({"title": txt, "url": href, "snippet": "", "display_url": href}) + if len(out) >= count: + break + # generic fallback if selector missed (DDG varies) + if not out: + for a in client.query("a"): + href = (a.get("attrs", {}).get("href") or "").strip() + if not href.startswith("http") or "duckduckgo.com" in href: + continue + if href in seen: + continue + if len((a.get("text") or "").strip()) < 10: + continue + seen.add(href) + out.append({"title": (a.get("text") or "").strip()[:200], "url": href, "snippet": "", "display_url": href}) + if len(out) >= count: + break + return out + + +def _brave_html_extract(client: Client, query: str, count: int = 10) -> list[dict]: + url = BRAVE_SEARCH_HTML.format(q=quote_plus(query)) + nav = client.navigate(url) + if nav.get("status", 200) >= 400: + # 429/503 rate-limit on Brave HTML — fallback to DDG HTML so the + # two-entry flow still completes (same parsed shape). + if nav.get("status") in (429, 503): + try: + ddg_url = "https://duckduckgo.com/html/?q=" + quote_plus(query) + dnav = client.navigate(ddg_url) + if dnav.get("status", 200) < 400: + return _ddg_html_extract(client, count=count) + except Exception: + pass + raise UnbrowserError(f"brave search navigate failed: {nav.get('status')} {url}") + # Prefer structured JS extraction; fallback to generic link scan + try: + raw = client.eval(f"({_BRAVE_SNIPPET_JS})({{limit:{int(count)}}})") + # eval returns either a JSON string (JS stringify) or a list (raw JS array). + if isinstance(raw, str): + try: + items = json.loads(raw) + except Exception: + items = [] + elif isinstance(raw, list): + items = raw + else: + items = [] + if items: + return items[:count] + except Exception: + pass + # fallback: take brave links that are https and not search.brave.com, dedup + seen = set() + out: list[dict] = [] + for a in client.query("a"): + href = (a.get("attrs", {}).get("href") or "").strip() + if not href.startswith("http") or "search.brave.com" in href: + continue + if href in seen: + continue + seen.add(href) + txt = (a.get("text") or "").strip().replace("\n", " ")[:200] + if len(txt) < 10: + continue + out.append({"title": txt, "url": href, "snippet": "", "display_url": href}) + if len(out) >= count: + break + return out + + +def _brave_api_extract(query: str, count: int = 10) -> list[dict] | None: + key = os.environ.get("BRAVE_API_KEY") or os.environ.get("BRAVE_SEARCH_API_KEY") + if not key: + return None + import time + from urllib.error import HTTPError + + for attempt in range(3): + try: + req = Request( + f"{BRAVE_API_ENDPOINT}?q={quote_plus(query)}&count={count}", + headers={"Accept": "application/json", "X-Subscription-Token": key}, + ) + with urlopen(req, timeout=10) as resp: + data = json.loads(resp.read().decode("utf-8")) + results = data.get("web", {}).get("results", []) or data.get("results", []) + out = [] + for r in results[:count]: + out.append({ + "title": r.get("title", "")[:300], + "url": r.get("url", ""), + "snippet": (r.get("description") or r.get("snippet") or "")[:500], + "display_url": r.get("url", "")[:120], + }) + return out + except HTTPError as e: + # retry 429/5xx with backoff, honor Retry-After + if e.code in (429, 500, 502, 503, 504) and attempt < 2: + retry_after = e.headers.get("Retry-After") + try: + wait = int(retry_after) if retry_after else (1 << attempt) + except Exception: + wait = 1 << attempt + time.sleep(min(wait, 8)) + continue + return None + except Exception: + # network/auth/malformed: fall through to HTML path; diagnostic is visible + # because caller will try _brave_html_extract next, so no silent loss of signal + return None + return None + + +# --------------------------------------------------------------------------- +# Help catalog (grouped 32) for progressive discovery +# --------------------------------------------------------------------------- + +HELP_CATALOG: dict[str, Any] = { + "core": { + "search": {"when": "find URLs for a query", "example": "ub.search('Pixel 11 review', count=5)"}, + "open": {"when": "fetch URL + auto discover routes/cards", "example": "ub.navigate_auto('https://example.com', goal='Pixel 11')"}, + "extract": {"when": "auto-strategy structured data (JSON-LD, Next, Nuxt, OG)", "example": "ub.call('extract')"}, + "help": {"when": "discover full 32", "example": "ub.help() or ub.help('query')"}, + }, + "query_text": { + "query": {"when": "CSS → [{ref, tag, attrs, text}] (stable refs for click/type)"}, + "query_debug": {"when": "diagnose selector miss (hints: selector_miss, thin_shell, embedded_json)"}, + "query_text": {"when": "find by visible text (hashed React classes, anchor-promotion)"}, + "find_text": {"when": "ranked text matches with before/after context"}, + "text_around": {"when": "400-char window around ref or text match"}, + }, + "reading": { + "text": {"when": "textContent of first match (default body) — hatnote trap"}, + "text_main": {"when": "main content (excludes header/nav/footer)"}, + "text_clean": {"when": "chrome+JSON stripped, whitespace-collapsed"}, + "blockmap": {"when": "recompute BlockMap after DOM mutation"}, + "body": {"when": "raw HTML fallback (100KB+)"}, + }, + "discovery": { + "page_model": {"when": "semantic objects (search_form, article_card, product_card...)"}, + "route_discover": {"when": "ranked links/forms + inferred query URLs"}, + "discover": {"when": "merged DOM+inferred+network graph with provenance (use this before guessing URLs)"}, + "network_extract": {"when": "parse captured JSON/API into semantic objects"}, + "network_stores": {"when": "ranked fetch/XHR captures (256KB preview)"}, + }, + "extraction": { + "extract": {"when": "auto-strategy JSON-LD → Next → Nuxt → OG → microdata"}, + "extract_cards": {"when": "repeated cards → [{title, price, url, snippet}]"}, + "extract_list": {"when": "explicit {item_selector, fields: {name: 'sel @attr'}}"}, + "extract_table": {"when": "table → {headers, rows}"}, + "table_to_json": {"when": "alias for extract_table (default table)"}, + }, + "interaction": { + "click": {"when": "dispatch click on e:NN (auto-follows )"}, + "activate": {"when": "probe click → {navigated, dom_changed, network_changed, no_effect}"}, + "type": {"when": "set input value + input/change events"}, + "submit": {"when": "form → GET or x-www-form-urlencoded POST"}, + "settle": {"when": "drain microtasks + timers"}, + "eval": {"when": "QuickJS JS (bounded, use for price selectors)"}, + }, + "session": { + "cookies_set": {"when": "replay clearance cookie (_px3 etc) from real Chrome"}, + "cookies_get": {"when": "export jar"}, + "cookies_clear": {"when": "clear jar"}, + "report_outcome": {"when": "bind success/failure to navigation_id for policy learning"}, + "network_stores_clear": {"when": "drop captures"}, + }, +} + + +def _help_catalog(topic: str | None = None) -> dict: + if topic: + t = topic.lower() + for group, tools in HELP_CATALOG.items(): + if t == group or t in tools: + return {group: tools} if t == group else {t: tools.get(t, {})} + # fuzzy: search in descriptions + out: dict[str, Any] = {} + for group, tools in HELP_CATALOG.items(): + for name, info in tools.items(): + if t in name or t in str(info.get("when", "")).lower(): + out.setdefault(group, {})[name] = info + return out or HELP_CATALOG + return HELP_CATALOG + + +def _next_tools_from_bundle(bundle: dict) -> list[dict]: + """Build next_tools from navigate signals + tool_likelihoods.""" + nxt: list[dict] = [] + raw = bundle.get("raw") or {} + recs = raw.get("tool_recommendations") or [] + likes = raw.get("tool_likelihoods") or {} + for name in recs[:6]: + nxt.append({"tool": name, "when": HELP_CATALOG.get("core", {}).get(name, {}).get("when") or "recommended", "confidence": float(likes.get(name, 0.7))}) + if not nxt: + bm = bundle.get("blockmap") or {} + density = bm.get("density") or {} + if density.get("likely_js_filled"): + nxt.append({"tool": "eval", "when": "JS-gated content, inspect script JSON", "confidence": 0.85}) + if (bundle.get("cards") is not None and len(bundle.get("cards") or []) == 0): + nxt.append({"tool": "extract_list", "when": "cards missed, try explicit fields", "confidence": 0.6}) + return nxt[:6] + + +def _escalation_for_bundle(bundle: dict) -> dict | None: + """Portable escalation: stable reason + evidence + severity + retryable. + + Rust emits facts (challenge, status, density, scripts, extract); Python maps + to host actions. This keeps reason codes stable and avoids phantom tools. + """ + status = bundle.get("status") + challenge = bundle.get("challenge") + bm = bundle.get("blockmap") or {} + density = bm.get("density") or {} + extract = bundle.get("extract") or {} + scripts = bundle.get("scripts") or {} + + # 1. challenge / bot wall — portable fact from Rust + if challenge: + provider = challenge.get("provider") or challenge.get("vendor") or "unknown" + return { + "reason": "challenge", + "category": "external_capability", + "confidence": float(challenge.get("confidence", 0.9)) if isinstance(challenge.get("confidence"), (int, float)) else 0.9, + "severity": "high", + "retryable": False, + "evidence": {"provider": provider, "status": status, "clearance_cookie": challenge.get("clearance_cookie")}, + "hint": "Continue using session state from a user-authorized browser, where permitted. Acquire a clearance cookie in real Chrome for this origin and replay via cookies_set.", + "options": [ + {"action": "replay_clearance_cookie", "tool": "cookies_set", "params": {"cookies": [{"name": challenge.get("clearance_cookie") or "_px3", "value": "", "domain": urlparse(bundle.get("url") or "").hostname or "example.com"}]}, "requires_user_confirmation": True}, + {"action": "external_action", "external_action": "chrome_escalation", "reason": provider, "params": {"reason": provider}}, + {"action": "try_help", "tool": "help", "params": {"topic": "session"}}, + ], + "next_tools": [{"tool": "cookies_set", "when": "replay user-authorized cookie", "confidence": 0.9}], + } + # 2. http errors — split per advisor + if isinstance(status, int) and status >= 400: + if status in (401, 403): + return {"reason": "auth_required", "category": "external_capability", "confidence": 0.95, "severity": "high", "retryable": False, "evidence": {"status": status}, "hint": f"HTTP {status} auth required or blocked. Acquire session state from a user-authorized browser if permitted.", "options": [{"action": "external_action", "external_action": "chrome_escalation", "reason": "auth"}, {"action": "try_help", "tool": "help", "params": {"topic": "session"}}], "next_tools": [{"tool": "help", "when": "session", "confidence": 0.7}]} + if status == 404: + return {"reason": "not_found", "category": "terminal", "confidence": 0.95, "severity": "low", "retryable": False, "evidence": {"status": status}, "hint": f"HTTP {status} not found — terminal, do not retry.", "options": [{"action": "try_help", "tool": "help", "params": {"topic": "discovery"}}], "next_tools": []} + if status == 429: + return {"reason": "rate_limited", "category": "retry", "confidence": 0.9, "severity": "medium", "retryable": True, "evidence": {"status": status}, "hint": f"HTTP {status} rate-limited. Back off and retry, or switch search provider (Brave→DDG fallback already handles this for search).", "options": [{"action": "retry_backoff", "tool": "open", "params": {"url": bundle.get("url")}}, {"action": "try_help", "tool": "help", "params": {"topic": "session"}}], "next_tools": [{"tool": "open", "when": "retry with backoff", "confidence": 0.6}]} + if status >= 500: + return {"reason": "server_error", "category": "retry", "confidence": 0.8, "severity": "medium", "retryable": True, "evidence": {"status": status}, "hint": f"HTTP {status} server error — retryable.", "options": [{"action": "retry", "tool": "open", "params": {"url": bundle.get("url")}}], "next_tools": [{"tool": "open", "when": "retry", "confidence": 0.6}]} + return {"reason": "http_error", "category": "retry", "confidence": 0.8, "severity": "medium", "retryable": True, "evidence": {"status": status}, "hint": f"HTTP {status} — check challenge field and retry.", "options": [{"action": "try_help", "tool": "help", "params": {"topic": "session"}}], "next_tools": [{"tool": "help", "when": "session", "confidence": 0.7}]} + # 3. timeout — enrichment bounded + if bundle.get("discover_timeout") or bundle.get("cards_timeout") or bundle.get("page_model_timeout"): + return { + "reason": "timeout", + "category": "retry", + "confidence": 0.7, + "severity": "medium", + "retryable": True, + "evidence": {"discover_timeout": bool(bundle.get("discover_timeout")), "cards_timeout": bool(bundle.get("cards_timeout"))}, + "hint": "Enrichment timed out (heavy DOM). Retry with smaller limits or skip discover.", + "options": [ + {"action": "retry_smaller", "tool": "open", "params": {"url": bundle.get("url"), "discover_limit": 3, "cards_limit": 3}}, + {"action": "skip_discover", "tool": "extract_cards", "params": {"limit": 5}}, + {"action": "try_help", "tool": "help", "params": {"topic": "extraction"}}, + ], + "next_tools": [{"tool": "extract_cards", "when": "cards only", "confidence": 0.7}], + } + # 4. unsupported JS feature / thin shell — stable codes + if density.get("thin_shell"): + return {"reason": "thin_shell", "category": "continuation", "confidence": 0.7, "severity": "medium", "retryable": True, "evidence": {"thin_shell": True}, "hint": "SSR shell with little content — try exec_scripts or check embedded JSON.", "options": [{"action": "retry_exec_scripts", "tool": "open", "params": {"url": bundle.get("url"), "exec_scripts": True}}, {"action": "try_extract", "tool": "extract", "params": {}}], "next_tools": [{"tool": "extract", "when": "auto-strategy", "confidence": 0.7}]} + if density.get("likely_js_filled"): + return {"reason": "unsupported_js_feature", "category": "external_capability", "confidence": 0.85, "severity": "high", "retryable": False, "evidence": {"likely_js_filled": True, "script_errors": (scripts.get("errors") or [])[:2]}, "hint": "Client-rendered content requires JS features not supported by QuickJS (e.g. ES modules, import maps, WASM). Use extract for embedded JSON or escalate to real Chrome, where permitted.", "options": [{"action": "extract_alternate", "tool": "extract", "params": {"strategy": "json_in_script"}}, {"action": "external_action", "external_action": "chrome_escalation", "reason": "unsupported_js_feature"}], "next_tools": [{"tool": "extract", "when": "try json_in_script / nuxt_data", "confidence": 0.8}]} + # check ES-module specifically (REI case) even when likely_js_filled is false + errs = " ".join(str(e) for e in (scripts.get("errors") or [])) + if "export" in errs or "import" in errs: + return {"reason": "unsupported_js_feature", "category": "external_capability", "confidence": 0.85, "severity": "high", "retryable": False, "evidence": {"script_errors": (scripts.get("errors") or [])[:2]}, "hint": "QuickJS cannot run ES-module bundles (export/import). Extract embedded data or use a real browser, where permitted.", "options": [{"action": "extract_alternate", "tool": "extract", "params": {"strategy": "nuxt_data"}}, {"action": "external_action", "external_action": "chrome_escalation", "reason": "ES-module"}], "next_tools": [{"tool": "extract", "when": "nuxt_data", "confidence": 0.8}]} + # 5. partial_result — supersedes old extract_truncated (informational, not escalation) + if extract.get("primary_truncated"): + pt = extract["primary_truncated"] + return { + "reason": "partial_result", + "category": "continuation", + "confidence": 0.8, + "severity": "low", + "retryable": True, + "evidence": {"strategy": pt.get("strategy"), "size_bytes": pt.get("size_bytes")}, + "hint": f"Primary {pt.get('strategy')} {pt.get('size_bytes')} bytes exceeds inline cap; call extract(strategy=\"{pt.get('strategy')}\") for full.", + "options": [{"action": "fetch_full_extract", "tool": "extract", "params": {"strategy": pt.get("strategy")}}, {"action": "try_help", "tool": "help", "params": {"topic": "extraction"}}], + "next_tools": [{"tool": "extract", "when": f"strategy={pt.get('strategy')}", "confidence": 0.9}], + } + # 6. cards miss + cards = bundle.get("cards") + if isinstance(cards, list) and len(cards) == 0 and (density.get("li", {}) or {}).get("total", 0) > 20: + return {"reason": "cards_miss", "category": "continuation", "confidence": 0.6, "severity": "low", "retryable": True, "evidence": {"li_total": density.get("li", {}).get("total")}, "hint": "extract_cards returned 0 but DOM has many list items — try explicit selectors.", "options": [{"action": "try_extract_list", "tool": "extract_list", "params": {"item_selector": "article", "fields": {"title": "h3"}}}, {"action": "try_page_model", "tool": "page_model", "params": {}}], "next_tools": [{"tool": "extract_list", "when": "explicit fields", "confidence": 0.6}]} + return None + + +def _micro_hint_for_bundle(bundle: dict) -> dict | None: + """Derive a concrete selector/next-step from signals already in the bundle. + + When auto-discovery (cards/discover) comes up empty but the page has usable + DOM facts (tables, forms, li, json scripts), give the agent an immediate, + specific micro-step instead of forcing it to re-scan. Zero extra network calls. + + Calibration rule: a hint that fires when it shouldn't trains agents to + ignore hints. Each branch gates on positive evidence, not mere presence — + e.g. a table only counts as data when it has enough cells to be a + data grid, not a docs-page layout table. + """ + bm = bundle.get("blockmap") or {} + density = bm.get("density") or {} + discover = bundle.get("discover") or {} + headings = bm.get("headings") or [] + # 1. tables — gated on cell count: data grids have many s; layout + # tables (spec sheets, docs sidebars) have ~4 and extract_table on + # them wastes a call. + tables = density.get("tables") or {} + td = density.get("td") or {} + td_total = td.get("total", 0) if isinstance(td, dict) else 0 + if isinstance(tables, dict) and tables.get("total", 0) > 0 and td_total >= 8: + return { + "tool": "extract_table", + "selector": "table", + "reason": f"{tables.get('total')} table(s) with {td_total} cells — run extract_table('table') (or query 'table tbody tr' to preview rows).", + } + # 1b. table shells with no static cells = JS-injected grid (CNBC trap): + # extract_table would return empty rows; the data needs scripts or + # the network captures. + if isinstance(tables, dict) and tables.get("total", 0) > 0 and density.get("likely_js_filled"): + return { + "tool": "navigate", + "selector": None, + "reason": f"{tables.get('total')} table shell(s) but cells are JS-injected — re-navigate with exec_scripts=true, or check network_stores for the underlying data API.", + } + # 2. search forms (discover found them) + forms = discover.get("forms") or [] + if forms: + f = forms[0] + return { + "tool": "type", + "selector": f.get("controls", [{}])[0].get("ref") or "form input", + "reason": f"Search form found ('{f.get('label','')}') — type a query then submit to navigate it.", + } + # 3. embedded JSON beats prose scanning when present + if density.get("json_scripts", 0) > 0: + return { + "tool": "extract", + "selector": "json_ld", + "reason": f"{density.get('json_scripts')} JSON-bearing script tag(s); call extract() for structured data without selector guessing.", + } + # 4. headings but no cards: text is present, just narrow + if len(headings) > 0 and not (bundle.get("cards") or []): + h = headings[0].get("text", "") if isinstance(headings[0], dict) else "" + return { + "tool": "query_text", + "selector": "body", + "reason": f"No repeated cards, but headings exist (e.g. '{h[:40]}') — use text_main or query_text on the content root.", + } + # 5. lots of
  • but cards failed + li = density.get("li") or {} + if isinstance(li, dict) and li.get("total", 0) > 20 and not (bundle.get("cards") or []): + return { + "tool": "extract_list", + "selector": "li", + "reason": f"{li.get('total')} list items present but extract_cards found none — try extract_list with an explicit item_selector.", + } + return None + + +def _avoid_for_bundle(bundle: dict) -> list[dict]: + """Tools with ~zero posterior given hard absence signals. + + Negative advice saves more tokens than positive advice: each avoided + call is a full round-trip + failed-parse cost. Only emit when the + evidence is structural (element class absent from the DOM), never + speculative. + """ + bm = bundle.get("blockmap") or {} + density = bm.get("density") or {} + interactives = bm.get("interactives") or {} + avoid: list[dict] = [] + if not density.get("json_scripts", 0): + avoid.append({"tool": "extract", "reason": "no JSON-bearing