[Feat] Add confidence-gated browser deferral

docs/browser-front.md

@@ -158,6 +158,16 @@ browser-use/jev-ultrafast (MIT), whose observe-decide-act tick this policy follo
     `elapsed_ms` stops before the screenshot, and neither the outcome nor the release of the browser and the
     session depends on it. Tests: `tests/test_browse.py`, and `tests/system/test_final_screenshot.py` on a live
     @playwright/mcp.
+21. Confidence-gated deferral (`--defer-below P`, off by default) keeps the decision model as the first-stage
+    selector. When its operation confidence is strictly below `P` (a finite value from 0 through 1), the configured
+    chat model receives the current goal, page and only the currently offered operation/target identifiers. Its JSON
+    response is decoded back through that same action space; an invented operation or target, malformed JSON or a
+    provider failure ends the task `BLOCKED` rather than executing the first-stage action. `decision_ticks.json`
+    records `deferred`, the original choice/confidence, fallback choice, latency and separately-known fallback
+    tokens. The cutoff is a workload- and backend-specific routing knob, not a calibrated probability or a general
+    quality claim. Compare it with decision-only and chat-only on the same bounded task before reporting outcomes.
+    `python scripts/calibrate_browser_deferral.py runs/.../decision_ticks.json --rate 0.15` derives such a cutoff
+    from saved artifacts only; ties are not split, so its output reports the actual deferred count/rate as well.
 
 ## Rejected
 

s1a/browser/browse.py

@@ -6,6 +6,7 @@
 import argparse
 import asyncio
 import json
+import math
 import re
 import shutil
 import sys
@@ -246,8 +247,12 @@ async def browse(
     """
     if policy.rethink_on and model_name == "llm":
         raise RuntimeError("--rethink on needs a decision model; --model llm cannot use it")
+    if policy.defer_below is not None and model_name == "llm":
+        raise RuntimeError("--defer-below needs a decision model; --model llm cannot use it")
     if policy.rethink_on and decision_model is None:
         raise RuntimeError(f"--rethink on needs a decision model; --model {model_name} was given none")
+    if policy.defer_below is not None and decision_model is None:
+        raise RuntimeError(f"--defer-below needs a decision model; --model {model_name} was given none")
     logs_dir.mkdir(parents=True, exist_ok=True)
     calls: list[dict[str, Any]] = []
     counted = CountingModel(chat, calls)
@@ -354,6 +359,13 @@ def parser(spec: BrowserAgentSpec) -> argparse.ArgumentParser:
         default="off",
         help="decision models only: offer values extracted from the goal as a choice head",
     )
+    build.add_argument(
+        "--defer-below",
+        type=float,
+        default=None,
+        metavar="P",
+        help="decision models only: ask the chat fallback to choose from the current action space when confidence is below P (0 to 1; off by default)",
+    )
     build.add_argument(
         "--rethink",
         choices=("on", "off"),
@@ -392,12 +404,16 @@ def parser(spec: BrowserAgentSpec) -> argparse.ArgumentParser:
 
 def policy_from_args(args: argparse.Namespace) -> BrowserPolicy:
     """The policy from the parsed flags. ``RecoveryLimits`` rejects a non-finite or non-positive budget here."""
+    defer_below = args.defer_below
+    if defer_below is not None and (not math.isfinite(defer_below) or not 0 <= defer_below <= 1):
+        raise ValueError("--defer-below must be finite and between 0 and 1")
     return BrowserPolicy(
         prefetch_values=args.prefetch == "on",
         batch_actions=args.batch == "on",
         goal_value_cache=args.goal_values == "on",
         rethink_on=args.rethink == "on",
         recovery_limits=RecoveryLimits(max_attempts=args.rethink_attempts, timeout_s=float(args.rethink_timeout)),
+        defer_below=defer_below,
     )
 
 

s1a/browser/decision_model.py

@@ -21,7 +21,7 @@
 import statistics
 import time
 from collections.abc import Awaitable
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from typing import Any, AsyncIterator
 from urllib.parse import urlparse
 from uuid import uuid4
@@ -120,6 +120,8 @@ class BrowserPolicy:
     goal_value_cache: bool  # offer values extracted from the goal to Jev as a choice head
     rethink_on: bool = False  # bounded recovery: refresh plus plan under a per-task budget, never a wider tool set
     recovery_limits: RecoveryLimits = field(default_factory=RecoveryLimits)  # attempts and active seconds, never reset
+    # An opt-in confidence gate.  ``None`` retains the historic decision-model-only path.
+    defer_below: float | None = None
 
 
 @dataclass
@@ -193,6 +195,7 @@ def __init__(
         self._batch_actions = policy.batch_actions
         self._rethink_on = policy.rethink_on
         self._recovery_limits = policy.recovery_limits
+        self._defer_below = policy.defer_below
         self._max_steps = max_steps
         self._language = spec.language
         self._decision_model = decision_model
@@ -332,6 +335,20 @@ async def _decide_message(self, messages: Any) -> AssistantMessage:
                 logger.warning("[BrowserDecisionModel] decision failed: %s", exc, exc_info=True)
                 return await self._final(run, "BLOCKED", snapshot, f"decision failed: {exc}")
             move = interpret(decision, space)
+            deferred = self._defer_below is not None and move.confidence < self._defer_below
+            fallback_record: dict[str, Any] = {}
+            if self._defer_below is not None:
+                fallback_record = {
+                    "fallback_choice": None,
+                    "fallback_ms": None,
+                    "fallback_usage_known": False,
+                }
+            if deferred:
+                try:
+                    move, fallback_record = await self._defer_move(run, space, observation, move)
+                except Exception as exc:  # noqa: BLE001 - a failed escalation must never execute the weak choice
+                    logger.warning("[BrowserDecisionModel] deferred choice failed", exc_info=True)
+                    return await self._final(run, "BLOCKED", snapshot, f"deferred choice failed: {type(exc).__name__}")
             probabilities, candidates = top_probabilities(decision, move, space, TOP_CANDIDATES)
             run.tick += 1
             record = {
@@ -365,6 +382,18 @@ async def _decide_message(self, messages: Any) -> AssistantMessage:
                 "model": move.model,  # who answered: the response's model id, else the one the request named
                 **decision.provenance,
             }
+            if self._defer_below is not None:
+                record.update(
+                    {
+                        "deferred": deferred,
+                        "decision_choice": {
+                            "operation": decision.choice("operation").key,
+                            "target": self._decision_target_key(decision),
+                        },
+                        "decision_confidence": round(decision.choice("operation").confidence, 3),
+                        **fallback_record,
+                    }
+                )
             if self._decision_model.name == "jev-vl-served":
                 record["decision_raw"] = decision.raw
             if screenshot_ms is not None:  # an image-reading model: the capture is part of the step's cost
@@ -409,6 +438,76 @@ async def _decide_message(self, messages: Any) -> AssistantMessage:
                 return await self._final(run, "BLOCKED", snapshot, "")
             return await self._act(run, move, snapshot, record)
 
+    @staticmethod
+    def _decision_target_key(decision: Any) -> str | None:
+        """The first-stage target key, retained even when a fallback replaces its move."""
+        operation = decision.choice("operation").key
+        name = f"{operation.lower()}_target"
+        return decision.choice(name).key if name in decision.answers else None
+
+    async def _defer_move(
+        self, run: _Run, space: Any, observation: Observation, move: Move
+    ) -> tuple[Move, dict[str, Any]]:
+        """Ask the chat fallback to choose one *existing* action-space entry.
+
+        The model receives identifiers rather than browser tool arguments.  Its response is decoded back through
+        ``space`` here, so an invented operation or target is a terminal failure rather than an executable action.
+        """
+        choices = {
+            "operations": space.operations,
+            "targets": {operation: sorted(head) for operation, head in space.heads.items()},
+        }
+        state = observation.state if isinstance(observation.state, dict) else {}
+        prompt = {
+            "goal": run.goal,
+            "page": state.get("page", {}),
+            "choices": choices,
+            "response": {"operation": "one item from choices.operations", "target": "target id or null"},
+        }
+        started_at = time.perf_counter()
+        response = await self._fallback.invoke(
+            [
+                {
+                    "role": "system",
+                    "content": "Choose exactly one offered browser action. Return only JSON; never invent an operation or target.",
+                },
+                {"role": "user", "content": json.dumps(prompt, ensure_ascii=False)},
+            ]
+        )
+        elapsed_ms = round((time.perf_counter() - started_at) * 1000)
+        payload = json.loads(str(getattr(response, "content", "")))
+        operation = payload.get("operation")
+        target = payload.get("target")
+        if operation not in space.operations:
+            raise ValueError("fallback operation is not offered")
+        candidate = None
+        if operation in space.heads:
+            if not isinstance(target, str) or target not in space.heads[operation]:
+                raise ValueError("fallback target is not offered")
+            candidate = space.heads[operation][target]
+        elif target is not None:
+            raise ValueError("control operation must not name a target")
+        # OpenJiuwen chat replies expose token accounting as ``usage_metadata``.  CountingModel independently
+        # records the call for the run-level total, while this copies the same evidence into the deferred tick.
+        usage = getattr(response, "usage_metadata", None)
+        usage_known = usage is not None
+        return (
+            replace(move, operation=operation, candidate=candidate),
+            {
+                "fallback_choice": {"operation": operation, "target": target},
+                "fallback_ms": elapsed_ms,
+                "fallback_usage_known": usage_known,
+                **(
+                    {
+                        "fallback_input_tokens": int(getattr(usage, "input_tokens", 0)),
+                        "fallback_output_tokens": int(getattr(usage, "output_tokens", 0)),
+                    }
+                    if usage_known
+                    else {}
+                ),
+            },
+        )
+
     # -- bounded recovery ---------------------------------------------------
 
     @staticmethod
@@ -1121,7 +1220,7 @@ def report(self) -> dict[str, Any]:
         """
         run = self._run
         if run is None:
-            return {
+            report = {
                 "elapsed_ms": 0,
                 "decisions": 0,
                 "interactions": 0,
@@ -1147,8 +1246,11 @@ def report(self) -> dict[str, Any]:
                     "events": [],
                 },
             }
+            if self._defer_below is not None:
+                report.update({"deferred_steps": 0, "deferred_rate": 0.0, "fallback_ms": 0})
+            return report
         jev = [t["decision_ms"] for t in run.ticks]
-        return {
+        report = {
             "elapsed_ms": round((time.perf_counter() - run.started_at) * 1000) if run.started_at else 0,
             "decisions": len(jev),
             "interactions": len([h for h in run.history if h["kind"] != "wait"]),
@@ -1169,6 +1271,16 @@ def report(self) -> dict[str, Any]:
             "history": run.history,
             "recovery": self._recovery_summary(run),
         }
+        if self._defer_below is not None:
+            deferred = [t for t in run.ticks if t["deferred"]]
+            report.update(
+                {
+                    "deferred_steps": len(deferred),
+                    "deferred_rate": round(len(deferred) / len(jev), 3) if jev else 0.0,
+                    "fallback_ms": sum(int(t["fallback_ms"] or 0) for t in deferred),
+                }
+            )
+        return report
 
 
 def _probe_failed(snapshot: dict[str, Any]) -> bool:

scripts/calibrate_browser_deferral.py

@@ -0,0 +1,81 @@
+"""Choose empirical browser-deferral cutoffs from saved ``decision_ticks.json`` files.
+
+This is deliberately an offline helper: it only reads JSON artifacts and reports the threshold that defers the
+requested lowest-confidence fraction of the observed steps.  It does not claim that a decision confidence is a
+calibrated probability.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+from pathlib import Path
+from typing import Any
+
+
+def read_confidences(paths: list[Path]) -> list[float]:
+    """Read finite confidence values from complete artifacts or bare tick lists."""
+    values: list[float] = []
+    for path in paths:
+        payload: Any = json.loads(path.read_text(encoding="utf-8"))
+        ticks = payload.get("ticks", []) if isinstance(payload, dict) else payload
+        if not isinstance(ticks, list):
+            raise ValueError(f"{path}: expected a ticks list")
+        for tick in ticks:
+            confidence = tick.get("confidence") if isinstance(tick, dict) else None
+            if isinstance(confidence, (int, float)) and not isinstance(confidence, bool) and math.isfinite(confidence):
+                values.append(float(confidence))
+    if not values:
+        raise ValueError("no finite confidence values found")
+    return values
+
+
+def calibrate(confidences: list[float], rates: list[float]) -> list[dict[str, float | int]]:
+    """Return the strict-below threshold and actual count for each requested fraction.
+
+    Ties are never split: the reported actual rate may be lower than requested, which makes a command's routing
+    behaviour reproducible instead of depending on artifact order.
+    """
+    ordered = sorted(confidences)
+    if not ordered:
+        raise ValueError("at least one confidence is required")
+    rows: list[dict[str, float | int]] = []
+    for rate in rates:
+        if not 0 <= rate <= 1 or not math.isfinite(rate):
+            raise ValueError("rates must be finite values from 0 through 1")
+        requested = math.floor(len(ordered) * rate)
+        if requested == 0:
+            threshold = ordered[0]
+        else:
+            threshold = ordered[min(requested, len(ordered) - 1)]
+        count = sum(value < threshold for value in ordered)
+        rows.append(
+            {
+                "requested_rate": rate,
+                "defer_below": threshold,
+                "deferred_steps": count,
+                "total_steps": len(ordered),
+                "actual_rate": count / len(ordered),
+            }
+        )
+    return rows
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Calibrate offline confidence-gated browser deferral.")
+    parser.add_argument("artifacts", type=Path, nargs="+", help="saved decision_ticks.json file(s)")
+    parser.add_argument(
+        "--rate",
+        action="append",
+        type=float,
+        default=None,
+        help="requested deferral fraction, repeatable (default: 0.1, 0.15, 0.25)",
+    )
+    args = parser.parse_args()
+    rates = args.rate or [0.1, 0.15, 0.25]
+    print(json.dumps({"calibration": calibrate(read_confidences(args.artifacts), rates)}, indent=2))
+
+
+if __name__ == "__main__":
+    main()

tests/test_browse.py

@@ -39,6 +39,7 @@ def test_the_spec_and_the_policy_defaults_and_their_overrides(self) -> None:
         args = browse.parser(SPEC).parse_args(GOAL)
         self.assertEqual((args.max_steps, args.timeout, args.profile_out), (100, 180.0, None))
         self.assertEqual((args.batch, args.prefetch, args.goal_values), ("off", "on", "off"))
+        self.assertIsNone(args.defer_below)
         self.assertEqual(
             browse.policy_from_args(args),
             BrowserPolicy(prefetch_values=True, batch_actions=False, goal_value_cache=False),
@@ -52,6 +53,15 @@ def test_the_spec_and_the_policy_defaults_and_their_overrides(self) -> None:
             BrowserPolicy(prefetch_values=False, batch_actions=True, goal_value_cache=False),
         )
 
+    def test_defer_below_is_opt_in_and_rejects_invalid_thresholds(self) -> None:
+        args = browse.parser(SPEC).parse_args([*GOAL, "--defer-below", "0.5"])
+        self.assertEqual(browse.policy_from_args(args).defer_below, 0.5)
+        for value in ("-0.01", "1.01", "nan", "inf"):
+            with self.subTest(value=value):
+                args = browse.parser(SPEC).parse_args([*GOAL, "--defer-below", value])
+                with self.assertRaises(ValueError):
+                    browse.policy_from_args(args)
+
     def test_a_timeout_or_max_steps_at_or_below_zero_is_a_usage_error(self) -> None:
         for flags in (["--timeout", "0"], ["--timeout", "-1"], ["--max-steps", "0"], ["--max-steps", "-3"]):
             with self.assertRaises(SystemExit) as caught: