Avoid asyncio-dangling-task violations on shadowed bindings (#9215)

## Summary

Ensures that we avoid flagging cases like:

```python
async def f(x: int):
    if x > 0:
        task = asyncio.create_task(make_request())
    else:
        task = asyncio.create_task(make_request())
    await task
```

Closes #9133.

crates/ruff_linter/resources/test/fixtures/ruff/RUF006.py

@@ -122,3 +122,33 @@ async def f():
 # OK
 async def f():
     task[i] = asyncio.create_task(coordinator.ws_connect())
+
+
+# OK
+async def f(x: int):
+    if x > 0:
+        task = asyncio.create_task(make_request())
+    else:
+        task = asyncio.create_task(make_request())
+    await task
+
+
+# OK
+async def f(x: bool):
+    if x:
+        t = asyncio.create_task(asyncio.sleep(1))
+    else:
+        t = None
+    try:
+        await asyncio.sleep(1)
+    finally:
+        if t:
+            await t
+
+
+# Error
+async def f(x: bool):
+    if x:
+        t = asyncio.create_task(asyncio.sleep(1))
+    else:
+        t = None

crates/ruff_linter/src/checkers/ast/analyze/bindings.rs

@@ -3,12 +3,11 @@ use ruff_text_size::Ranged;
 
 use crate::checkers::ast::Checker;
 use crate::codes::Rule;
-use crate::rules::{flake8_import_conventions, flake8_pyi, pyflakes, pylint, ruff};
+use crate::rules::{flake8_import_conventions, flake8_pyi, pyflakes, pylint};
 
 /// Run lint rules over the [`Binding`]s.
 pub(crate) fn bindings(checker: &mut Checker) {
     if !checker.any_enabled(&[
-        Rule::AsyncioDanglingTask,
         Rule::InvalidAllFormat,
         Rule::InvalidAllObject,
         Rule::NonAsciiName,
@@ -72,12 +71,5 @@ pub(crate) fn bindings(checker: &mut Checker) {
                 checker.diagnostics.push(diagnostic);
             }
         }
-        if checker.enabled(Rule::AsyncioDanglingTask) {
-            if let Some(diagnostic) =
-                ruff::rules::asyncio_dangling_binding(binding, &checker.semantic)
-            {
-                checker.diagnostics.push(diagnostic);
-            }
-        }
     }
 }

crates/ruff_linter/src/checkers/ast/analyze/deferred_scopes.rs

@@ -5,13 +5,17 @@ use ruff_text_size::Ranged;
 
 use crate::checkers::ast::Checker;
 use crate::codes::Rule;
-use crate::rules::{flake8_pyi, flake8_type_checking, flake8_unused_arguments, pyflakes, pylint};
+use crate::rules::{
+    flake8_pyi, flake8_type_checking, flake8_unused_arguments, pyflakes, pylint, ruff,
+};
 
 /// Run lint rules over all deferred scopes in the [`SemanticModel`].
 pub(crate) fn deferred_scopes(checker: &mut Checker) {
     if !checker.any_enabled(&[
+        Rule::AsyncioDanglingTask,
         Rule::GlobalVariableNotAssigned,
         Rule::ImportShadowedByLoopVar,
+        Rule::NoSelfUse,
         Rule::RedefinedArgumentFromLocal,
         Rule::RedefinedWhileUnused,
         Rule::RuntimeImportInTypeCheckingBlock,
@@ -32,7 +36,6 @@ pub(crate) fn deferred_scopes(checker: &mut Checker) {
         Rule::UnusedPrivateTypedDict,
         Rule::UnusedStaticMethodArgument,
         Rule::UnusedVariable,
-        Rule::NoSelfUse,
     ]) {
         return;
     }
@@ -270,6 +273,10 @@ pub(crate) fn deferred_scopes(checker: &mut Checker) {
             flake8_pyi::rules::unused_private_typed_dict(checker, scope, &mut diagnostics);
         }
 
+        if checker.enabled(Rule::AsyncioDanglingTask) {
+            ruff::rules::asyncio_dangling_binding(scope, &checker.semantic, &mut diagnostics);
+        }
+
         if matches!(scope.kind, ScopeKind::Function(_) | ScopeKind::Lambda(_)) {
             if checker.enabled(Rule::UnusedVariable) {
                 pyflakes::rules::unused_variable(checker, scope, &mut diagnostics);

crates/ruff_linter/src/rules/ruff/rules/asyncio_dangling_task.rs

@@ -4,7 +4,7 @@ use ast::Stmt;
 use ruff_diagnostics::{Diagnostic, Violation};
 use ruff_macros::{derive_message_formats, violation};
 use ruff_python_ast::{self as ast, Expr};
-use ruff_python_semantic::{analyze::typing, Binding, SemanticModel};
+use ruff_python_semantic::{analyze::typing, Scope, SemanticModel};
 use ruff_text_size::Ranged;
 
 /// ## What it does
@@ -105,22 +105,50 @@ pub(crate) fn asyncio_dangling_task(expr: &Expr, semantic: &SemanticModel) -> Op
 
 /// RUF006
 pub(crate) fn asyncio_dangling_binding(
-    binding: &Binding,
+    scope: &Scope,
     semantic: &SemanticModel,
-) -> Option<Diagnostic> {
-    if binding.is_used() || !binding.kind.is_assignment() {
-        return None;
-    }
+    diagnostics: &mut Vec<Diagnostic>,
+) {
+    for binding_id in scope.binding_ids() {
+        // If the binding itself is used, or it's not an assignment, skip it.
+        let binding = semantic.binding(binding_id);
+        if binding.is_used() || !binding.kind.is_assignment() {
+            continue;
+        }
+
+        // Otherwise, any dangling tasks, including those that are shadowed, as in:
+        // ```python
+        // if x > 0:
+        //     task = asyncio.create_task(make_request())
+        // else:
+        //     task = asyncio.create_task(make_request())
+        // ```
+        for binding_id in
+            std::iter::successors(Some(binding_id), |id| semantic.shadowed_binding(*id))
+        {
+            let binding = semantic.binding(binding_id);
+            if binding.is_used() || !binding.kind.is_assignment() {
+                continue;
+            }
 
-    let source = binding.source?;
-    match semantic.statement(source) {
-        Stmt::Assign(ast::StmtAssign { value, targets, .. }) if targets.len() == 1 => {
-            asyncio_dangling_task(value, semantic)
+            let Some(source) = binding.source else {
+                continue;
+            };
+
+            let diagnostic = match semantic.statement(source) {
+                Stmt::Assign(ast::StmtAssign { value, targets, .. }) if targets.len() == 1 => {
+                    asyncio_dangling_task(value, semantic)
+                }
+                Stmt::AnnAssign(ast::StmtAnnAssign {
+                    value: Some(value), ..
+                }) => asyncio_dangling_task(value, semantic),
+                _ => None,
+            };
+
+            if let Some(diagnostic) = diagnostic {
+                diagnostics.push(diagnostic);
+            }
         }
-        Stmt::AnnAssign(ast::StmtAnnAssign {
-            value: Some(value), ..
-        }) => asyncio_dangling_task(value, semantic),
-        _ => None,
     }
 }