SKY-10323: add main-world JS evaluation hook for remote browsers (#6653)

skyvern/forge/sdk/routes/streaming/channels/cdp.py

@@ -26,6 +26,7 @@ from skyvern.webeye.cdp_connection import (
     is_local_pbs_cdp_url,
     resolve_local_pbs_cdp_url,
 )
+from skyvern.webeye.main_world_eval import evaluate_in_main_world
 
 if t.TYPE_CHECKING:
     from skyvern.forge.sdk.routes.streaming.channels.vnc import VncChannel
@@ -185,7 +186,7 @@ class CdpChannel:
         LOG.info(f"{self.class_name} evaluating js", expression=expression[:100], **self.identity)
 
         try:
-            result = await self.page.evaluate(expression, arg)
+            result = await evaluate_in_main_world(self.page, expression, arg)
             LOG.info(f"{self.class_name} evaluated js successfully", **self.identity)
             return result
         except Exception:

skyvern/forge/sdk/routes/streaming/channels/execution.py

@@ -24,6 +24,7 @@ from skyvern.config import settings
 from skyvern.forge.sdk.routes.streaming.channels.cdp import CdpChannel
 from skyvern.forge.sdk.routes.streaming.payload_limits import MAX_SCREENSHOT_BYTES
 from skyvern.forge.sdk.routes.streaming.registries import get_vnc_channel
+from skyvern.webeye.main_world_eval import evaluate_in_main_world
 
 if t.TYPE_CHECKING:
     from skyvern.forge.sdk.routes.streaming.channels.message import MessageChannel
@@ -279,7 +280,7 @@ class LocalExecutionChannel(ExecutionChannel):
         # which LocalExecutionChannel intentionally leaves None.
         if not self.page:
             raise RuntimeError(f"{self.class_name} evaluate_js: no page available.")
-        return await self.page.evaluate(expression, arg)
+        return await evaluate_in_main_world(self.page, expression, arg)
 
     async def close(self) -> None:
         # We don't own the page or context; do not close.

skyvern/webeye/actions/handler.py

@@ -121,6 +121,7 @@ from skyvern.webeye.cdp_download_interceptor import (
     is_download_response,
     normalize_download_filename,
 )
+from skyvern.webeye.main_world_eval import evaluate_in_main_world
 from skyvern.webeye.scraper.scraped_page import (
     CleanupElementTreeFunc,
     ElementTreeBuilder,
@@ -3282,7 +3283,7 @@ async def handle_execute_js_action(
 ) -> list[ActionResult]:
     import json as _json
 
-    result = await page.evaluate(action.js_code)
+    result = await evaluate_in_main_world(page, action.js_code)
     if result is None:
         return [ActionSuccess(data="undefined")]
     if isinstance(result, str):

skyvern/webeye/main_world_eval.py

@@ -0,0 +1,110 @@
+"""Generic main-world JS evaluation hook.
+
+When a context has a prefix configured via ``configure_main_world_prefix``,
+``evaluate_in_main_world`` runs the script via a single CDP ``Runtime.evaluate``
+call with the prefix prepended, so middleware that inspects script content sees
+the prefix intact. With no prefix configured the call falls through to
+``page.evaluate`` with no extra CDP overhead. The prefix is opaque text.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import json
+import re
+import weakref
+from typing import Any
+
+from playwright.async_api import BrowserContext, Page
+
+_CONTEXT_PREFIXES: weakref.WeakKeyDictionary[BrowserContext, str] = weakref.WeakKeyDictionary()
+
+
+def configure_main_world_prefix(context: BrowserContext, prefix: str) -> None:
+    """Attach an opaque text prefix to ``context``; prepended to every JS body."""
+    _CONTEXT_PREFIXES[context] = prefix
+
+
+def clear_main_world_prefix(context: BrowserContext) -> None:
+    _CONTEXT_PREFIXES.pop(context, None)
+
+
+def get_main_world_prefix(context: BrowserContext) -> str | None:
+    return _CONTEXT_PREFIXES.get(context)
+
+
+def _resolve_prefix(page: Page) -> str | None:
+    return _CONTEXT_PREFIXES.get(page.context)
+
+
+# Conservative: only arrow / function declarations that Playwright would auto-wrap
+# as an IIFE. Bare expressions / object literals must NOT be wrapped.
+# Limitation: arrow param list is ``[^()]*``, so nested-paren params (e.g.
+# ``(a = (1+2)) => a``) don't match — current callers don't use that shape.
+_ARROW_FN_RE = re.compile(r"^\s*(async\s+)?(\([^()]*\)|\w+)\s*=>")
+_FUNCTION_DECL_RE = re.compile(r"^\s*(async\s+)?function\b")
+
+
+def _is_function_form(expression: str) -> bool:
+    return bool(_ARROW_FN_RE.match(expression) or _FUNCTION_DECL_RE.match(expression))
+
+
+def _extract_runtime_result(result: dict[str, Any]) -> Any:
+    # CDP returns ``value`` for JSON-serialisable types; unserialisable primitives
+    # (NaN/Infinity/-0/BigInt) fall through to None, matching what page.evaluate
+    # gives callers that don't opt into custom serialisation.
+    result_obj = result.get("result") or {}
+    if "value" in result_obj:
+        return result_obj["value"]
+    return None
+
+
+async def evaluate_in_main_world(page: Page, expression: str, arg: Any = None) -> Any:
+    """Evaluate ``expression`` in the page main world when a prefix is configured.
+
+    No prefix → identical to ``page.evaluate(expression, arg)``. Prefix → single
+    ``Runtime.evaluate`` call so middleware sees the prefix intact. Function-form
+    expressions are auto-wrapped as IIFEs; ``arg`` is inlined as a JSON literal.
+    Non-function expressions drop ``arg`` (mirroring page.evaluate). page.evaluate
+    is avoided here because Playwright's function-string normalisation requires
+    an unprefixed function head, which a leading prefix line breaks.
+    """
+    prefix = _resolve_prefix(page)
+    if prefix is None:
+        if arg is None:
+            return await page.evaluate(expression)
+        return await page.evaluate(expression, arg)
+
+    if _is_function_form(expression):
+        if arg is None:
+            wrapped = f"({expression})()"
+        else:
+            wrapped = f"({expression})({json.dumps(arg)})"
+    else:
+        wrapped = expression
+    body = f"{prefix}\n{wrapped}"
+    cdp_session = await page.context.new_cdp_session(page)
+    try:
+        result = await cdp_session.send(
+            "Runtime.evaluate",
+            {
+                "expression": body,
+                "returnByValue": True,
+                "awaitPromise": True,
+            },
+        )
+    finally:
+        with contextlib.suppress(Exception):
+            await cdp_session.detach()
+
+    exception_details = result.get("exceptionDetails")
+    if exception_details:
+        exception = exception_details.get("exception") or {}
+        description = (
+            exception.get("description")
+            or exception.get("value")
+            or exception_details.get("text", "Runtime.evaluate exception")
+        )
+        raise RuntimeError(f"main-world evaluate raised: {description}")
+
+    return _extract_runtime_result(result)

skyvern/webeye/utils/page.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import time
 from enum import StrEnum
 from io import BytesIO
@@ -18,6 +19,7 @@ from skyvern.exceptions import FailedToTakeScreenshot
 from skyvern.forge.sdk.core import skyvern_context
 from skyvern.forge.sdk.settings_manager import SettingsManager
 from skyvern.forge.sdk.trace import apply_context_attrs, traced
+from skyvern.webeye.main_world_eval import evaluate_in_main_world, get_main_world_prefix
 
 LOG = structlog.get_logger()
 
@@ -47,6 +49,29 @@ def _is_navigation_context_lost(error_msg: str) -> bool:
     return "ReferenceError" in error_msg and "is not defined" in error_msg
 
 
+def _is_json_inlinable(arg: Any) -> bool:
+    # ElementHandle / JSHandle aren't JSON-serialisable; those must keep
+    # Playwright's own marshalling instead of being inlined into Runtime.evaluate.
+    try:
+        json.dumps(arg)
+    except (TypeError, ValueError):
+        return False
+    return True
+
+
+async def _dispatch_evaluate(frame: Page | Frame, expression: str, arg: Any | None) -> Any:
+    # Page + prefix + JSON-safe arg → main-world hook (preserves the marker).
+    # Iframe Frames and non-JSON args fall back to per-frame evaluate so iframe
+    # contexts and Playwright handle-marshalling keep working.
+    if not isinstance(frame, Page):
+        return await frame.evaluate(expression=expression, arg=arg)
+    if get_main_world_prefix(frame.context) is None:
+        return await frame.evaluate(expression=expression, arg=arg)
+    if arg is not None and not _is_json_inlinable(arg):
+        return await frame.evaluate(expression=expression, arg=arg)
+    return await evaluate_in_main_world(frame, expression, arg)
+
+
 async def _wait_for_navigation_settle(frame: Page | Frame, timeout_ms: float) -> None:
     if timeout_ms <= 0:
         return
@@ -288,7 +313,7 @@ class SkyvernFrame:
     ) -> Any:
         try:
             async with asyncio.timeout(timeout_ms / 1000):
-                return await frame.evaluate(expression=expression, arg=arg)
+                return await _dispatch_evaluate(frame, expression, arg)
         except PlaywrightError as e:
             error_msg = str(e)
             if not _is_navigation_context_lost(error_msg):
@@ -300,6 +325,19 @@ class SkyvernFrame:
                 timeout_ms=timeout_ms,
                 initial_error=error_msg,
             )
+        except RuntimeError as e:
+            # `evaluate_in_main_world` raises RuntimeError on Runtime.evaluate
+            # exception payloads; only navigation-context-lost text recovers here.
+            error_msg = str(e)
+            if not _is_navigation_context_lost(error_msg):
+                raise
+            return await SkyvernFrame._evaluate_with_navigation_recovery(
+                frame=frame,
+                expression=expression,
+                arg=arg,
+                timeout_ms=timeout_ms,
+                initial_error=error_msg,
+            )
         except asyncio.TimeoutError:
             LOG.exception("Skyvern timed out trying to analyze the page", expression=expression)
             raise TimeoutError("Skyvern timed out trying to analyze the page")
@@ -351,14 +389,16 @@ class SkyvernFrame:
                 raise TimeoutError("Skyvern timed out trying to analyze the page")
             try:
                 async with asyncio.timeout(inject_budget):
-                    await frame.evaluate(expression=JS_FUNCTION_DEFS)
+                    # Same dispatch helper so a prefixed Page re-injects
+                    # JS_FUNCTION_DEFS via Runtime.evaluate (preserving the marker).
+                    await _dispatch_evaluate(frame, JS_FUNCTION_DEFS, None)
             except asyncio.TimeoutError:
                 LOG.exception(
                     "Skyvern timed out trying to analyze the page during domUtils.js re-injection",
                     expression=expression,
                 )
                 raise TimeoutError("Skyvern timed out trying to analyze the page")
-            except PlaywrightError as inject_err:
+            except (PlaywrightError, RuntimeError) as inject_err:
                 last_error_msg = str(inject_err)
                 if attempt == _NAVIGATION_RECOVERY_MAX_ATTEMPTS or not _is_navigation_context_lost(last_error_msg):
                     LOG.warning(
@@ -377,11 +417,11 @@ class SkyvernFrame:
                 raise TimeoutError("Skyvern timed out trying to analyze the page")
             try:
                 async with asyncio.timeout(retry_budget):
-                    return await frame.evaluate(expression=expression, arg=arg)
+                    return await _dispatch_evaluate(frame, expression, arg)
             except asyncio.TimeoutError:
                 LOG.exception("Skyvern timed out on retry after JS context re-injection", expression=expression)
                 raise TimeoutError("Skyvern timed out trying to analyze the page")
-            except PlaywrightError as retry_err:
+            except (PlaywrightError, RuntimeError) as retry_err:
                 last_error_msg = str(retry_err)
                 if attempt == _NAVIGATION_RECOVERY_MAX_ATTEMPTS or not _is_navigation_context_lost(last_error_msg):
                     raise

tests/unit/forge/sdk/routes/streaming/test_message_reify.py

@@ -241,7 +241,7 @@ class TestLocalExecutionChannel:
         result = await channel.evaluate_js("() => 'hello'")
 
         assert result == "hello"
-        page.evaluate.assert_awaited_once_with("() => 'hello'", None)
+        page.evaluate.assert_awaited_once_with("() => 'hello'")
 
     @pytest.mark.asyncio
     async def test_evaluate_js_forwards_arg(self) -> None:

tests/unit/webeye/test_main_world_eval.py

@@ -0,0 +1,290 @@
+"""Tests for the generic main-world evaluation hook."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from skyvern.webeye.main_world_eval import (
+    clear_main_world_prefix,
+    configure_main_world_prefix,
+    evaluate_in_main_world,
+    get_main_world_prefix,
+)
+
+
+class _MockBrowserContext:
+    """Hashable stand-in for BrowserContext (WeakKeyDictionary key)."""
+
+
+def _mock_page(prefix: str | None = None) -> tuple[MagicMock, _MockBrowserContext]:
+    context = _MockBrowserContext()
+    page = MagicMock()
+    page.context = context
+    page.evaluate = AsyncMock()
+    cdp_session = MagicMock()
+    cdp_session.send = AsyncMock(return_value={"result": {"value": "ok"}})
+    cdp_session.detach = AsyncMock()
+    page.context_cdp_session = cdp_session
+    context_mock = MagicMock(wraps=context)
+    context_mock.new_cdp_session = AsyncMock(return_value=cdp_session)
+    page.context = context_mock
+    if prefix is not None:
+        configure_main_world_prefix(context_mock, prefix)
+    return page, context_mock
+
+
+@pytest.fixture(autouse=True)
+def _reset_prefix_state() -> None:
+    # WeakKeyDictionary already drops mock contexts when they go out of scope.
+    return None
+
+
+class TestNoPrefixConfigured:
+    @pytest.mark.asyncio
+    async def test_passes_through_to_page_evaluate_no_arg(self) -> None:
+        page = MagicMock()
+        page.context = _MockBrowserContext()
+        page.evaluate = AsyncMock(return_value=42)
+
+        result = await evaluate_in_main_world(page, "() => 42")
+
+        assert result == 42
+        page.evaluate.assert_awaited_once_with("() => 42")
+
+    @pytest.mark.asyncio
+    async def test_passes_through_to_page_evaluate_with_arg(self) -> None:
+        page = MagicMock()
+        page.context = _MockBrowserContext()
+        page.evaluate = AsyncMock(return_value="hi")
+
+        result = await evaluate_in_main_world(page, "(x) => x", "hi")
+
+        assert result == "hi"
+        page.evaluate.assert_awaited_once_with("(x) => x", "hi")
+
+
+class TestPrefixConfiguredNoArg:
+    @pytest.mark.asyncio
+    async def test_function_form_wraps_as_iife_and_uses_runtime_evaluate(self) -> None:
+        page, context = _mock_page(prefix="// MARKER")
+        page.context.new_cdp_session.return_value.send = AsyncMock(return_value={"result": {"value": 7}})
+
+        result = await evaluate_in_main_world(page, "() => 7")
+
+        assert result == 7
+        page.evaluate.assert_not_awaited()
+        page.context.new_cdp_session.assert_awaited_once_with(page)
+        send_mock = page.context.new_cdp_session.return_value.send
+        send_mock.assert_awaited_once()
+        method, params = send_mock.await_args.args
+        assert method == "Runtime.evaluate"
+        assert params["expression"].startswith("// MARKER\n")
+        assert "(() => 7)()" in params["expression"]
+        assert params["returnByValue"] is True
+        assert params["awaitPromise"] is True
+        page.context.new_cdp_session.return_value.detach.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_statement_form_passes_through_unchanged_to_runtime_evaluate(self) -> None:
+        page, context = _mock_page(prefix="/* P */")
+        page.context.new_cdp_session.return_value.send = AsyncMock(return_value={"result": {"value": None}})
+
+        await evaluate_in_main_world(page, "window.x = 1;")
+
+        send_mock = page.context.new_cdp_session.return_value.send
+        method, params = send_mock.await_args.args
+        assert method == "Runtime.evaluate"
+        assert params["expression"] == "/* P */\nwindow.x = 1;"
+
+    @pytest.mark.asyncio
+    async def test_async_arrow_function_wraps_as_iife(self) -> None:
+        page, _ = _mock_page(prefix="// P")
+        page.context.new_cdp_session.return_value.send = AsyncMock(return_value={"result": {"value": "done"}})
+
+        await evaluate_in_main_world(page, "async () => fetch('/x')")
+
+        params = page.context.new_cdp_session.return_value.send.await_args.args[1]
+        assert "(async () => fetch('/x'))()" in params["expression"]
+
+    @pytest.mark.asyncio
+    async def test_runtime_evaluate_exception_is_raised(self) -> None:
+        page, _ = _mock_page(prefix="// P")
+        page.context.new_cdp_session.return_value.send = AsyncMock(
+            return_value={
+                "exceptionDetails": {
+                    "text": "Uncaught",
+                    "exception": {"description": "ReferenceError: x is not defined"},
+                }
+            }
+        )
+
+        with pytest.raises(RuntimeError, match="ReferenceError: x is not defined"):
+            await evaluate_in_main_world(page, "() => x")
+
+    @pytest.mark.asyncio
+    async def test_detach_is_called_even_on_failure(self) -> None:
+        page, _ = _mock_page(prefix="// P")
+        page.context.new_cdp_session.return_value.send = AsyncMock(side_effect=RuntimeError("boom"))
+        detach_mock = page.context.new_cdp_session.return_value.detach
+
+        with pytest.raises(RuntimeError, match="boom"):
+            await evaluate_in_main_world(page, "() => 1")
+
+        detach_mock.assert_awaited_once()
+
+
+class TestPrefixConfiguredWithArg:
+    @pytest.mark.asyncio
+    async def test_function_form_with_arg_inlines_arg_via_runtime_evaluate(self) -> None:
+        page, _ = _mock_page(prefix="// MARK")
+        page.context.new_cdp_session.return_value.send = AsyncMock(return_value={"result": {"value": "arg-value"}})
+
+        result = await evaluate_in_main_world(page, "(x) => x", "arg-value")
+
+        assert result == "arg-value"
+        # No page.evaluate: a leading prefix line breaks Playwright's function-string normalisation.
+        page.evaluate.assert_not_awaited()
+        params = page.context.new_cdp_session.return_value.send.await_args.args[1]
+        assert params["expression"] == '// MARK\n((x) => x)("arg-value")'
+
+    @pytest.mark.asyncio
+    async def test_function_declaration_form_with_arg_inlines_arg(self) -> None:
+        page, _ = _mock_page(prefix="// MARK")
+        page.context.new_cdp_session.return_value.send = AsyncMock(return_value={"result": {"value": 5}})
+
+        await evaluate_in_main_world(page, "function (x) { return x + 1 }", 4)
+
+        params = page.context.new_cdp_session.return_value.send.await_args.args[1]
+        assert params["expression"] == "// MARK\n(function (x) { return x + 1 })(4)"
+
+    @pytest.mark.asyncio
+    async def test_non_function_expression_with_arg_drops_arg(self) -> None:
+        """page.evaluate ignores extra args for non-function strings; mirror that."""
+        page, _ = _mock_page(prefix="// MARK")
+        page.context.new_cdp_session.return_value.send = AsyncMock(return_value={"result": {"value": 3}})
+
+        await evaluate_in_main_world(page, "1 + 2", "ignored")
+
+        params = page.context.new_cdp_session.return_value.send.await_args.args[1]
+        assert params["expression"] == "// MARK\n1 + 2"
+
+
+class TestExpressionShapeRegressions:
+    """Shapes the earlier prefix-startswith-`(` heuristic mis-wrapped."""
+
+    @pytest.mark.asyncio
+    async def test_parenthesised_arithmetic_is_not_wrapped(self) -> None:
+        page, _ = _mock_page(prefix="// P")
+        page.context.new_cdp_session.return_value.send = AsyncMock(return_value={"result": {"value": 3}})
+
+        await evaluate_in_main_world(page, "(1 + 2)")
+
+        params = page.context.new_cdp_session.return_value.send.await_args.args[1]
+        assert params["expression"] == "// P\n(1 + 2)"  # not ((1 + 2))() — that would TypeError
+
+    @pytest.mark.asyncio
+    async def test_object_literal_in_parens_is_not_wrapped(self) -> None:
+        page, _ = _mock_page(prefix="// P")
+        page.context.new_cdp_session.return_value.send = AsyncMock(return_value={"result": {"value": {"foo": 1}}})
+
+        await evaluate_in_main_world(page, "({foo: 1})")
+
+        params = page.context.new_cdp_session.return_value.send.await_args.args[1]
+        assert params["expression"] == "// P\n({foo: 1})"  # not (({foo: 1}))() — that would TypeError
+
+    @pytest.mark.asyncio
+    async def test_single_param_arrow_function_wraps_as_iife(self) -> None:
+        page, _ = _mock_page(prefix="// P")
+        page.context.new_cdp_session.return_value.send = AsyncMock(return_value={"result": {"value": 1}})
+
+        await evaluate_in_main_world(page, "x => x")
+
+        params = page.context.new_cdp_session.return_value.send.await_args.args[1]
+        assert "(x => x)()" in params["expression"]
+
+    @pytest.mark.asyncio
+    async def test_destructured_param_arrow_function_wraps_as_iife(self) -> None:
+        page, _ = _mock_page(prefix="// P")
+        page.context.new_cdp_session.return_value.send = AsyncMock(return_value={"result": {"value": 3}})
+
+        await evaluate_in_main_world(page, "({a, b}) => a + b")
+
+        params = page.context.new_cdp_session.return_value.send.await_args.args[1]
+        assert "(({a, b}) => a + b)()" in params["expression"]
+
+    @pytest.mark.asyncio
+    async def test_default_param_arrow_function_wraps_as_iife(self) -> None:
+        page, _ = _mock_page(prefix="// P")
+        page.context.new_cdp_session.return_value.send = AsyncMock(return_value={"result": {"value": 1}})
+
+        await evaluate_in_main_world(page, "(x = 1) => x")
+
+        params = page.context.new_cdp_session.return_value.send.await_args.args[1]
+        assert "((x = 1) => x)()" in params["expression"]
+
+    @pytest.mark.asyncio
+    async def test_named_function_with_arg_inlines_arg(self) -> None:
+        page, _ = _mock_page(prefix="// P")
+        page.context.new_cdp_session.return_value.send = AsyncMock(return_value={"result": {"value": 5}})
+
+        await evaluate_in_main_world(page, "function foo(x) { return x }", 5)
+
+        params = page.context.new_cdp_session.return_value.send.await_args.args[1]
+        assert "(function foo(x) { return x })(5)" in params["expression"]
+
+
+class TestRuntimeResultDecoding:
+    @pytest.mark.asyncio
+    async def test_returns_none_when_runtime_value_absent(self) -> None:
+        page, _ = _mock_page(prefix="// P")
+        page.context.new_cdp_session.return_value.send = AsyncMock(return_value={"result": {}})
+
+        result = await evaluate_in_main_world(page, "() => undefined")
+
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_unserializable_value_falls_through_to_none(self) -> None:
+        """CDP encodes NaN/Infinity via ``unserializableValue``; callers expect None."""
+        page, _ = _mock_page(prefix="// P")
+        page.context.new_cdp_session.return_value.send = AsyncMock(
+            return_value={"result": {"type": "number", "unserializableValue": "NaN"}}
+        )
+
+        result = await evaluate_in_main_world(page, "() => NaN")
+
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_exception_details_without_description_uses_text_fallback(self) -> None:
+        page, _ = _mock_page(prefix="// P")
+        page.context.new_cdp_session.return_value.send = AsyncMock(
+            return_value={
+                "exceptionDetails": {
+                    "text": "Uncaught (in promise)",
+                    "exception": {"type": "object"},
+                }
+            }
+        )
+
+        with pytest.raises(RuntimeError, match="Uncaught"):
+            await evaluate_in_main_world(page, "() => 1")
+
+
+class TestPrefixRegistry:
+    def test_configure_and_get_round_trip(self) -> None:
+        context = _MockBrowserContext()
+        configure_main_world_prefix(context, "// FOO")
+        assert get_main_world_prefix(context) == "// FOO"
+
+    def test_clear_removes_prefix(self) -> None:
+        context = _MockBrowserContext()
+        configure_main_world_prefix(context, "// FOO")
+        clear_main_world_prefix(context)
+        assert get_main_world_prefix(context) is None
+
+    def test_get_returns_none_when_unconfigured(self) -> None:
+        context = _MockBrowserContext()
+        assert get_main_world_prefix(context) is None

tests/unit/webeye/test_skyvern_frame_main_world_routing.py

@@ -0,0 +1,245 @@
+"""Tests for SkyvernFrame.evaluate routing into the main-world eval hook."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from playwright.async_api import Frame, Page
+
+from skyvern.webeye.main_world_eval import (
+    clear_main_world_prefix,
+    configure_main_world_prefix,
+)
+from skyvern.webeye.utils.page import SkyvernFrame
+
+
+class _HashableContext:
+    """Hashable + weak-ref-able stand-in for BrowserContext (WeakKeyDictionary key)."""
+
+
+def _make_page_mock(prefix: str | None) -> tuple[MagicMock, _HashableContext]:
+    context = _HashableContext()
+    page = MagicMock(spec=Page)
+    page.context = context
+    page.evaluate = AsyncMock(return_value="page-evaluate-result")
+    cdp_session = MagicMock()
+    cdp_session.send = AsyncMock(return_value={"result": {"value": "runtime-evaluate-result"}})
+    cdp_session.detach = AsyncMock()
+    page.context.new_cdp_session = AsyncMock(return_value=cdp_session)  # type: ignore[attr-defined]
+    if prefix is not None:
+        configure_main_world_prefix(context, prefix)  # type: ignore[arg-type]
+    return page, context
+
+
+def _make_frame_mock() -> MagicMock:
+    """A Frame mock that is NOT a Page (iframe-style)."""
+    frame = MagicMock(spec=Frame)
+    frame.evaluate = AsyncMock(return_value="frame-evaluate-result")
+    return frame
+
+
+@pytest.fixture(autouse=True)
+def _short_timeout() -> object:
+    return None
+
+
+class TestSkyvernFrameEvaluateRouting:
+    @pytest.mark.asyncio
+    async def test_page_without_prefix_delegates_to_page_evaluate(self) -> None:
+        page, ctx = _make_page_mock(prefix=None)
+
+        result = await SkyvernFrame.evaluate(page, "() => 1")
+
+        assert result == "page-evaluate-result"
+        page.evaluate.assert_awaited_once_with(expression="() => 1", arg=None)
+        page.context.new_cdp_session.assert_not_awaited()  # type: ignore[attr-defined]
+
+    @pytest.mark.asyncio
+    async def test_page_with_prefix_and_no_arg_routes_through_runtime_evaluate(self) -> None:
+        page, ctx = _make_page_mock(prefix="// MARK")
+        try:
+            result = await SkyvernFrame.evaluate(page, "() => 7")
+        finally:
+            clear_main_world_prefix(ctx)  # type: ignore[arg-type]
+
+        assert result == "runtime-evaluate-result"
+        page.evaluate.assert_not_awaited()
+        params = page.context.new_cdp_session.return_value.send.await_args.args[1]  # type: ignore[attr-defined]
+        assert params["expression"].startswith("// MARK\n")
+        assert "(() => 7)()" in params["expression"]
+
+    @pytest.mark.asyncio
+    async def test_page_with_prefix_and_json_arg_inlines_arg(self) -> None:
+        page, ctx = _make_page_mock(prefix="// MARK")
+        try:
+            await SkyvernFrame.evaluate(page, "(pos) => __pwCursorMove(pos)", [1.5, 2.5])
+        finally:
+            clear_main_world_prefix(ctx)  # type: ignore[arg-type]
+
+        params = page.context.new_cdp_session.return_value.send.await_args.args[1]  # type: ignore[attr-defined]
+        assert "((pos) => __pwCursorMove(pos))([1.5, 2.5])" in params["expression"]
+        page.evaluate.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_page_with_prefix_and_non_json_arg_falls_back_to_page_evaluate(self) -> None:
+        """ElementHandle args can't be JSON-inlined → keep Playwright marshalling."""
+        page, ctx = _make_page_mock(prefix="// MARK")
+        element_handle_like = MagicMock()  # stands in for an ElementHandle/JSHandle
+        try:
+            await SkyvernFrame.evaluate(page, "(el) => el.blur()", element_handle_like)
+        finally:
+            clear_main_world_prefix(ctx)  # type: ignore[arg-type]
+
+        page.evaluate.assert_awaited_once_with(expression="(el) => el.blur()", arg=element_handle_like)
+        page.context.new_cdp_session.assert_not_awaited()  # type: ignore[attr-defined]
+
+    @pytest.mark.asyncio
+    async def test_frame_target_never_routes_through_runtime_evaluate(self) -> None:
+        """Iframe Frames have their own world; page-level Runtime.evaluate lands elsewhere."""
+        frame = _make_frame_mock()
+
+        await SkyvernFrame.evaluate(frame, "() => 1")
+
+        frame.evaluate.assert_awaited_once_with(expression="() => 1", arg=None)
+
+    @pytest.mark.asyncio
+    async def test_runtime_evaluate_navigation_context_lost_triggers_recovery(self) -> None:
+        """Runtime.evaluate context-destroyed must hit recovery, like PlaywrightError."""
+        page, ctx = _make_page_mock(prefix="// MARK")
+        page.context.new_cdp_session.return_value.send = AsyncMock(  # type: ignore[attr-defined]
+            return_value={
+                "exceptionDetails": {
+                    "text": "Uncaught",
+                    "exception": {"description": "Error: Execution context was destroyed by a navigation"},
+                }
+            }
+        )
+
+        with patch.object(
+            SkyvernFrame,
+            "_evaluate_with_navigation_recovery",
+            new=AsyncMock(return_value="recovered"),
+        ) as recovery:
+            try:
+                result = await SkyvernFrame.evaluate(page, "() => 1")
+            finally:
+                clear_main_world_prefix(ctx)  # type: ignore[arg-type]
+
+        assert result == "recovered"
+        recovery.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_runtime_evaluate_non_recovery_runtime_error_propagates(self) -> None:
+        """Non-navigation RuntimeError must propagate, not get swallowed."""
+        page, ctx = _make_page_mock(prefix="// MARK")
+        page.context.new_cdp_session.return_value.send = AsyncMock(  # type: ignore[attr-defined]
+            return_value={
+                "exceptionDetails": {
+                    "text": "Uncaught",
+                    "exception": {"description": "TypeError: undefined is not a function"},
+                }
+            }
+        )
+
+        try:
+            with pytest.raises(RuntimeError, match="undefined is not a function"):
+                await SkyvernFrame.evaluate(page, "() => x.y()")
+        finally:
+            clear_main_world_prefix(ctx)  # type: ignore[arg-type]
+
+
+class TestNavigationRecoveryRouting:
+    """Recovery loop must keep using the main-world hook for prefixed Pages,
+    otherwise the marker is dropped on the post-navigation re-injection + retry."""
+
+    @pytest.mark.asyncio
+    async def test_prefixed_page_recovery_uses_runtime_evaluate_throughout(self) -> None:
+        """For a prefixed Page, the original call, the JS_FUNCTION_DEFS
+        re-injection, AND the final retry must all run via CDP Runtime.evaluate."""
+        page, ctx = _make_page_mock(prefix="// MARK")
+        page.wait_for_load_state = AsyncMock()
+
+        send_calls: list[dict[str, str]] = []
+
+        async def fake_send(method: str, params: dict[str, str]) -> dict:
+            send_calls.append({"method": method, "expression": params["expression"]})
+            if len(send_calls) == 1:
+                return {
+                    "exceptionDetails": {
+                        "text": "Uncaught",
+                        "exception": {"description": "Error: Execution context was destroyed by a navigation"},
+                    }
+                }
+            if len(send_calls) == 2:
+                return {"result": {"value": None}}  # re-injection of JS_FUNCTION_DEFS
+            return {"result": {"value": "final"}}  # final retry
+
+        cdp_session = MagicMock()
+        cdp_session.send = AsyncMock(side_effect=fake_send)
+        cdp_session.detach = AsyncMock()
+        page.context.new_cdp_session = AsyncMock(return_value=cdp_session)  # type: ignore[attr-defined]
+
+        try:
+            result = await SkyvernFrame.evaluate(page, "() => 1", timeout_ms=1000)
+        finally:
+            clear_main_world_prefix(ctx)  # type: ignore[arg-type]
+
+        assert result == "final"
+        assert len(send_calls) == 3
+        # Every CDP call carried the configured prefix.
+        for call in send_calls:
+            assert call["method"] == "Runtime.evaluate"
+            assert call["expression"].startswith("// MARK\n")
+        # Recovery must NOT have fallen back to raw page.evaluate for a JSON-safe call.
+        page.evaluate.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_recovery_with_element_handle_arg_preserves_page_evaluate_on_final_retry(
+        self,
+    ) -> None:
+        """ElementHandle arg → final retry keeps page.evaluate (re-injection has
+        arg=None so it can still go through the main-world hook)."""
+        from playwright._impl._errors import Error as PlaywrightError
+
+        page, ctx = _make_page_mock(prefix="// MARK")
+        page.wait_for_load_state = AsyncMock()
+
+        element_handle_like = MagicMock()  # non-JSON-inlinable
+        page.evaluate = AsyncMock(
+            side_effect=[
+                PlaywrightError("Execution context was destroyed by a navigation"),
+                "final-via-page-evaluate",
+            ]
+        )
+
+        try:
+            result = await SkyvernFrame.evaluate(page, "(el) => el.blur()", element_handle_like, timeout_ms=1000)
+        finally:
+            clear_main_world_prefix(ctx)  # type: ignore[arg-type]
+
+        assert result == "final-via-page-evaluate"
+        # Original call + final retry both went through page.evaluate (non-JSON arg).
+        # Re-injection used Runtime.evaluate (None arg → main-world hook is safe).
+        assert page.evaluate.await_count == 2
+        page.context.new_cdp_session.assert_awaited_once()  # type: ignore[attr-defined]
+
+    @pytest.mark.asyncio
+    async def test_frame_target_recovery_still_uses_frame_evaluate(self) -> None:
+        """Iframe Frames keep per-frame evaluate even during recovery."""
+        from playwright._impl._errors import Error as PlaywrightError
+
+        frame = _make_frame_mock()
+        frame.wait_for_load_state = AsyncMock()
+        frame.evaluate = AsyncMock(
+            side_effect=[
+                PlaywrightError("Execution context was destroyed by a navigation"),
+                None,
+                "frame-final",
+            ]
+        )
+
+        result = await SkyvernFrame.evaluate(frame, "() => 1", timeout_ms=1000)
+
+        assert result == "frame-final"
+        assert frame.evaluate.await_count == 3