added pagination fan out pagination abstraction

src/monitor_utils/runner.py

@@ -1,7 +1,7 @@
 import asyncio
 import logging
 from collections.abc import Awaitable, Callable, Iterable
-from contextlib import nullcontext
+from contextlib import asynccontextmanager, nullcontext
 
 import aiohttp
 from aioproxies import AioProxies
@@ -568,3 +568,38 @@ def first_error(group_exc: BaseExceptionGroup) -> BaseException:
         error = error.exceptions[0]
 
     return error
+
+
+@asynccontextmanager
+async def fan_out_group():
+    """A TaskGroup for page-one fan-out whose failure surfaces unwrapped.
+
+    Use it where one page's processing must be bound to the pagination that
+    page spawns: create the processing task on the yielded group, then await
+    execute_pagination in the body. The group cancels whichever sibling is
+    still running when the other fails, so a dead category stops doing work
+    nobody will use, and a cancelled cycle takes both down together.
+
+    The unwrapping is the reason this is shared rather than inlined. A bare
+    TaskGroup always raises an ExceptionGroup, and run_monitor_cycle classifies
+    task results with isinstance against EXPECTED_TASK_ERRORS -- an
+    ExceptionGroup matches neither branch as intended, so a routine page
+    failure would be reclassified as an unexpected one and kill the monitor.
+    first_error unwraps to a leaf, preferring an unexpected exception so a real
+    bug is never masked by a network error.
+
+    Note the caller's own body is inside the group, so a plain exception raised
+    there (not from a child) propagates unchanged rather than being wrapped,
+    and an early `return` still awaits the child tasks.
+
+    Example:
+        async with fan_out_group() as group:
+            group.create_task(process_page(data, ...))
+            await execute_pagination(...)
+    """
+
+    try:
+        async with asyncio.TaskGroup() as group:
+            yield group
+    except BaseExceptionGroup as group_exc:
+        raise first_error(group_exc) from None

tests/test_fan_out_group.py

@@ -0,0 +1,126 @@
+"""unit test: fan_out_group binds sibling tasks and surfaces errors unwrapped
+
+run from the repo root:
+    uv run pytest tests/test_fan_out_group.py
+
+offline — no network. These lock the contract monitors depend on: a bare
+TaskGroup always raises an ExceptionGroup, which run_monitor_cycle would
+misclassify (it matches neither the EXPECTED_TASK_ERRORS branch nor the
+unexpected-Exception branch as intended), so the unwrapping is load-bearing and
+not cosmetic.
+"""
+
+import asyncio
+
+import aiohttp
+import pytest
+
+from monitor_utils.runner import fan_out_group
+
+
+async def _boom(error: BaseException):
+    raise error
+
+
+def test_child_error_surfaces_unwrapped():
+    """A failing child arrives as itself, not wrapped in an ExceptionGroup."""
+
+    async def scenario():
+        async with fan_out_group() as group:
+            group.create_task(_boom(RuntimeError("child")))
+            await asyncio.sleep(0)
+
+    with pytest.raises(RuntimeError) as exc_info:
+        asyncio.run(scenario())
+
+    assert not isinstance(exc_info.value, BaseExceptionGroup)
+    assert str(exc_info.value) == "child"
+
+
+def test_body_error_surfaces_unwrapped():
+    """An error raised by the caller's own body propagates unchanged."""
+
+    async def ok():
+        await asyncio.sleep(0)
+
+    async def scenario():
+        async with fan_out_group() as group:
+            group.create_task(ok())
+            raise ValueError("body")
+
+    with pytest.raises(ValueError) as exc_info:
+        asyncio.run(scenario())
+
+    assert not isinstance(exc_info.value, BaseExceptionGroup)
+
+
+def test_unexpected_error_is_preferred_over_an_expected_one():
+    """first_error must not let a routine network error mask a real bug."""
+
+    async def scenario():
+        async with fan_out_group() as group:
+            group.create_task(_boom(aiohttp.ClientError("routine")))
+            await asyncio.sleep(0)
+            raise ValueError("real bug")
+
+    with pytest.raises(ValueError):
+        asyncio.run(scenario())
+
+
+def test_early_return_still_awaits_children():
+    """A `return` inside the body must not abandon an in-flight child task."""
+
+    finished = []
+
+    async def child():
+        await asyncio.sleep(0.01)
+        finished.append("done")
+
+    async def scenario():
+        async with fan_out_group() as group:
+            group.create_task(child())
+            return "returned"
+
+    assert asyncio.run(scenario()) == "returned"
+    assert finished == ["done"]
+
+
+def test_parked_sibling_is_cancelled_when_the_body_fails():
+    """The binding is the point: a child that never finishes must not hang.
+
+    Without the group, a create_task + `finally: await` shape would park here
+    forever. The deadline is the assertion — a hang fails this test.
+    """
+
+    async def parked():
+        await asyncio.Event().wait()
+
+    async def scenario():
+        async with asyncio.timeout(2):
+            async with fan_out_group() as group:
+                group.create_task(parked())
+                await asyncio.sleep(0)
+                raise RuntimeError("body")
+
+    with pytest.raises(RuntimeError):
+        asyncio.run(scenario())
+
+
+def test_children_run_concurrently_with_the_body():
+    """Both siblings make progress; the group is not a sequencing device."""
+
+    order = []
+
+    async def child():
+        order.append("child-start")
+        await asyncio.sleep(0.01)
+        order.append("child-end")
+
+    async def scenario():
+        async with fan_out_group() as group:
+            group.create_task(child())
+            order.append("body")
+            await asyncio.sleep(0.02)
+
+    asyncio.run(scenario())
+    assert order == ["body", "child-start", "child-end"]