Fail fast on reused coroutine callback errors

Co-authored-by: Shri Sukhani <shrisukhani@users.noreply.github.com>

Author: cursoragent

hyperbrowser/client/polling.py

@@ -122,6 +122,10 @@ def _invoke_non_retryable_callback(
 
 
 def _is_retryable_exception(exc: Exception) -> bool:
+    if isinstance(
+        exc, RuntimeError
+    ) and "cannot reuse already awaited coroutine" in str(exc):
+        return False
     if isinstance(exc, ConcurrentCancelledError):
         return False
     if isinstance(exc, _NonRetryablePollingError):

tests/test_polling.py

@@ -1007,6 +1007,60 @@ async def operation() -> str:
     asyncio.run(run())
 
 
+def test_poll_until_terminal_status_async_does_not_retry_reused_coroutines():
+    async def run() -> None:
+        attempts = {"count": 0}
+        shared_status_coroutine = asyncio.sleep(0, result="running")
+
+        async def get_status() -> str:
+            attempts["count"] += 1
+            return await shared_status_coroutine
+
+        with pytest.raises(
+            RuntimeError, match="cannot reuse already awaited coroutine"
+        ):
+            await poll_until_terminal_status_async(
+                operation_name="async poll reused coroutine",
+                get_status=get_status,
+                is_terminal_status=lambda value: value == "completed",
+                poll_interval_seconds=0.0001,
+                max_wait_seconds=1.0,
+                max_status_failures=5,
+            )
+
+        assert attempts["count"] == 2
+
+    asyncio.run(run())
+
+
+def test_retry_operation_async_does_not_retry_reused_coroutines():
+    async def run() -> None:
+        attempts = {"count": 0}
+
+        async def shared_operation() -> str:
+            raise ValueError("transient")
+
+        shared_operation_coroutine = shared_operation()
+
+        async def operation() -> str:
+            attempts["count"] += 1
+            return await shared_operation_coroutine
+
+        with pytest.raises(
+            RuntimeError, match="cannot reuse already awaited coroutine"
+        ):
+            await retry_operation_async(
+                operation_name="async retry reused coroutine",
+                operation=operation,
+                max_attempts=5,
+                retry_delay_seconds=0.0001,
+            )
+
+        assert attempts["count"] == 2
+
+    asyncio.run(run())
+
+
 def test_async_poll_until_terminal_status_allows_immediate_terminal_on_zero_max_wait():
     async def run() -> None:
         status = await poll_until_terminal_status_async(
@@ -2056,6 +2110,36 @@ async def get_next_page(page: int) -> dict:
     asyncio.run(run())
 
 
+def test_collect_paginated_results_async_does_not_retry_reused_coroutines():
+    async def run() -> None:
+        attempts = {"count": 0}
+        shared_page_coroutine = asyncio.sleep(
+            0, result={"current": 1, "total": 2, "items": []}
+        )
+
+        async def get_next_page(page: int) -> dict:
+            attempts["count"] += 1
+            return await shared_page_coroutine
+
+        with pytest.raises(
+            RuntimeError, match="cannot reuse already awaited coroutine"
+        ):
+            await collect_paginated_results_async(
+                operation_name="async paginated reused coroutine",
+                get_next_page=get_next_page,
+                get_current_page_batch=lambda response: response["current"],
+                get_total_page_batches=lambda response: response["total"],
+                on_page_success=lambda response: None,
+                max_wait_seconds=1.0,
+                max_attempts=5,
+                retry_delay_seconds=0.0001,
+            )
+
+        assert attempts["count"] == 2
+
+    asyncio.run(run())
+
+
 def test_wait_for_job_result_returns_fetched_value():
     status_values = iter(["running", "completed"])