tweaking scroll_by

Author: SentienceDEV

predicate/agent_runtime.py

@@ -326,6 +326,34 @@ async def get_url(self) -> str:
         self._cached_url = url
         return url
 
+    async def get_viewport_height(self) -> int:
+        """
+        Get current viewport height in pixels.
+
+        Returns:
+            Viewport height in pixels, or 800 as fallback if unavailable
+        """
+        try:
+            # Try refresh_page_info first (PlaywrightBackend)
+            refresh_fn = getattr(self.backend, "refresh_page_info", None)
+            if callable(refresh_fn):
+                info = await refresh_fn()
+                height = getattr(info, "height", None)
+                if height and height > 0:
+                    return int(height)
+
+            # Try evaluating JavaScript directly
+            eval_fn = getattr(self.backend, "eval", None)
+            if callable(eval_fn):
+                height = await eval_fn("window.innerHeight")
+                if height and height > 0:
+                    return int(height)
+        except Exception:
+            pass
+
+        # Fallback to reasonable default
+        return 800
+
     # -------------------------------------------------------------------------
     # Action methods for PlannerExecutorAgent compatibility
     # -------------------------------------------------------------------------

predicate/agents/planner_executor_agent.py

@@ -218,6 +218,9 @@ class SnapshotEscalationConfig:
 
         # Enable scroll-after-escalation to find elements below/above viewport
         config = SnapshotEscalationConfig(scroll_after_escalation=True, scroll_directions=("down", "up"))
+
+        # Custom scroll amount as fraction of viewport height (default: 0.4 = 40%)
+        config = SnapshotEscalationConfig(scroll_viewport_fraction=0.5)  # 50% of viewport
     """
 
     enabled: bool = True
@@ -228,6 +231,7 @@ class SnapshotEscalationConfig:
     scroll_after_escalation: bool = True
     scroll_max_attempts: int = 3  # Max scrolls per direction
     scroll_directions: tuple[str, ...] = ("down", "up")  # Directions to try
+    scroll_viewport_fraction: float = 0.4  # Scroll by 40% of viewport height (adaptive to screen size)
 
 
 @dataclass(frozen=True)
@@ -455,8 +459,9 @@ class AuthBoundaryConfig:
         "/log-in",
         "/auth",
         "/authenticate",
-        "/ap/signin",  # Amazon
-        "/ap/register",  # Amazon
+        "/ap/signin",  # Amazon sign-in
+        "/ap/register",  # Amazon registration
+        "/ax/claim",  # Amazon CAPTCHA/verification
         "/account/login",
         "/accounts/login",
         "/user/login",
@@ -1866,6 +1871,9 @@ async def _snapshot_with_escalation(
                     break
 
                 # Check element count - if sufficient, no need to escalate
+                # NOTE: Limit escalation is based on element COUNT only, not on whether
+                # a specific target element was found. Intent heuristics are only used
+                # for scroll-after-escalation AFTER limit escalation is exhausted.
                 elements = getattr(snap, "elements", []) or []
                 if len(elements) >= 10:
                     break
@@ -1909,15 +1917,35 @@ async def _snapshot_with_escalation(
                 if self.config.verbose:
                     print(f"  [SNAPSHOT-ESCALATION] Target element not found, trying scroll-after-escalation...", flush=True)
 
+                # Get viewport height and calculate scroll delta
+                viewport_height = await runtime.get_viewport_height()
+                scroll_delta = viewport_height * cfg.scroll_viewport_fraction
+
                 for direction in cfg.scroll_directions:
+                    # Map direction to dy (pixels): down=positive, up=negative
+                    scroll_dy = scroll_delta if direction == "down" else -scroll_delta
+
                     for scroll_num in range(cfg.scroll_max_attempts):
                         if self.config.verbose:
                             print(f"  [SNAPSHOT-ESCALATION] Scrolling {direction} ({scroll_num + 1}/{cfg.scroll_max_attempts})...", flush=True)
 
-                        # Scroll
-                        await runtime.scroll(direction)
+                        # Scroll with deterministic verification
+                        # scroll_by() returns False if scroll had no effect (reached page boundary)
+                        scroll_effective = await runtime.scroll_by(
+                            dy=scroll_dy,
+                            verify=True,
+                            min_delta_px=50.0,
+                            js_fallback=True,
+                            required=False,  # Don't fail the task if scroll doesn't work
+                            timeout_s=5.0,
+                        )
 
-                        # Wait for stabilization
+                        if not scroll_effective:
+                            if self.config.verbose:
+                                print(f"  [SNAPSHOT-ESCALATION] Scroll {direction} had no effect (reached boundary), skipping remaining attempts", flush=True)
+                            break  # No point trying more scrolls in this direction
+
+                        # Wait for stabilization after successful scroll
                         if self.config.stabilize_enabled:
                             await asyncio.sleep(self.config.stabilize_poll_s)
 
@@ -2778,6 +2806,40 @@ async def _execute_step(
 
                     return outcome
 
+            # Pre-step auth boundary check: stop early if on signin page without credentials
+            # This prevents executing login steps that would fail or use fake credentials
+            if self.config.auth_boundary.enabled and self.config.auth_boundary.stop_on_auth:
+                is_auth_page = await self._detect_auth_boundary(runtime)
+                if is_auth_page:
+                    if self.config.verbose:
+                        print(f"  [AUTH] Auth boundary detected at step start - stopping gracefully", flush=True)
+
+                    outcome = StepOutcome(
+                        step_id=step.id,
+                        goal=step.goal,
+                        status=StepStatus.SUCCESS,  # Graceful stop = success
+                        action_taken="AUTH_BOUNDARY_REACHED",
+                        verification_passed=True,
+                        error=self.config.auth_boundary.auth_success_message,
+                        duration_ms=int((time.time() - start_time) * 1000),
+                        url_before=pre_url,
+                        url_after=pre_url,
+                    )
+
+                    self._emit_step_end(
+                        step_id=step_id,
+                        step_index=step_index,
+                        step=step,
+                        outcome=outcome,
+                        pre_url=pre_url,
+                        post_url=pre_url,
+                        llm_response=None,
+                        snapshot_digest=None,
+                    )
+
+                    # Return special outcome that signals run completion
+                    return outcome
+
             # Wait for page to stabilize before snapshot
             if self.config.stabilize_enabled:
                 await runtime.stabilize()
@@ -3243,6 +3305,12 @@ async def run(
                 outcome = await self._execute_step(step, runtime, step_index)
                 step_outcomes.append(outcome)
 
+                # Check if auth boundary was reached at step start (graceful termination)
+                if outcome.action_taken == "AUTH_BOUNDARY_REACHED":
+                    if self.config.verbose:
+                        print(f"  [AUTH] Run completed at authentication boundary", flush=True)
+                    break  # Graceful termination - no error
+
                 # Record checkpoint on success (for recovery)
                 if outcome.status in (StepStatus.SUCCESS, StepStatus.VISION_FALLBACK):
                     if self._recovery_state is not None and outcome.url_after:

tests/unit/test_planner_executor_agent.py

@@ -17,6 +17,7 @@
 from typing import Any
 
 from predicate.agents.planner_executor_agent import (
+    AuthBoundaryConfig,
     ExecutorOverride,
     IntentHeuristics,
     ModalDismissalConfig,
@@ -687,6 +688,14 @@ def test_custom_scroll_settings(self) -> None:
         assert config.scroll_max_attempts == 5
         assert config.scroll_directions == ("down",)
 
+    def test_scroll_viewport_fraction_default(self) -> None:
+        config = SnapshotEscalationConfig()
+        assert config.scroll_viewport_fraction == 0.4
+
+    def test_scroll_viewport_fraction_custom(self) -> None:
+        config = SnapshotEscalationConfig(scroll_viewport_fraction=0.5)
+        assert config.scroll_viewport_fraction == 0.5
+
     def test_scroll_directions_can_be_reordered(self) -> None:
         # Try up first, then down
         config = SnapshotEscalationConfig(
@@ -788,13 +797,26 @@ def __init__(self, elements: list[MockElement], url: str = "https://example.com"
 
 
 class MockRuntime:
-    """Mock runtime for testing scroll-after-escalation."""
+    """Mock runtime for testing scroll-after-escalation and auth boundary."""
 
-    def __init__(self, snapshots_by_scroll: dict[int, MockSnapshot] | None = None):
+    def __init__(
+        self,
+        snapshots_by_scroll: dict[int, MockSnapshot] | None = None,
+        url: str = "https://example.com",
+    ):
         self.scroll_count = 0
         self.scroll_directions: list[str] = []
         self.snapshots_by_scroll = snapshots_by_scroll or {}
         self.default_snapshot = MockSnapshot([MockElement(1, "button", "Submit")])
+        self._url = url
+
+    async def get_url(self) -> str:
+        """Return the current URL for auth boundary detection."""
+        return self._url
+
+    async def get_viewport_height(self) -> int:
+        """Return mock viewport height."""
+        return 800  # Standard viewport height
 
     async def snapshot(
         self,
@@ -810,6 +832,23 @@ async def scroll(self, direction: str = "down") -> None:
         self.scroll_count += 1
         self.scroll_directions.append(direction)
 
+    async def scroll_by(
+        self,
+        dy: float,
+        *,
+        verify: bool = True,
+        min_delta_px: float = 50.0,
+        js_fallback: bool = True,
+        required: bool = True,
+        timeout_s: float = 10.0,
+        **kwargs: Any,
+    ) -> bool:
+        """Mock scroll_by with verification - always returns True (scroll effective)."""
+        direction = "down" if dy > 0 else "up"
+        self.scroll_count += 1
+        self.scroll_directions.append(direction)
+        return True  # Scroll always effective in tests
+
     async def stabilize(self) -> None:
         pass
 
@@ -1136,3 +1175,74 @@ async def test_no_scroll_without_intent_heuristics(self) -> None:
 
         # No scrolling should occur without intent_heuristics
         assert runtime.scroll_count == 0
+
+
+# ---------------------------------------------------------------------------
+# Test AuthBoundaryConfig
+# ---------------------------------------------------------------------------
+
+
+class TestAuthBoundaryConfig:
+    """Tests for AuthBoundaryConfig."""
+
+    def test_default_values(self) -> None:
+        config = AuthBoundaryConfig()
+        assert config.enabled is True
+        assert config.stop_on_auth is True
+        assert "/signin" in config.url_patterns
+        assert "/ap/signin" in config.url_patterns
+        assert "/ax/claim" in config.url_patterns  # Amazon CAPTCHA
+
+    def test_default_url_patterns_include_common_patterns(self) -> None:
+        config = AuthBoundaryConfig()
+        expected_patterns = [
+            "/signin",
+            "/sign-in",
+            "/login",
+            "/log-in",
+            "/auth",
+            "/authenticate",
+            "/ap/signin",  # Amazon
+            "/ap/register",  # Amazon
+            "/ax/claim",  # Amazon CAPTCHA
+            "/account/login",
+            "/accounts/login",
+            "/user/login",
+        ]
+        for pattern in expected_patterns:
+            assert pattern in config.url_patterns, f"Missing pattern: {pattern}"
+
+    def test_can_be_disabled(self) -> None:
+        config = AuthBoundaryConfig(enabled=False)
+        assert config.enabled is False
+
+    def test_stop_on_auth_can_be_disabled(self) -> None:
+        config = AuthBoundaryConfig(stop_on_auth=False)
+        assert config.stop_on_auth is False
+
+    def test_custom_url_patterns(self) -> None:
+        config = AuthBoundaryConfig(
+            url_patterns=("/custom/login", "/my-signin"),
+        )
+        assert config.url_patterns == ("/custom/login", "/my-signin")
+
+    def test_custom_auth_success_message(self) -> None:
+        config = AuthBoundaryConfig(
+            auth_success_message="Custom: Login required",
+        )
+        assert config.auth_success_message == "Custom: Login required"
+
+    def test_planner_executor_config_has_auth_boundary(self) -> None:
+        config = PlannerExecutorConfig()
+        assert config.auth_boundary is not None
+        assert config.auth_boundary.enabled is True
+
+    def test_planner_executor_config_custom_auth_boundary(self) -> None:
+        config = PlannerExecutorConfig(
+            auth_boundary=AuthBoundaryConfig(
+                enabled=True,
+                url_patterns=("/custom-signin",),
+                stop_on_auth=True,
+            ),
+        )
+        assert config.auth_boundary.url_patterns == ("/custom-signin",)