[eric] browser: code-side plan dispatch (OSW_PLAN_DISPATCH, default off) = one aux call compiles the task's mechanical prefix into verified steps, code executes them resolve-late+verified, the big model starts with that work done; irreversibles refused in the PARSER, fail-open everywhere; the turn-collapser that doesn't wait for model tool adoption (0/3 A/Bs), 4 tests

backend/apps/agents/browser/browser_agent.py

@@ -1257,6 +1257,21 @@ async def run_browser_agent(
                 # Clicked but the composer did NOT clear: the send is UNVERIFIED. Leave send_confirmed False so the loop can't shortcut to a "done" it never earned (r264 set it True here and the model then FALSELY claimed delivery). The model gets ONE truthful verify pass, never a blind resend.
                 task = f"{task}\n\n[{p_script['note']}]"
 
+    # Code-side plan dispatch (the turn-collapser that doesn't wait for the model to adopt a tool): one aux call compiles the task's mechanical prefix into verified steps, code executes them, and the big model starts with that work DONE. Fail-open: no plan/steps = today's loop untouched.
+    from backend.apps.agents.browser import browser_plan_dispatch
+    if (browser_plan_dispatch.plan_dispatch_enabled() and not app_mode and not done_called
+            and preloaded_perception and not cancel_event.is_set()):
+        try:
+            p_plan_note = await asyncio.wait_for(browser_plan_dispatch.run_plan_dispatch(
+                task, preloaded_perception, browser_id, tab_id,
+                load_settings(), get_api_type(model), execute_browser_tool,
+            ), timeout=45.0)
+        except Exception as p_pe:
+            logger.info(f"[plan-dispatch] outer skip ({p_pe})")
+            p_plan_note = ""
+        if p_plan_note:
+            task = f"{task}\n\n{p_plan_note}"
+
     try:
         for turn in range(MAX_TURNS):
             if done_called or cancel_event.is_set():

backend/apps/agents/browser/browser_plan_dispatch.py

@@ -0,0 +1,111 @@
+"""Code-side plan dispatch: the turn-collapser that does NOT depend on the model
+adopting a tool (it never does; 0/3 live A/Bs). ONE cheap aux call maps the task +
+live page state to a chain of verified steps; run_verified_step executes them in
+code (resolve-late, verify-effect, re-aim); the big model then starts with the
+mechanical work DONE instead of spending a ~4-6s turn per click.
+
+Safety mirrors the send-script: the aux may only emit click/fill on elements it
+names from the live list, anything irreversible-smelling is refused in code, and
+every step must VERIFY or the chain stops and hands off honestly. Fail-open
+everywhere: no aux, bad JSON, zero steps = the loop runs exactly as today.
+"""
+
+import asyncio
+import json
+import logging
+import os
+import re
+import time
+
+from backend.apps.agents.browser import browser_verified_step
+from backend.apps.agents.browser.browser_prestage import P_BLOCKED_CLICK_RE
+
+logger = logging.getLogger(__name__)
+
+P_MAX_STEPS = 4
+P_AUX_TIMEOUT_S = 10.0
+
+P_SYSTEM = (
+    "You compile the MECHANICAL prefix of a browser task into steps a dumb executor "
+    "runs. You see the task and the page's interactive elements. Emit ONLY steps you "
+    "are confident about, in order, as STRICT JSON (no prose): an array of\n"
+    '{"action":"click"|"fill","target":"<element name EXACTLY as listed>",'
+    '"role":"button"|"link"|"textbox"|"","text":"<for fill>",'
+    '"expect":"appeared:<text>"|"gone:<text>"|"url_changed"|"changed"|""}\n'
+    "Rules: target must be copied verbatim from a listed element name. STOP before "
+    "anything irreversible (send/submit/post/pay/delete/confirm/apply), before any "
+    "ambiguous choice, and before steps whose elements are not yet on the page. "
+    "0-4 steps; [] when nothing is safely mechanical."
+)
+
+
+def parse_plan(reply: str) -> list:
+    """Strict-ish JSON array extraction; anything malformed = [] (fail-open)."""
+    m = re.search(r"\[.*\]", (reply or "").strip(), re.S)
+    if not m:
+        return []
+    try:
+        raw = json.loads(m.group(0))
+    except Exception:
+        return []
+    steps = []
+    for r in raw[:P_MAX_STEPS]:
+        if not isinstance(r, dict):
+            continue
+        action = str(r.get("action") or "")
+        target = str(r.get("target") or "").strip()
+        if action not in ("click", "fill") or not target:
+            continue
+        if P_BLOCKED_CLICK_RE.search(target):
+            break  # irreversible-smelling: refuse this and everything after it
+        steps.append(browser_verified_step.VerifiedStep(
+            kind=action, target=target, role=str(r.get("role") or ""),
+            text=str(r.get("text") or ""), expect=str(r.get("expect") or "")))
+    return steps
+
+
+def plan_dispatch_enabled() -> bool:
+    return os.environ.get("OSW_PLAN_DISPATCH", "0") == "1"
+
+
+async def run_plan_dispatch(
+    task: str, state_text: str, browser_id: str, tab_id: str,
+    settings, primary_api, execute_tool,
+) -> str:
+    """Returns a handoff note describing verified-executed steps ('' = nothing ran).
+    Never raises; never acts irreversibly."""
+    t0 = time.monotonic()
+    try:
+        from backend.apps.settings.credentials import get_anthropic_client_for_model
+        from backend.apps.agents.providers.registry import resolve_aux_model
+        from backend.apps.agents.core.aux_llm import safe_resp_text
+
+        aux_model, _ = await resolve_aux_model(settings, preferred_tier="haiku", primary_api=primary_api)
+        client = get_anthropic_client_for_model(settings, aux_model)
+        reply = safe_resp_text(await asyncio.wait_for(
+            client.messages.create(
+                model=aux_model, max_tokens=400, temperature=0, system=P_SYSTEM,
+                messages=[{"role": "user", "content": (
+                    f"Task: {task[:1200]}\n\nInteractive elements:\n{state_text[:3500]}")}],
+            ), timeout=P_AUX_TIMEOUT_S))
+        steps = parse_plan(reply)
+        if not steps:
+            logger.info("[plan-dispatch] aux emitted no safe mechanical steps")
+            return ""
+        done: list[str] = []
+        for step in steps:
+            r = await browser_verified_step.run_verified_step(
+                step, browser_id, tab_id, execute_tool)
+            if not r["ok"]:
+                done.append(f"{step.kind} {step.target!r} FAILED ({r['note']}); stopped there")
+                break
+            done.append(f"{step.kind} {step.target!r} done+verified")
+        note = (
+            f"[Plan pre-executed and VERIFIED in code: {'; '.join(done)}. "
+            "Do NOT redo these; continue from the page's CURRENT state below.]"
+        )
+        logger.info(f"[plan-dispatch] {len(done)} step(s) in {int((time.monotonic() - t0) * 1000)}ms: {'; '.join(done)[:160]}")
+        return note
+    except Exception as e:
+        logger.info(f"[plan-dispatch] skipped ({e})")
+        return ""

backend/tests/test_browser_plan_dispatch.py

@@ -0,0 +1,29 @@
+"""Code-side plan dispatch: the aux's JSON plan is parsed fail-open and the
+irreversible wall holds in the PARSER (before anything could ever run)."""
+from backend.apps.agents.browser import browser_plan_dispatch as pd
+
+
+def test_parse_valid_plan():
+    steps = pd.parse_plan('[{"action":"click","target":"Options","role":"button"},'
+                          '{"action":"fill","target":"Search","text":"cats"}]')
+    assert [s.kind for s in steps] == ["click", "fill"]
+    assert steps[1].text == "cats"
+
+
+def test_parse_stops_at_irreversible_and_drops_the_rest():
+    steps = pd.parse_plan('[{"action":"click","target":"Options"},'
+                          '{"action":"click","target":"Send message"},'
+                          '{"action":"click","target":"Home"}]')
+    assert [s.target for s in steps] == ["Options"]  # Send refused, Home never reached
+
+
+def test_parse_malformed_and_junk_fail_open():
+    assert pd.parse_plan("I think you should click Options") == []
+    assert pd.parse_plan('{"action":"click"}') == []
+    assert pd.parse_plan("[]") == []
+    assert pd.parse_plan('[{"action":"hover","target":"X"},{"action":"click","target":""}]') == []
+
+
+def test_parse_caps_at_four_steps():
+    plan = "[" + ",".join('{"action":"click","target":"B%d"}' % i for i in range(9)) + "]"
+    assert len(pd.parse_plan(plan)) == 4