fix: prevent deadlock in page navigation and add page stability tests

- Initialize _page_switch_lock in Stagehand constructor
- Skip page stability check for navigation methods (goto, reload, go_back, go_forward)
- Use lock when switching active pages in context
- Add comprehensive tests for LivePageProxy functionality

Author: filip-michalsky

stagehand/context.py

@@ -101,15 +101,16 @@ def _handle_new_page(self, pw_page: Page):
 
         async def _async_handle():
             try:
-                self.stagehand.logger.debug(
-                    f"Creating StagehandPage for new page with URL: {pw_page.url}",
-                    category="context",
-                )
-                stagehand_page = await self.create_stagehand_page(pw_page)
-                self.set_active_page(stagehand_page)
-                self.stagehand.logger.debug(
-                    "New page detected and initialized", category="context"
-                )
+                async with self.stagehand._page_switch_lock:
+                    self.stagehand.logger.debug(
+                        f"Creating StagehandPage for new page with URL: {pw_page.url}",
+                        category="context",
+                    )
+                    stagehand_page = await self.create_stagehand_page(pw_page)
+                    self.set_active_page(stagehand_page)
+                    self.stagehand.logger.debug(
+                        "New page detected and initialized", category="context"
+                    )
             except Exception as e:
                 self.stagehand.logger.error(
                     f"Failed to initialize new page: {str(e)}", category="context"

stagehand/main.py

@@ -64,6 +64,9 @@ def __getattr__(self, name):
         # For async operations, make them wait for stability
         attr = getattr(active_page, name)
         if callable(attr) and asyncio.iscoroutinefunction(attr):
+            # Don't wait for stability on navigation methods
+            if name in ["goto", "reload", "go_back", "go_forward"]:
+                return attr
 
             async def wrapped(*args, **kwargs):
                 await self._ensure_page_stability()
@@ -266,6 +269,7 @@ def __init__(
         self._initialized = False  # Flag to track if init() has run
         self._closed = False  # Flag to track if resources have been closed
         self._live_page_proxy = None  # Live page proxy
+        self._page_switch_lock = asyncio.Lock()  # Lock for page stability
 
         # Setup LLM client if LOCAL mode
         self.llm = None

tests/unit/core/test_live_page_proxy.py

@@ -0,0 +1,212 @@
+"""Test the LivePageProxy functionality"""
+
+import asyncio
+import pytest
+from unittest.mock import AsyncMock, MagicMock, patch
+from stagehand.main import LivePageProxy, Stagehand
+from stagehand.page import StagehandPage
+
+
+@pytest.mark.asyncio
+async def test_live_page_proxy_basic_delegation(mock_stagehand_config):
+    """Test that LivePageProxy delegates to the active page"""
+    # Create a Stagehand instance
+    stagehand = Stagehand(config=mock_stagehand_config)
+    
+    # Mock pages
+    mock_original_page = MagicMock(spec=StagehandPage)
+    mock_original_page.url = "https://original.com"
+    mock_original_page.title = AsyncMock(return_value="Original Page")
+    
+    mock_active_page = MagicMock(spec=StagehandPage)
+    mock_active_page.url = "https://active.com"
+    mock_active_page.title = AsyncMock(return_value="Active Page")
+    
+    # Set up the pages
+    stagehand._original_page = mock_original_page
+    stagehand._active_page = mock_active_page
+    stagehand._initialized = True
+    
+    # Get the proxy
+    proxy = stagehand.page
+    
+    # Test that it delegates to the active page
+    assert proxy.url == "https://active.com"
+    title = await proxy.title()
+    assert title == "Active Page"
+
+
+@pytest.mark.asyncio
+async def test_live_page_proxy_falls_back_to_original(mock_stagehand_config):
+    """Test that LivePageProxy falls back to original page when no active page"""
+    # Create a Stagehand instance
+    stagehand = Stagehand(config=mock_stagehand_config)
+    
+    # Mock original page only
+    mock_original_page = MagicMock(spec=StagehandPage)
+    mock_original_page.url = "https://original.com"
+    
+    # Set up the pages
+    stagehand._original_page = mock_original_page
+    stagehand._active_page = None
+    stagehand._initialized = True
+    
+    # Get the proxy
+    proxy = stagehand.page
+    
+    # Test that it delegates to the original page
+    assert proxy.url == "https://original.com"
+
+
+@pytest.mark.asyncio
+async def test_live_page_proxy_page_stability(mock_stagehand_config):
+    """Test that LivePageProxy waits for page stability on async operations"""
+    # Create a Stagehand instance
+    stagehand = Stagehand(config=mock_stagehand_config)
+    
+    # Track lock acquisition
+    lock_acquired = False
+    lock_released = False
+    
+    class TestLock:
+        async def __aenter__(self):
+            nonlocal lock_acquired
+            lock_acquired = True
+            await asyncio.sleep(0.1)  # Simulate some work
+            return self
+            
+        async def __aexit__(self, *args):
+            nonlocal lock_released
+            lock_released = True
+    
+    stagehand._page_switch_lock = TestLock()
+    
+    # Mock page with async method
+    mock_page = MagicMock(spec=StagehandPage)
+    mock_page.click = AsyncMock(return_value=None)
+    
+    # Set up the pages
+    stagehand._original_page = mock_page
+    stagehand._active_page = mock_page
+    stagehand._initialized = True
+    
+    # Get the proxy
+    proxy = stagehand.page
+    
+    # Call an async method (should wait for stability)
+    await proxy.click("button")
+    
+    # Verify lock was acquired and released
+    assert lock_acquired
+    assert lock_released
+    mock_page.click.assert_called_once_with("button")
+
+
+@pytest.mark.asyncio
+async def test_live_page_proxy_navigation_no_stability_check(mock_stagehand_config):
+    """Test that navigation methods don't wait for page stability"""
+    # Create a Stagehand instance
+    stagehand = Stagehand(config=mock_stagehand_config)
+    
+    # Track lock acquisition (should not happen)
+    lock_acquired = False
+    
+    class TestLock:
+        async def __aenter__(self):
+            nonlocal lock_acquired
+            lock_acquired = True
+            return self
+            
+        async def __aexit__(self, *args):
+            pass
+    
+    stagehand._page_switch_lock = TestLock()
+    
+    # Mock page with navigation methods
+    mock_page = MagicMock(spec=StagehandPage)
+    mock_page.goto = AsyncMock(return_value=None)
+    mock_page.reload = AsyncMock(return_value=None)
+    mock_page.go_back = AsyncMock(return_value=None)
+    mock_page.go_forward = AsyncMock(return_value=None)
+    
+    # Set up the pages
+    stagehand._original_page = mock_page
+    stagehand._active_page = mock_page
+    stagehand._initialized = True
+    
+    # Get the proxy
+    proxy = stagehand.page
+    
+    # Call navigation methods (should NOT wait for stability)
+    await proxy.goto("https://example.com")
+    await proxy.reload()
+    await proxy.go_back()
+    await proxy.go_forward()
+    
+    # Verify lock was NOT acquired
+    assert not lock_acquired
+    
+    # Verify methods were called
+    mock_page.goto.assert_called_once_with("https://example.com")
+    mock_page.reload.assert_called_once()
+    mock_page.go_back.assert_called_once()
+    mock_page.go_forward.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_live_page_proxy_dynamic_page_switching(mock_stagehand_config):
+    """Test that LivePageProxy dynamically switches between pages"""
+    # Create a Stagehand instance
+    stagehand = Stagehand(config=mock_stagehand_config)
+    
+    # Mock pages
+    page1 = MagicMock(spec=StagehandPage)
+    page1.url = "https://page1.com"
+    
+    page2 = MagicMock(spec=StagehandPage)
+    page2.url = "https://page2.com"
+    
+    # Set up initial state
+    stagehand._original_page = page1
+    stagehand._active_page = page1
+    stagehand._initialized = True
+    
+    # Get the proxy
+    proxy = stagehand.page
+    
+    # Initially points to page1
+    assert proxy.url == "https://page1.com"
+    
+    # Switch active page
+    stagehand._active_page = page2
+    
+    # Now points to page2 without creating a new proxy
+    assert proxy.url == "https://page2.com"
+
+
+def test_live_page_proxy_no_page_error(mock_stagehand_config):
+    """Test that LivePageProxy raises error when no page is available"""
+    # Create a Stagehand instance
+    stagehand = Stagehand(config=mock_stagehand_config)
+    
+    # No pages set
+    stagehand._original_page = None
+    stagehand._active_page = None
+    stagehand._initialized = True
+    
+    # Get the proxy
+    proxy = stagehand.page
+    
+    # Accessing attributes should raise RuntimeError
+    with pytest.raises(RuntimeError, match="No active page available"):
+        _ = proxy.url
+
+
+def test_live_page_proxy_not_initialized(mock_stagehand_config):
+    """Test that page property returns None when not initialized"""
+    # Create a Stagehand instance
+    stagehand = Stagehand(config=mock_stagehand_config)
+    stagehand._initialized = False
+    
+    # Should return None
+    assert stagehand.page is None