fix(backend): stop the async scanner flagging coroutines handed to asyncio

backend/scripts/scan_async_blockers.py

@@ -1,4 +1,4 @@
 #!/usr/bin/env python3
 """
 FastAPI async blocker scanner for backend/.
 
@@ -76,6 +76,16 @@
     "utils.notifications": frozenset({"send_notification"}),
 }
 
+# asyncio entry points whose arguments must already be awaitables. Handing a call to one of
+# them is the same handoff as `await call()`: the coroutine is driven by the event loop, not
+# by the calling frame. Keep this list exact — a name outside it is not assumed to await.
+COROUTINE_HANDOFF_ASYNCIO_FUNCTIONS = frozenset(
+    {"as_completed", "ensure_future", "gather", "shield", "wait", "wait_for"}
+)
+# `create_task` is matched on the attribute alone because the receiver is either the asyncio
+# module or a TaskGroup/loop instance, and all three drive the argument on the event loop.
+COROUTINE_HANDOFF_METHODS = frozenset({"create_task"})
+
 
 def _node_lineno(node: ast.AST) -> int:
     """Best-effort line number for an arbitrary AST node."""
@@ -281,6 +291,54 @@
     return awaited
 
 
+def _coroutine_handoff_call_ids(node: FunctionNode) -> Set[int]:
+    """Identities of the Call nodes handed to asyncio to be awaited on the event loop.
+
+    `await asyncio.gather(load_a(), load_b())`, `asyncio.create_task(load())` and
+    `await asyncio.wait_for(load(), timeout=5)` are the ordinary ways to await more than one
+    coroutine, or to await one with a deadline. The inner call is not the direct operand of an
+    `await`, so `_awaited_call_ids` cannot see it, yet it blocks the calling frame no more than
+    a directly awaited call does: calling an `async def` only builds a coroutine object.
+
+    Only arguments written at the call site are covered — directly, unpacked with `*`, or as
+    elements of a literal list/tuple/set, which is how `asyncio.wait` and `as_completed` take
+    their awaitables. A name passed in from elsewhere stays outside this analysis, in keeping
+    with the rest of this scanner.
+    """
+    handed_off: Set[int] = set()
+    for child in ast.walk(node):
+        if not isinstance(child, ast.Call) or not _is_coroutine_handoff(child.func):
+            continue
+        for argument in [*child.args, *(keyword.value for keyword in child.keywords)]:
+            for candidate in _handed_off_operands(argument):
+                handed_off.add(id(candidate))
+    return handed_off
+
+
+def _is_coroutine_handoff(func: ast.expr) -> bool:
+    """True for `asyncio.<awaiting entry point>` and for any `<receiver>.create_task`."""
+    if not isinstance(func, ast.Attribute):
+        return False
+    if func.attr in COROUTINE_HANDOFF_METHODS:
+        return True
+    return (
+        isinstance(func.value, ast.Name)
+        and func.value.id == "asyncio"
+        and func.attr in COROUTINE_HANDOFF_ASYNCIO_FUNCTIONS
+    )
+
+
+def _handed_off_operands(argument: ast.expr) -> Iterator[ast.Call]:
+    """Yield the calls an argument hands over: itself, `*unpacked`, or literal collection items."""
+    if isinstance(argument, ast.Starred):
+        yield from _handed_off_operands(argument.value)
+    elif isinstance(argument, (ast.List, ast.Tuple, ast.Set)):
+        for element in argument.elts:
+            yield from _handed_off_operands(element)
+    elif isinstance(argument, ast.Call):
+        yield argument
+
+
 def _scan_function_body(
     node: FunctionNode,
     db_names: Set[str],
@@ -298,7 +356,7 @@
 
     offloaded = _get_offloaded_lines(node)
     nested = _collect_nested_func_lines(node)
-    awaited = _awaited_call_ids(node)
+    awaited = _awaited_call_ids(node) | _coroutine_handoff_call_ids(node)
 
     for child in _walk_body(node):
         if not isinstance(child, ast.Call):
@@ -309,7 +367,8 @@
         # `await f()` yields to the event loop; it is the correct way to call an async
         # helper, including the async accessors in `database.*`. Counting it as blocking
         # made a correct fix unpushable and offered no way to say so — there is no
-        # allowlist or inline waiver in this scanner.
+        # allowlist or inline waiver in this scanner. The same holds when the coroutine
+        # reaches the loop through asyncio instead of a direct `await`.
         if id(child) in awaited:
             continue
         body_call_lines.add(line)

backend/tests/unit/test_scan_async_blockers.py

@@ -590,3 +590,93 @@ async def dispatcher():
     )
 
     assert len(results["async_helpers_with_blocking"]) == 1
+
+
+def test_coroutines_handed_to_asyncio_are_not_blocking(scanner, tmp_path):
+    """Awaiting through asyncio is the same handoff as `await`, so it is not blocking either.
+
+    `_awaited_call_ids` only sees a call that is the direct operand of an `await`. Awaiting two
+    accessors at once, or one with a deadline, means writing `asyncio.gather(...)`,
+    `asyncio.create_task(...)` or `asyncio.wait_for(...)`, and the inner call stops being that
+    operand — so the same async accessor the previous fix cleared was reported again the moment
+    a second one was awaited beside it.
+    """
+    for form in (
+        "await asyncio.gather(get_async_redis_client(), get_async_cache_client())",
+        "await asyncio.wait_for(get_async_redis_client(), timeout=5)",
+        "await asyncio.wait([get_async_redis_client(), get_async_cache_client()])",
+        "await asyncio.create_task(get_async_redis_client())",
+        "await asyncio.shield(get_async_redis_client())",
+    ):
+        _source_path, results = _scan_source(
+            scanner,
+            tmp_path,
+            f"""
+            import asyncio
+
+            from database.redis_db import get_async_cache_client, get_async_redis_client
+
+            async def dispatcher():
+                return {form}
+            """,
+        )
+
+        assert results["async_helpers_with_blocking"] == [], form
+
+
+def test_task_group_create_task_is_not_blocking(scanner, tmp_path):
+    """A TaskGroup drives its argument on the event loop exactly as `asyncio.create_task` does."""
+    _source_path, results = _scan_source(
+        scanner,
+        tmp_path,
+        """
+        import asyncio
+
+        from database.redis_db import get_async_redis_client
+
+        async def dispatcher():
+            async with asyncio.TaskGroup() as group:
+                group.create_task(get_async_redis_client())
+        """,
+    )
+
+    assert results["async_helpers_with_blocking"] == []
+
+
+def test_a_sync_db_call_beside_a_handoff_on_one_line_is_still_blocking(scanner, tmp_path):
+    """The skip stays keyed on the call node: only the handed-off half of a line is cleared."""
+    _source_path, results = _scan_source(
+        scanner,
+        tmp_path,
+        """
+        import asyncio
+
+        from database.redis_db import get_async_redis_client, get_redis_client
+
+        async def dispatcher():
+            return await asyncio.gather(get_async_redis_client()), get_redis_client()
+        """,
+    )
+
+    assert [call["call"] for finding in results["async_helpers_with_blocking"] for call in finding["db_calls"]] == [
+        "get_redis_client"
+    ]
+
+
+def test_a_call_asyncio_never_receives_is_still_blocking(scanner, tmp_path):
+    """The guard must not widen: importing asyncio near a sync DB call does not clear it."""
+    _source_path, results = _scan_source(
+        scanner,
+        tmp_path,
+        """
+        import asyncio
+
+        from database.redis_db import get_redis_client
+
+        async def dispatcher():
+            await asyncio.sleep(1)
+            return get_redis_client()
+        """,
+    )
+
+    assert len(results["async_helpers_with_blocking"]) == 1