async_lint.py

#!/usr/bin/env python3
"""Lint asyncio code: blocking calls, lost tasks, swallowed cancellation, unbounded fan-out.

Static analysis only: the analysed code is parsed with `ast`, never imported
or executed. Rules (run --list-rules for the why and a fix snippet of each):

  AIO001 blocking-sleep            time.sleep() inside async def
  AIO002 blocking-http             requests / urllib / sync httpx / sync OpenAI-style client in async def
  AIO003 blocking-subprocess       subprocess.run/call/check_output, os.system in async def
  AIO004 blocking-file-io          open(), Path.read_text/write_text, shutil copies in async def
  AIO005 blocking-db               sqlite3 / psycopg2 / pymysql / redis.Redis / sync vector-store client
  AIO006 blocking-socket-or-input  input(), blocking socket recv/accept/connect, DNS lookups
  AIO007 coroutine-not-awaited     coroutine called as a statement and never awaited
  AIO008 dangling-task             create_task / ensure_future result discarded (task can be GC'd)
  AIO009 nested-asyncio-run        asyncio.run() inside async def
  AIO010 get-event-loop            asyncio.get_event_loop() (deprecated pattern)
  AIO011 run-until-complete-in-async loop.run_until_complete() inside async def
  AIO012 unbounded-gather          gather(*[f(x) for x in items]) with no concurrency limit
  AIO013 sequential-await-loop     `for x in items: r = await f(x)` over independent items
  AIO014 cancellation-swallowed    bare except / BaseException / CancelledError not re-raised
  AIO015 wait-for-on-311           asyncio.wait_for when --target-version >= 3.11
  AIO016 thread-primitive-in-async threading.Lock / queue.Queue / threading.Event used in async def
  AIO017 async-without-await       async def that never awaits
  AIO018 sleep-zero-busy-wait      `while ...: await asyncio.sleep(0)` polling loop
  AIO019 client-without-async-with AsyncClient / ClientSession / aiofiles.open never closed
  AIO020 task-never-awaited        task stored in a variable that is never used again
  AIO021 unbounded-queue           asyncio.Queue() without maxsize used as a pipeline buffer
  AIO022 return-exceptions-unchecked gather(..., return_exceptions=True) results never checked
  AIO023 sync-route-asyncio-run    sync FastAPI/Starlette route that calls asyncio.run
  AIO024 version-feature           asyncio API newer than --target-version (TaskGroup, timeout, ...)

Suppress a finding on its line with `# noqa` or `# noqa: AIO001,AIO013`.

Usage:
    python async_lint.py app/llm_client.py
    python async_lint.py src/ tests/ --json
    cat worker.py | python async_lint.py -
    python async_lint.py src/ --strict --target-version 3.9
    python async_lint.py src/ --select AIO001,AIO008 --ignore sequential-await-loop
    python async_lint.py --list-rules

Exit codes: 0 no findings (or findings without --strict), 1 findings with
--strict, 2 usage error, or a file named on the command line that cannot be
read or parsed. Files found by walking a directory that cannot be parsed are
reported as warnings and skipped.

No dependencies (standard library only, Python 3.9+).
"""

import argparse
import ast
import json
import os
import re
import sys

# ---------------------------------------------------------------- rules

RULES = {
    "AIO001": {
        "name": "blocking-sleep", "severity": "high",
        "summary": "time.sleep() inside async def",
        "why": "time.sleep blocks the event loop thread, so every other task (in-flight LLM calls, the web "
               "server, heartbeats) freezes for the whole delay.",
        "fix": "await asyncio.sleep(delay)",
    },
    "AIO002": {
        "name": "blocking-http", "severity": "high",
        "summary": "synchronous HTTP (requests, urllib, httpx.Client, sync OpenAI/Anthropic client) in async def",
        "why": "A sync HTTP call holds the loop for the full network round trip; 100 concurrent LLM calls "
               "become 100 sequential ones and the server stops answering health checks.",
        "fix": "async with httpx.AsyncClient(timeout=30) as http:   # create once, reuse\n"
               "    resp = await http.get(url)\n"
               "# SDKs: openai.AsyncOpenAI / anthropic.AsyncAnthropic, then `await client....create(...)`\n"
               "# no async client available: resp = await asyncio.to_thread(requests.get, url, timeout=30)",
    },
    "AIO003": {
        "name": "blocking-subprocess", "severity": "high",
        "summary": "subprocess.run/call/check_output or os.system in async def",
        "why": "subprocess.run waits for the child process on the loop thread; nothing else runs until it exits.",
        "fix": "proc = await asyncio.create_subprocess_exec(\"ffmpeg\", \"-i\", src, dst,\n"
               "    stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)\n"
               "out, err = await proc.communicate()\n"
               "# or: await asyncio.to_thread(subprocess.run, cmd, check=True)",
    },
    "AIO004": {
        "name": "blocking-file-io", "severity": "medium",
        "summary": "open() / Path.read_text() / shutil copy inside async def",
        "why": "Disk I/O is blocking (and slow on network filesystems / cold caches); large reads stall every "
               "task. Small config reads at startup are fine - suppress with # noqa: AIO004.",
        "fix": "text = await asyncio.to_thread(path.read_text, encoding=\"utf-8\")\n"
               "# many files / streaming: async with aiofiles.open(path) as f: data = await f.read()",
    },
    "AIO005": {
        "name": "blocking-db", "severity": "high",
        "summary": "sync DB / vector-store client (sqlite3, psycopg2, pymysql, redis.Redis, QdrantClient...) in async def",
        "why": "Every query blocks the loop for its full latency; under load p99 latency of unrelated requests explodes.",
        "fix": "# use the async driver: asyncpg / psycopg.AsyncConnection / aiosqlite / redis.asyncio /\n"
               "# AsyncQdrantClient, created once (FastAPI lifespan) and awaited:\n"
               "rows = await pool.fetch(\"SELECT id FROM docs WHERE tenant = $1\", tenant)\n"
               "# no async driver: rows = await asyncio.to_thread(run_query, sql, params)",
    },
    "AIO006": {
        "name": "blocking-socket-or-input", "severity": "high",
        "summary": "input(), blocking socket calls or DNS lookups in async def",
        "why": "input() and blocking socket recv/accept/connect park the loop thread until data arrives.",
        "fix": "reader, writer = await asyncio.open_connection(host, port)\n"
               "data = await reader.read(4096)\n"
               "# console input: line = await asyncio.to_thread(input, \"> \")",
    },
    "AIO007": {
        "name": "coroutine-not-awaited", "severity": "high",
        "summary": "coroutine function called as a statement without await",
        "why": "Calling an async def only creates a coroutine object; the body never runs and Python prints "
               "'RuntimeWarning: coroutine ... was never awaited' (often hidden in logs).",
        "fix": "await embed_batch(batch)\n"
               "# run it concurrently instead: task = asyncio.create_task(embed_batch(batch)) and await it later",
    },
    "AIO008": {
        "name": "dangling-task", "severity": "high",
        "summary": "create_task / ensure_future result discarded",
        "why": "The loop keeps only a weak reference to tasks: an unreferenced task can be garbage-collected "
               "mid-flight, and its exception is never retrieved (only 'Task exception was never retrieved' at exit).",
        "fix": "background: set[asyncio.Task] = set()\n"
               "task = asyncio.create_task(log_usage(event))\n"
               "background.add(task)\n"
               "task.add_done_callback(background.discard)\n"
               "# better (3.11+): create it inside `async with asyncio.TaskGroup() as tg:`",
    },
    "AIO009": {
        "name": "nested-asyncio-run", "severity": "high",
        "summary": "asyncio.run() inside async def",
        "why": "asyncio.run cannot be called while a loop is running in the thread: "
               "'RuntimeError: asyncio.run() cannot be called from a running event loop'.",
        "fix": "result = await main_coro()          # already inside a loop: just await it\n"
               "# Jupyter: top-level `await main_coro()` works in a cell (the kernel owns the loop)",
    },
    "AIO010": {
        "name": "get-event-loop", "severity": "medium",
        "summary": "asyncio.get_event_loop()",
        "why": "Outside a running loop it is deprecated (DeprecationWarning in several 3.10-3.13 releases) and "
               "raises RuntimeError in 3.14 when no loop is set; inside a coroutine get_running_loop() states the "
               "intent and never creates a stray loop.",
        "fix": "loop = asyncio.get_running_loop()   # inside async code\n"
               "asyncio.run(main())                 # at the entry point instead of get_event_loop().run_until_complete",
    },
    "AIO011": {
        "name": "run-until-complete-in-async", "severity": "high",
        "summary": "loop.run_until_complete() inside async def",
        "why": "The loop is already running: 'RuntimeError: This event loop is already running'.",
        "fix": "result = await coro()",
    },
    "AIO012": {
        "name": "unbounded-gather", "severity": "medium",
        "summary": "gather over a comprehension of calls with no concurrency limit",
        "why": "gather starts every coroutine at once: 10k prompts means 10k sockets, instant 429s from the "
               "provider, file-descriptor exhaustion and memory spikes. Fine when the input is known to be small.",
        "fix": "sem = asyncio.Semaphore(16)\n\n"
               "async def bounded(doc):\n"
               "    async with sem:\n"
               "        return await embed(doc)\n\n"
               "results = await asyncio.gather(*(bounded(d) for d in docs))",
    },
    "AIO013": {
        "name": "sequential-await-loop", "severity": "low",
        "summary": "await inside a for loop over independent items",
        "why": "Each iteration waits for the previous one, so N calls take N x latency. Only applies when the "
               "iterations are truly independent (no ordering requirement, no shared state, no rate-limit pacing).",
        "fix": "async with asyncio.TaskGroup() as tg:          # 3.11+\n"
               "    tasks = [tg.create_task(summarize(doc)) for doc in docs]\n"
               "summaries = [t.result() for t in tasks]\n"
               "# add a Semaphore when the list can be large",
    },
    "AIO014": {
        "name": "cancellation-swallowed", "severity": "high",
        "summary": "bare except / except BaseException / except CancelledError without re-raise in async def",
        "why": "CancelledError is a BaseException (3.8+). Swallowing it makes the task uncancellable: timeouts "
               "stop working, TaskGroup and shutdown hang, Ctrl+C does nothing.",
        "fix": "except asyncio.CancelledError:\n"
               "    await stream.aclose()   # cleanup only\n"
               "    raise",
    },
    "AIO015": {
        "name": "wait-for-on-311", "severity": "low",
        "summary": "asyncio.wait_for in 3.11+ code",
        "why": "asyncio.timeout() (3.11) scopes a deadline over a whole block, needs no extra task and composes "
               "with TaskGroup; wait_for wraps a single awaitable.",
        "fix": "async with asyncio.timeout(30):\n"
               "    resp = await client.chat.completions.create(**req)",
    },
    "AIO016": {
        "name": "thread-primitive-in-async", "severity": "medium",
        "summary": "threading.Lock / queue.Queue / threading.Event used inside async def",
        "why": "Thread primitives block the loop thread while waiting, and a threading.Lock held across an "
               "await deadlocks when another task on the same loop tries to take it.",
        "fix": "lock = asyncio.Lock()          # async with lock: ...\n"
               "queue = asyncio.Queue(maxsize=100)\n"
               "event = asyncio.Event()        # await event.wait()\n"
               "# talking to a worker thread: loop.call_soon_threadsafe / asyncio.run_coroutine_threadsafe",
    },
    "AIO017": {
        "name": "async-without-await", "severity": "low",
        "summary": "async def that never awaits",
        "why": "It runs synchronously on the loop anyway, so the async keyword adds overhead and misleads "
               "readers; if it does blocking work it should be sync and run via to_thread. Fine when it must "
               "match an async interface.",
        "fix": "def build_prompt(question: str, chunks: list[str]) -> str:   # make it a plain function\n"
               "    ...",
    },
    "AIO018": {
        "name": "sleep-zero-busy-wait", "severity": "medium",
        "summary": "polling loop whose only await is asyncio.sleep(0) / a tiny sleep",
        "why": "The loop spins at 100% CPU re-checking a flag and starves other tasks of fair scheduling.",
        "fix": "done = asyncio.Event()\n"
               "...\n"
               "await done.wait()        # the producer calls done.set()\n"
               "# or await queue.get() / await asyncio.wait_for(...) / async with asyncio.timeout(...)",
    },
    "AIO019": {
        "name": "client-without-async-with", "severity": "medium",
        "summary": "httpx.AsyncClient / aiohttp.ClientSession / aiofiles.open never closed",
        "why": "Unclosed clients leak connection pools and sockets ('Unclosed client session', ResourceWarning); "
               "creating one per request also throws away keep-alive and TLS reuse.",
        "fix": "async with httpx.AsyncClient(timeout=30) as http:\n"
               "    resp = await http.post(url, json=payload)\n"
               "# app-lifetime client: create in FastAPI lifespan, `await client.aclose()` after `yield`",
    },
    "AIO020": {
        "name": "task-never-awaited", "severity": "medium",
        "summary": "task assigned to a variable that is never awaited, gathered or stored",
        "why": "If the task fails, the exception is never retrieved; when the function returns the only "
               "reference disappears and the task may be garbage-collected or outlive its owner.",
        "fix": "task = asyncio.create_task(fetch_context(q))\n"
               "...\n"
               "context = await task      # or gather / TaskGroup / keep it in a set with a done callback",
    },
    "AIO021": {
        "name": "unbounded-queue", "severity": "low",
        "summary": "asyncio.Queue() without maxsize used as a pipeline buffer",
        "why": "With no maxsize a fast producer (file reader, scraper) never waits for slow consumers "
               "(embedding, GPU) and memory grows without limit: no backpressure.",
        "fix": "queue: asyncio.Queue[Doc] = asyncio.Queue(maxsize=2 * n_workers)\n"
               "await queue.put(doc)      # now blocks when consumers fall behind",
    },
    "AIO022": {
        "name": "return-exceptions-unchecked", "severity": "medium",
        "summary": "gather(..., return_exceptions=True) results never checked for exceptions",
        "why": "With return_exceptions=True failures come back as values in the result list; unless you check "
               "them they are silently written to the vector store / eval report as if they were results.",
        "fix": "results = await asyncio.gather(*coros, return_exceptions=True)\n"
               "ok = [r for r in results if not isinstance(r, BaseException)]\n"
               "failed = [(x, r) for x, r in zip(items, results) if isinstance(r, BaseException)]\n"
               "for item, err in failed:\n"
               "    logger.error(\"item %s failed\", item.id, exc_info=err)",
    },
    "AIO023": {
        "name": "sync-route-asyncio-run", "severity": "high",
        "summary": "sync def route (FastAPI/Starlette/Flask-style decorator) calls asyncio.run",
        "why": "Each request spins up and tears down a whole event loop in a threadpool thread, async clients "
               "created in lifespan cannot be used from it, and it breaks if the route is ever made async.",
        "fix": "@app.post(\"/summarize\")\n"
               "async def summarize(req: SummarizeRequest) -> SummarizeResponse:\n"
               "    text = await llm_summarize(req.text)\n"
               "    return SummarizeResponse(summary=text)",
    },
    "AIO024": {
        "name": "version-feature", "severity": "high",
        "summary": "asyncio API newer than --target-version",
        "why": "The code fails with AttributeError / SyntaxError on the Python version you said you support.",
        "fix": "# raise --target-version, or use the older equivalent:\n"
               "# TaskGroup -> gather (+ cancel siblings on error); asyncio.timeout -> asyncio.wait_for;\n"
               "# Runner -> asyncio.run; eager_task_factory -> plain create_task",
    },
}
SEVERITY_ORDER = {"high": 0, "medium": 1, "low": 2}
BLOCKING_RULES = {"AIO001", "AIO002", "AIO003", "AIO004", "AIO005", "AIO006", "AIO016"}

# ---------------------------------------------------------------- knowledge

HTTP_VERBS = {"get", "post", "put", "patch", "delete", "head", "options", "request", "stream"}
BLOCKING_CALLS = {
    "time.sleep": "AIO001",
    "urllib.request.urlopen": "AIO002", "urllib.request.urlretrieve": "AIO002",
    "urllib3.request": "AIO002",
    "subprocess.run": "AIO003", "subprocess.call": "AIO003", "subprocess.check_call": "AIO003",
    "subprocess.check_output": "AIO003", "subprocess.getoutput": "AIO003",
    "subprocess.getstatusoutput": "AIO003", "os.system": "AIO003", "os.popen": "AIO003",
    "os.waitpid": "AIO003",
    "shutil.copy": "AIO004", "shutil.copy2": "AIO004", "shutil.copyfile": "AIO004",
    "shutil.copytree": "AIO004", "shutil.rmtree": "AIO004", "shutil.move": "AIO004",
    "socket.gethostbyname": "AIO006", "socket.getaddrinfo": "AIO006", "socket.create_connection": "AIO006",
}
# constructors of synchronous clients -> rule used when they (or methods on them) appear in async def
SYNC_CLIENTS = {
    "requests.Session": "AIO002", "requests.session": "AIO002", "httpx.Client": "AIO002",
    "urllib3.PoolManager": "AIO002", "openai.OpenAI": "AIO002", "openai.AzureOpenAI": "AIO002",
    "anthropic.Anthropic": "AIO002", "cohere.Client": "AIO002", "cohere.ClientV2": "AIO002",
    "sqlite3.connect": "AIO005", "psycopg2.connect": "AIO005", "psycopg.connect": "AIO005",
    "pymysql.connect": "AIO005", "mysql.connector.connect": "AIO005", "pymongo.MongoClient": "AIO005",
    "redis.Redis": "AIO005", "redis.StrictRedis": "AIO005", "redis.from_url": "AIO005",
    "redis.Redis.from_url": "AIO005", "psycopg_pool.ConnectionPool": "AIO005",
    "qdrant_client.QdrantClient": "AIO005", "chromadb.Client": "AIO005",
    "chromadb.PersistentClient": "AIO005", "chromadb.HttpClient": "AIO005",
    "elasticsearch.Elasticsearch": "AIO005", "pinecone.Pinecone": "AIO005",
}
SOCKET_CTORS = {"socket.socket", "socket.socketpair", "socket.fromfd"}
SYNC_HTTP_METHODS = HTTP_VERBS | {"send", "create", "parse", "retrieve", "list", "update", "upload", "urlopen"}
NONBLOCKING_METHODS = {"setblocking", "settimeout", "setsockopt", "fileno", "getsockname", "getpeername",
                       "cursor", "with_options", "copy"}
SOCKET_BLOCKING = {"recv", "recv_into", "recvfrom", "accept", "sendall", "connect"}
PATH_IO = {"read_text", "write_text", "read_bytes", "write_bytes"}
THREAD_LOCKS = {"threading.Lock", "threading.RLock", "threading.Semaphore", "threading.BoundedSemaphore",
                "threading.Condition"}
THREAD_PRIMS = THREAD_LOCKS | {"threading.Event", "threading.Barrier", "queue.Queue", "queue.LifoQueue",
                               "queue.PriorityQueue", "queue.SimpleQueue", "multiprocessing.Queue"}
THREAD_BLOCKING_METHODS = {"acquire", "wait", "get", "join"}
ASYNC_QUEUES = {"asyncio.Queue", "asyncio.PriorityQueue", "asyncio.LifoQueue"}
ASYNC_PRIM_METHODS = {
    "asyncio.Queue": {"put", "get", "join"}, "asyncio.PriorityQueue": {"put", "get", "join"},
    "asyncio.LifoQueue": {"put", "get", "join"}, "asyncio.Event": {"wait"},
    "asyncio.Lock": {"acquire"}, "asyncio.Semaphore": {"acquire"}, "asyncio.BoundedSemaphore": {"acquire"},
    "asyncio.Condition": {"acquire", "wait", "wait_for"},
}
ASYNCIO_COROS = {"asyncio.sleep", "asyncio.wait_for", "asyncio.wait", "asyncio.gather", "asyncio.to_thread",
                 "asyncio.open_connection", "asyncio.start_server", "asyncio.open_unix_connection",
                 "asyncio.start_unix_server", "asyncio.create_subprocess_exec",
                 "asyncio.create_subprocess_shell"}
ALWAYS_ASYNC_ATTRS = {"aclose", "asend", "athrow", "__aenter__", "__aexit__", "__anext__"}
LOOP_APIS = {"create_task", "ensure_future", "get_running_loop", "current_task", "all_tasks", "call_soon",
             "call_later", "call_at", "run_in_executor", "create_future", "call_soon_threadsafe"}
ASYNC_RESOURCES = {"httpx.AsyncClient", "aiohttp.ClientSession", "aiofiles.open", "openai.AsyncOpenAI",
                   "openai.AsyncAzureOpenAI", "anthropic.AsyncAnthropic"}
ORDERED_METHODS = {"put", "put_nowait", "get", "send", "send_json", "send_text", "send_bytes", "write",
                   "drain", "commit", "rollback", "execute", "executemany", "flush", "sleep", "wait", "acquire",
                   "release", "join", "aclose", "close", "fetch", "fetchrow", "fetchval", "fetchone",
                   "fetchall", "ack", "nack", "publish", "emit", "readline", "read", "readexactly", "xadd"}
ROUTE_ATTRS = {"get", "post", "put", "patch", "delete", "head", "options", "api_route", "route", "websocket"}
INTERFACE_HINTS = ("callback", "handler", "hook", "mock", "fake", "stub", "noop", "dummy", "default_")
FANOUT_SAFE_HINTS = ("handler", "callback", "manager", "listener", "subscriber", "hook", "workers", "shards")
SEM_HINTS = ("sem", "limit", "throttle", "bucket", "rate", "pool", "slots")
VERSION_FEATURES = {
    "asyncio.TaskGroup": (3, 11), "asyncio.timeout": (3, 11), "asyncio.timeout_at": (3, 11),
    "asyncio.Timeout": (3, 11), "asyncio.Runner": (3, 11), "asyncio.Barrier": (3, 11),
    "asyncio.BrokenBarrierError": (3, 11),
    "asyncio.eager_task_factory": (3, 12), "asyncio.create_eager_task_factory": (3, 12),
    "asyncio.QueueShutDown": (3, 13), "asyncio.EventLoop": (3, 13),
}
EXIT_CALLS = {"sys.exit", "exit", "quit", "os._exit"}
TRY_TYPES = (ast.Try,) + ((ast.TryStar,) if hasattr(ast, "TryStar") else ())
SCOPES = (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda)


# ---------------------------------------------------------------- helpers

def dotted(node):
    """Return 'a.b.c' for Name/Attribute chains, else ''."""
    parts = []
    while isinstance(node, ast.Attribute):
        parts.append(node.attr)
        node = node.value
    if isinstance(node, ast.Name):
        parts.append(node.id)
        return ".".join(reversed(parts))
    return ""


def walk_local(stmts):
    """ast.walk over statements without descending into nested functions, classes or lambdas."""
    stack = list(stmts)
    while stack:
        node = stack.pop()
        yield node
        if isinstance(node, SCOPES):
            continue  # a nested def/class/lambda is its own scope
        for child in ast.iter_child_nodes(node):
            stack.append(child)


def parent(node):
    return getattr(node, "_aio_parent", None)


def reraises(stmts):
    for n in walk_local(stmts):
        if isinstance(n, ast.Raise):
            return True
        if isinstance(n, ast.Call) and dotted(n.func) in EXIT_CALLS:
            return True
    return False


CAPTURE_ATTRS = {"set_exception", "append", "add", "put_nowait", "throw", "cancel", "fail"}


def captures(stmts, name):
    """True if the caught exception is stored or handed on (err = e; fut.set_exception(e)) to be re-raised later."""
    if not name:
        return False
    for n in walk_local(stmts):
        if isinstance(n, (ast.Assign, ast.AnnAssign, ast.AugAssign)) and n.value is not None \
                and name in names_in(n.value):
            return True
        if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) and n.func.attr in CAPTURE_ATTRS \
                and any(name in names_in(a) for a in n.args):
            return True
    return False


def is_stub(body):
    for s in body:
        if isinstance(s, ast.Pass):
            continue
        if isinstance(s, ast.Expr) and isinstance(s.value, ast.Constant):
            continue
        if isinstance(s, ast.Raise):
            continue
        return False
    return True


def names_in(node, ctx=ast.Load):
    return {n.id for n in ast.walk(node) if isinstance(n, ast.Name) and isinstance(n.ctx, ctx)}


def small_literal_iter(it):
    if isinstance(it, (ast.List, ast.Tuple, ast.Set)):
        return True
    if isinstance(it, ast.Call) and dotted(it.func) == "range" and it.args:
        last = it.args[0] if len(it.args) == 1 else it.args[1]
        return isinstance(last, ast.Constant) and isinstance(last.value, int) and last.value <= 16
    if isinstance(it, ast.Subscript) and isinstance(it.slice, ast.Slice):
        up = it.slice.upper
        return isinstance(up, ast.Constant) and isinstance(up.value, int) and 0 <= up.value <= 16
    return False


def has_limit_hint(node):
    """True if the subtree shows a concurrency limit (Semaphore, limiter, acquire())."""
    for n in ast.walk(node):
        if isinstance(n, ast.Call):
            d = dotted(n.func)
            last = d.rsplit(".", 1)[-1]
            if last in ("Semaphore", "BoundedSemaphore", "CapacityLimiter", "Limiter", "AsyncLimiter") \
                    or last == "acquire":
                return True
            if last in ("Queue", "PriorityQueue", "LifoQueue") and (n.args or any(k.arg == "maxsize"
                                                                                  for k in n.keywords)):
                return True  # bounded queue: producers get backpressure
        if isinstance(n, ast.withitem):
            expr = n.context_expr
            d = dotted(expr.func if isinstance(expr, ast.Call) else expr).lower()
            if any(h in d for h in SEM_HINTS):
                return True
    return False


def fmt_version(v):
    return "%d.%d" % v


# ---------------------------------------------------------------- analyzer

class Analyzer:
    def __init__(self, path, source, tree, target):
        self.path = path
        self.lines = source.splitlines()
        self.tree = tree
        self.target = target
        self.findings = []
        self.seen = set()
        self.imports = {}
        self.module_funcs = {}       # name -> def node (module level)
        self.async_funcs, self.sync_funcs = set(), set()
        self.global_bindings, self.attr_bindings = {}, {}
        self.local_bindings, self.shadowed = {}, set()
        self.awaited_put = set()       # receivers of `await X.put(...)`: pipeline producers
        self.value_names = set()       # names loaded other than as the callee of a call
        self.nonblocking_sockets = False  # module calls setblocking(False) / settimeout(0)
        self.close_targets = set()   # dotted receivers of .aclose() / .close()

    # ------------------------------------------------------------ indexing
    def index(self):
        for node in ast.walk(self.tree):
            for child in ast.iter_child_nodes(node):
                child._aio_parent = node
            if isinstance(node, ast.Import):
                for a in node.names:
                    if a.asname:
                        self.imports[a.asname] = a.name
                    else:
                        head = a.name.split(".")[0]
                        self.imports[head] = head
            elif isinstance(node, ast.ImportFrom):
                mod = ("." * node.level) + (node.module or "")
                for a in node.names:
                    if a.name != "*":
                        self.imports[a.asname or a.name] = mod + "." + a.name
        for stmt in self._module_level_defs(self.tree.body):
            self.module_funcs[stmt.name] = stmt
            (self.async_funcs if isinstance(stmt, ast.AsyncFunctionDef) else self.sync_funcs).add(stmt.name)
        self.global_bindings = self.collect_bindings(walk_local(self.tree.body), plain=True)
        self.attr_bindings = self.collect_bindings(ast.walk(self.tree), plain=False)
        for node in ast.walk(self.tree):
            if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load):
                p = parent(node)
                if not (isinstance(p, ast.Call) and p.func is node):
                    self.value_names.add(node.id)
            if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
                if node.func.attr == "put" and isinstance(parent(node), ast.Await):
                    self.awaited_put.add(dotted(node.func.value))
                if node.func.attr in ("setblocking", "settimeout") and node.args \
                        and isinstance(node.args[0], ast.Constant) and node.args[0].value in (False, 0):
                    self.nonblocking_sockets = True
                if node.func.attr in ("aclose", "close"):
                    self.close_targets.add(dotted(node.func.value))

    def collect_bindings(self, nodes, plain):
        """Map variable keys to the constructor they were assigned from; ambiguous keys are dropped."""
        out, conflict = {}, set()

        def bind(key, ctor):
            if not key or ("." in key) == plain:
                return
            if key in out and out[key] != ctor:
                conflict.add(key)
            out[key] = ctor

        for node in nodes:
            if isinstance(node, (ast.Assign, ast.AnnAssign)) and node.value is not None:
                ctor = self.resolve(node.value.func) if isinstance(node.value, ast.Call) else ""
                targets = node.targets if isinstance(node, ast.Assign) else [node.target]
                for t in targets:
                    bind(dotted(t), ctor or "?")
            elif isinstance(node, ast.withitem) and node.optional_vars is not None:
                expr = node.context_expr
                ctor = self.resolve(expr.func) if isinstance(expr, ast.Call) else ""
                bind(dotted(node.optional_vars), ctor or "?")
        return {k: v for k, v in out.items() if k not in conflict and v != "?"}

    def lookup(self, key):
        """Constructor a receiver was bound to, respecting function-local shadowing."""
        if "." in key:
            return self.attr_bindings.get(key)
        if key in self.local_bindings:
            return self.local_bindings[key]
        if key in self.shadowed:
            return None
        return self.global_bindings.get(key)

    def _module_level_defs(self, body):
        for stmt in body:
            if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef)):
                yield stmt
            elif isinstance(stmt, (ast.If,) + TRY_TYPES):
                for field in ("body", "orelse", "finalbody"):
                    yield from self._module_level_defs(getattr(stmt, field, []) or [])

    def resolve(self, node):
        """Fully-qualified dotted name when the head of the chain is an imported name, else ''."""
        d = dotted(node)
        if not d:
            return ""
        head, _, rest = d.partition(".")
        if head not in self.imports:
            return ""
        base = self.imports[head]
        return base + ("." + rest if rest else "")

    def is_builtin(self, node, name):
        return isinstance(node, ast.Name) and node.id == name and name not in self.imports \
            and name not in self.module_funcs

    # ------------------------------------------------------------ reporting
    def add(self, rule, node, message, fix=None):
        line = getattr(node, "lineno", 1)
        if (rule, line) in self.seen:
            return
        self.seen.add((rule, line))
        src = self.lines[line - 1] if 0 < line <= len(self.lines) else ""
        m = re.search(r"#\s*noqa(?::\s*([A-Za-z0-9_, ]+))?", src)
        if m and (not m.group(1) or rule in [c.strip().upper() for c in m.group(1).split(",")]):
            return
        info = RULES[rule]
        self.findings.append({
            "path": self.path, "line": line, "col": getattr(node, "col_offset", 0) + 1,
            "rule": rule, "name": info["name"], "severity": info["severity"], "message": message,
            "code": src.strip(), "why": info["why"], "fix": fix or info["fix"],
        })

    # ------------------------------------------------------------ driver
    def run(self):
        self.index()
        self.scope(self.tree.body, None)
        for node in ast.walk(self.tree):
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                self.scope(node.body, node)
        self.check_versions()
        return self.findings

    def enclosing_class(self, func):
        p = parent(func) if func is not None else None
        return p if isinstance(p, ast.ClassDef) else None

    def class_method(self, cls, name):
        if cls is None:
            return None
        for s in cls.body:
            if isinstance(s, (ast.FunctionDef, ast.AsyncFunctionDef)) and s.name == name:
                return s
        return None

    def scope(self, body, func):
        is_async = isinstance(func, ast.AsyncFunctionDef)
        cls = self.enclosing_class(func)
        if func is None:
            self.local_bindings, self.shadowed = {}, set()
        else:
            self.local_bindings = self.collect_bindings(walk_local(func.body), plain=True)
            a = func.args
            self.shadowed = {x.arg for x in a.args + a.kwonlyargs + getattr(a, "posonlyargs", [])}
            self.shadowed |= {x.arg for x in (a.vararg, a.kwarg) if x is not None}
            self.shadowed |= {n.id for n in walk_local(func.body) if isinstance(n, ast.Name)
                              and isinstance(n.ctx, ast.Store)}
        for n in walk_local(body):
            if isinstance(n, ast.Call):
                self.check_call(n, func, is_async, cls)
            elif is_async and isinstance(n, TRY_TYPES):
                self.check_handlers(n, func)
            elif is_async and isinstance(n, ast.For):
                self.check_sequential_loop(n)
            elif is_async and isinstance(n, ast.While) and not func.name.startswith("test"):
                self.check_busy_wait(n)
            elif is_async and isinstance(n, ast.With):
                self.check_thread_lock_with(n)
        if func is None:
            return
        if is_async:
            self.check_no_await(func, cls)
        else:
            self.check_sync_route(func)
        self.check_unused_tasks(func)

    # ------------------------------------------------------------ calls
    def is_task_spawn(self, call):
        if not isinstance(call, ast.Call):
            return False
        name = self.resolve(call.func)
        if name in ("asyncio.create_task", "asyncio.ensure_future"):
            return True
        f = call.func
        if isinstance(f, ast.Attribute) and f.attr in ("create_task", "ensure_future"):
            recv = f.value
            if isinstance(recv, ast.Call) and self.resolve(recv.func) in (
                    "asyncio.get_running_loop", "asyncio.get_event_loop", "asyncio.new_event_loop"):
                return True
            if "loop" in dotted(recv).lower():
                return True
        return False

    def is_coroutine_call(self, call, cls):
        f = call.func
        name = self.resolve(f)
        if name in ASYNCIO_COROS:
            return name
        if isinstance(f, ast.Name) and f.id in self.async_funcs and f.id not in self.sync_funcs \
                and f.id not in self.imports:
            return f.id
        if isinstance(f, ast.Attribute):
            recv = dotted(f.value)
            if recv in ("self", "cls"):
                m = self.class_method(cls, f.attr)
                if isinstance(m, ast.AsyncFunctionDef):
                    return recv + "." + f.attr
            if f.attr in ALWAYS_ASYNC_ATTRS:
                return (recv or "<obj>") + "." + f.attr
            ctor = self.lookup(recv)
            if ctor in ASYNC_PRIM_METHODS and f.attr in ASYNC_PRIM_METHODS[ctor]:
                return recv + "." + f.attr
            if f.attr in ("drain", "wait_closed") and "writer" in recv.lower():
                return recv + "." + f.attr
        return ""

    def queue_key(self, call):
        p = parent(call)
        if isinstance(p, ast.Assign):
            return dotted(p.targets[0])
        if isinstance(p, ast.AnnAssign):
            return dotted(p.target)
        return ""

    def check_call(self, call, func, is_async, cls):
        f = call.func
        name = self.resolve(f)
        attr = f.attr if isinstance(f, ast.Attribute) else None
        par = parent(call)
        awaited = isinstance(par, ast.Await)
        statement = isinstance(par, ast.Expr)

        if statement:
            coro = self.is_coroutine_call(call, cls)
            if coro:
                self.add("AIO007", call, f"`{coro}(...)` is a coroutine but is never awaited; its body never runs",
                         f"await {ast.unparse(call)}")
            elif self.is_task_spawn(call):
                self.add("AIO008", call, f"result of `{dotted(f) or 'create_task'}(...)` is discarded: the task can "
                         "be garbage-collected mid-flight and its exception is never retrieved")
        if name == "asyncio.get_event_loop":
            where = "inside a coroutine use asyncio.get_running_loop()" if is_async else \
                "at the entry point use asyncio.run(main()) / asyncio.Runner"
            self.add("AIO010", call, f"asyncio.get_event_loop() is deprecated outside a running loop "
                     f"(RuntimeError in 3.14 when no loop is set); {where}")
        if name == "asyncio.wait_for" and self.target >= (3, 11):
            self.add("AIO015", call, "asyncio.wait_for in code targeting %s: prefer `async with asyncio.timeout(...)`"
                     % fmt_version(self.target))
        if name in ASYNC_QUEUES and self.queue_key(call) in self.awaited_put:
            maxsize = call.args[0] if call.args else next((k.value for k in call.keywords if k.arg == "maxsize"), None)
            if maxsize is None or (isinstance(maxsize, ast.Constant) and maxsize.value in (0, None)):
                self.add("AIO021", call, f"`{name}()` has no maxsize: producers never block, so there is no "
                         "backpressure and memory can grow without limit")
        if name == "asyncio.gather":
            self.check_gather(call, func, cls)
            if any(k.arg == "return_exceptions" and isinstance(k.value, ast.Constant) and k.value.value is True
                   for k in call.keywords):
                self.check_return_exceptions(call, func)
        if func is not None and name in ASYNC_RESOURCES:
            self.check_resource(call, func, cls, name)
        if not is_async:
            return

        # ---- inside async def only
        if name == "asyncio.run":
            self.add("AIO009", call, "asyncio.run() inside async def raises 'cannot be called from a running "
                     "event loop'", f"result = await {ast.unparse(call.args[0]) if call.args else 'coro()'}")
        if attr == "run_until_complete":
            self.add("AIO011", call, "run_until_complete() inside async def: the loop is already running",
                     f"result = await {ast.unparse(call.args[0]) if call.args else 'coro()'}")
        if awaited:
            return
        rule = BLOCKING_CALLS.get(name)
        if not rule and name:
            head, _, last = name.rpartition(".")
            if head in ("requests", "httpx") and last in HTTP_VERBS:
                rule = "AIO002"
            elif name in SYNC_CLIENTS:
                rule = SYNC_CLIENTS[name]
            elif name in THREAD_LOCKS:
                self.add("AIO016", call, f"`{name}()` created inside async def: thread primitives block the "
                         "event loop; use the asyncio equivalent")
                return
        if rule:
            label = dotted(f) or name
            self.add(rule, call, f"`{label}(...)` blocks the event loop inside `async def`")
            return
        if self.is_builtin(f, "open"):
            self.add("AIO004", call, "builtin open() inside async def does blocking disk I/O on the event loop")
            return
        if self.is_builtin(f, "input"):
            self.add("AIO006", call, "input() inside async def blocks the event loop until the user types")
            return
        if attr in PATH_IO and not (isinstance(f.value, ast.Name) and f.value.id in self.imports):
            self.add("AIO004", call, f"`.{attr}()` inside async def does blocking disk I/O on the event loop")
            return
        if attr:
            recv = dotted(f.value)
            parts = dotted(f).split(".")
            for i in range(len(parts) - 1, 0, -1):
                key = ".".join(parts[:i])
                ctor = self.lookup(key)
                if ctor in SYNC_CLIENTS and attr not in NONBLOCKING_METHODS and (
                        SYNC_CLIENTS[ctor] != "AIO002" or attr in SYNC_HTTP_METHODS):
                    self.add(SYNC_CLIENTS[ctor], call, f"`{dotted(f)}(...)` uses the synchronous `{ctor}` client "
                             "inside async def; it blocks the event loop")
                    return
                if ctor in THREAD_PRIMS and i == len(parts) - 1 and attr in THREAD_BLOCKING_METHODS:
                    self.add("AIO016", call, f"`{dotted(f)}()` on a `{ctor}` blocks the event loop thread")
                    return
            sockish = self.lookup(recv) in SOCKET_CTORS or "sock" in recv.lower().rsplit(".", 1)[-1]
            if attr in SOCKET_BLOCKING and sockish and not self.nonblocking_sockets:
                self.add("AIO006", call, f"`{dotted(f)}()` is a blocking socket call inside async def")

    # ------------------------------------------------------------ gather
    def find_assignment(self, func, name):
        if func is None:
            return None
        found = None
        for n in walk_local(func.body):
            if isinstance(n, ast.Assign) and len(n.targets) == 1 and isinstance(n.targets[0], ast.Name) \
                    and n.targets[0].id == name:
                found = n.value
        return found

    def callee_def(self, call, func, cls):
        if not isinstance(call, ast.Call):
            return None
        f = call.func
        if isinstance(f, ast.Name):
            if func is not None:
                for n in ast.walk(func):
                    if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) and n.name == f.id and n is not func:
                        return n
            return self.module_funcs.get(f.id)
        if isinstance(f, ast.Attribute) and dotted(f.value) in ("self", "cls"):
            return self.class_method(cls, f.attr)
        return None

    def classes_defining(self, method):
        return [c for c in ast.walk(self.tree) if isinstance(c, ast.ClassDef)
                and any(isinstance(m, (ast.FunctionDef, ast.AsyncFunctionDef)) and m.name == method
                        for m in c.body)]

    def check_gather(self, call, func, cls):
        for arg in call.args:
            if not isinstance(arg, ast.Starred):
                continue
            v = arg.value
            if isinstance(v, ast.Name):
                v = self.find_assignment(func, v.id)
            elt, it = None, None
            if isinstance(v, (ast.ListComp, ast.GeneratorExp)):
                elt, it = v.elt, v.generators[0].iter
            elif isinstance(v, ast.Call) and dotted(v.func) == "map" and len(v.args) >= 2:
                elt, it = ast.Call(func=v.args[0], args=[], keywords=[]), v.args[1]
            if elt is None or not isinstance(elt, ast.Call) or small_literal_iter(it):
                continue
            if any(h in ast.unparse(it).lower() for h in FANOUT_SAFE_HINTS):
                continue  # callback handlers / run managers / listeners: small by construction
            if self.is_task_spawn(elt) and elt.args:
                elt = elt.args[0]
            if not isinstance(elt, ast.Call):
                continue
            if self.resolve(elt.func) == "asyncio.to_thread" or (
                    isinstance(elt.func, ast.Attribute) and elt.func.attr == "run_in_executor"):
                continue  # the executor's worker count bounds concurrency
            if func is not None and isinstance(it, ast.Name) and func.args.vararg is not None \
                    and func.args.vararg.arg == it.id:
                continue  # generic helper over *aws: the caller decides the fan-out
            if func is not None and has_limit_hint(func):
                continue
            callee = self.callee_def(elt, func, cls)
            if callee is not None and has_limit_hint(callee):
                continue
            if isinstance(elt.func, ast.Attribute) and any(
                    has_limit_hint(c) for c in self.classes_defining(elt.func.attr)):
                continue  # obj.method(...) where the class owns a Semaphore / bounded queue
            target = "TaskGroup + Semaphore" if self.target >= (3, 11) else "Semaphore"
            self.add("AIO012", call, f"gather fans out one coroutine per item of `{ast.unparse(it)}` with no "
                     f"concurrency limit (heuristic: fine if the input is small; otherwise bound it with a {target})")

    def check_return_exceptions(self, call, func):
        p = parent(call)
        holder = parent(p) if isinstance(p, ast.Await) else p
        if isinstance(holder, ast.Expr):
            fresh = any(isinstance(a, ast.Call) or (isinstance(a, ast.Starred)
                        and isinstance(a.value, (ast.ListComp, ast.GeneratorExp))
                        and not self.is_task_spawn(a.value.elt)) for a in call.args)
            if not fresh:
                return  # `await gather(*tasks, return_exceptions=True)` after cancel(): cleanup idiom
            self.add("AIO022", call, "gather(..., return_exceptions=True) result is discarded, so every failure "
                     "is silently dropped")
            return
        if func is None or not isinstance(holder, (ast.Assign, ast.AnnAssign)):
            return
        targets = holder.targets if isinstance(holder, ast.Assign) else [holder.target]
        names = set()
        for t in targets:
            names |= names_in(t, ast.Store)
        for n in ast.walk(func):
            if isinstance(n, ast.Call) and dotted(n.func) == "isinstance" and len(n.args) == 2:
                if any(("Exception" in x or "Error" in x) for x in names_in(n.args[1]) | {dotted(n.args[1])}):
                    return
                if any(isinstance(x, ast.Attribute) and ("Exception" in x.attr or "Error" in x.attr)
                       for x in ast.walk(n.args[1])):
                    return
            if isinstance(n, (ast.Return, ast.Yield)) and isinstance(n.value, ast.Name) and n.value.id in names:
                return  # raw results handed to the caller, who checks them
            if isinstance(n, ast.Call) and n is not call and dotted(n.func) not in (
                    "len", "print", "zip", "enumerate", "list", "dict", "sum", "sorted", "range", "iter"):
                if any(isinstance(a, ast.Name) and a.id in names for a in n.args):
                    return
        self.add("AIO022", call, "gather(..., return_exceptions=True) results are used without checking "
                 "`isinstance(r, BaseException)`; failures will be treated as results")

    # ------------------------------------------------------------ resources
    def check_resource(self, call, func, cls, name):
        node = call
        p = parent(call)
        while isinstance(p, (ast.Await, ast.IfExp, ast.BoolOp)):
            node, p = p, parent(p)
        if isinstance(p, ast.Attribute):          # AsyncOpenAI(...).embeddings stored, or inline .get(...)
            top = p
            while isinstance(parent(top), ast.Attribute):
                top = parent(top)
            holder = parent(top)
            if isinstance(holder, (ast.Assign, ast.AnnAssign)) and not isinstance(
                    (holder.targets[0] if isinstance(holder, ast.Assign) else holder.target), ast.Name):
                return
            self.add("AIO019", call, f"`{name}(...)` created inline and never closed: a new connection pool per "
                     "call that leaks sockets")
            return
        if isinstance(p, (ast.withitem, ast.Return, ast.Yield, ast.YieldFrom, ast.Call, ast.keyword,
                          ast.Dict, ast.List, ast.Tuple, ast.Subscript)):
            return  # async with / factory / handed to an owner (AsyncExitStack, Depends, container)
        if isinstance(p, (ast.Assign, ast.AnnAssign)):
            targets = p.targets if isinstance(p, ast.Assign) else [p.target]
            t = targets[0]
            if isinstance(t, ast.Attribute):
                key = dotted(t)
                if key in self.close_targets or not key.startswith(("self.", "cls.")):
                    return  # closed somewhere, or stored on app.state / another owner
                if cls is not None and any(self.class_method(cls, m) for m in ("aclose", "close", "__aexit__")):
                    return
                if name not in ("httpx.AsyncClient", "aiohttp.ClientSession", "aiofiles.open"):
                    return  # SDK clients held by a long-lived object close themselves at exit
                self.add("AIO019", call, f"`{key} = {name}(...)` is never closed (no `{key}.aclose()` / "
                         "`close()` anywhere in the module and the class has no aclose/__aexit__)")
                return
            if isinstance(t, ast.Name):
                if self.local_escapes_or_closed(func, t.id):
                    return
                self.add("AIO019", call, f"`{t.id} = {name}(...)` is never closed; use `async with` so the "
                         "connection pool is released on every exit path")
                return
            return
        self.add("AIO019", call, f"`{name}(...)` used without `async with`: it is created per call and never closed")

    def local_escapes_or_closed(self, func, var):
        for n in ast.walk(func):
            if isinstance(n, ast.Call):
                if isinstance(n.func, ast.Attribute) and n.func.attr in ("aclose", "close") \
                        and dotted(n.func.value) == var:
                    return True
                if any(isinstance(a, ast.Name) and a.id == var for a in n.args) or \
                        any(isinstance(k.value, ast.Name) and k.value.id == var for k in n.keywords):
                    return True
            elif isinstance(n, ast.withitem) and dotted(n.context_expr) == var:
                return True
            elif isinstance(n, (ast.Return, ast.Yield)) and n.value is not None and var in names_in(n.value):
                return True
            elif isinstance(n, ast.Assign) and isinstance(n.value, ast.Name) and n.value.id == var \
                    and not all(isinstance(t, ast.Name) for t in n.targets):
                return True
        return False

    # ------------------------------------------------------------ statements
    def awaits_cancelled_task(self, node, func):
        """`t.cancel(); try: await t; except CancelledError: pass` - reaping a task we cancelled ourselves."""
        awaited = set()
        for n in walk_local(node.body):
            if isinstance(n, ast.Await):
                v = n.value
                if isinstance(v, ast.Call) and self.resolve(v.func) in ("asyncio.wait_for", "asyncio.gather",
                                                                         "asyncio.shield") and v.args:
                    v = v.args[0]
                if dotted(v):
                    awaited.add(dotted(v))
        if not awaited or func is None:
            return False
        for n in walk_local(func.body):
            if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) and n.func.attr == "cancel" \
                    and dotted(n.func.value) in awaited:
                return True
        return False

    def check_handlers(self, node, func):
        for h in node.handlers:
            if h.type is None:
                kind = "bare `except:`"
            else:
                elts = h.type.elts if isinstance(h.type, ast.Tuple) else [h.type]
                names = {dotted(e).rsplit(".", 1)[-1] for e in elts}
                hit = sorted(names & {"BaseException", "CancelledError"})
                if not hit:
                    continue
                kind = f"`except {hit[0]}`"
            if kind != "bare `except:`" and self.awaits_cancelled_task(node, func):
                continue
            if not reraises(h.body) and not captures(h.body, h.name):
                self.add("AIO014", h, f"{kind} in async def does not re-raise: CancelledError is swallowed, so "
                         "timeouts, TaskGroup cancellation and shutdown stop working")

    def check_sequential_loop(self, loop):
        targets = names_in(loop.target, ast.Store)
        if not targets or small_literal_iter(loop.iter):
            return
        body = list(walk_local(loop.body))
        if any(isinstance(n, (ast.Break, ast.Return, ast.Yield, ast.YieldFrom)) for n in body):
            return
        awaits = [n for n in body if isinstance(n, ast.Await)]
        for a in awaits:
            v = a.value
            if isinstance(v, ast.Call) and self.resolve(v.func) == "asyncio.sleep":
                return  # deliberate pacing (rate limit)
        stored = set()
        for n in body:
            if isinstance(n, ast.Name) and isinstance(n.ctx, ast.Store):
                stored.add(n.id)
        for n in body:
            if isinstance(n, (ast.Assign, ast.AugAssign, ast.AnnAssign)):
                for t in (n.targets if isinstance(n, ast.Assign) else [n.target]):
                    if isinstance(t, (ast.Attribute, ast.Subscript)):
                        stored |= names_in(t.value) - targets  # obj.x = ... / d[k] = ...: shared state
        for a in awaits:
            call = a.value
            if not isinstance(call, ast.Call):
                continue
            f = call.func
            if isinstance(f, ast.Attribute) and f.attr in ORDERED_METHODS:
                continue
            holder = parent(a)
            collects = isinstance(holder, (ast.Assign, ast.AnnAssign, ast.AugAssign)) or (
                isinstance(holder, ast.Call) and isinstance(holder.func, ast.Attribute)
                and holder.func.attr in ("append", "extend", "add"))
            if not collects:
                continue
            used = set()
            for x in list(call.args) + [k.value for k in call.keywords]:
                used |= names_in(x)
            if not used & targets:
                continue
            receiver = names_in(f.value) if isinstance(f, ast.Attribute) else set()
            deps = (used | receiver) & (stored - targets)
            if isinstance(holder, (ast.Assign, ast.AnnAssign)):
                tnames = set()
                for t in (holder.targets if isinstance(holder, ast.Assign) else [holder.target]):
                    tnames |= names_in(t, ast.Store)
                deps |= (used | receiver) & tnames
            if deps:
                continue  # loop-carried dependency (e.g. history = await llm(history, x))
            hint = "TaskGroup" if self.target >= (3, 11) else "asyncio.gather"
            self.add("AIO013", a, f"`await {ast.unparse(call)}` runs one item at a time in this for loop; if the "
                     f"items are independent, run them concurrently with {hint} (+ Semaphore). Ignore if order, "
                     "shared state or rate-limit pacing requires sequential calls")
            return

    def check_busy_wait(self, loop):
        awaits = [n for n in walk_local(loop.body) if isinstance(n, ast.Await)]
        sleeps = []
        for a in awaits:
            v = a.value
            if isinstance(v, ast.Call) and self.resolve(v.func) == "asyncio.sleep":
                d = v.args[0] if v.args else next((k.value for k in v.keywords if k.arg == "delay"), None)
                if isinstance(d, ast.Constant) and isinstance(d.value, (int, float)) and d.value <= 0.01:
                    sleeps.append(a)
                    continue
            return  # another await: the loop does real async work
        if sleeps:
            self.add("AIO018", sleeps[0], f"`{ast.unparse(sleeps[0])}` is the only await in this while loop: it "
                     "busy-waits at 100% CPU; wait on an asyncio.Event / Queue / Condition instead")

    def check_thread_lock_with(self, node):
        for item in node.items:
            key = dotted(item.context_expr)
            ctor = self.lookup(key)
            if ctor in THREAD_LOCKS and any(isinstance(n, ast.Await) for n in walk_local(node.body)):
                self.add("AIO016", node, f"`with {key}:` holds a `{ctor}` across an await; another task on the "
                         "same loop that takes it blocks the loop thread forever (deadlock)")

    def check_no_await(self, func, cls):
        if func.decorator_list or func.name.startswith("__") or func.name.startswith("test") or cls is not None \
                or not isinstance(parent(func), ast.Module) or is_stub(func.body):
            return  # decorated (routes, fixtures), methods, nested callbacks, tests and stubs are exempt
        lname = func.name.lower()
        if any(h in lname for h in INTERFACE_HINTS):
            return
        a = func.args
        if any(x.arg.startswith("_") for x in a.args + a.kwonlyargs + getattr(a, "posonlyargs", [])):
            return  # unused `_params` signal a callback signature imposed by a framework
        if func.name in self.value_names:
            return  # passed around as a callable (stage function, callback): must stay async
        end = getattr(func, "end_lineno", func.lineno)
        if any(f["rule"] in BLOCKING_RULES and func.lineno <= f["line"] <= end and f["path"] == self.path
               for f in self.findings):
            return  # the blocking-call finding already says what to fix
        for n in walk_local(func.body):
            if isinstance(n, (ast.Await, ast.AsyncFor, ast.AsyncWith, ast.Yield, ast.YieldFrom)):
                return
            if isinstance(n, ast.comprehension) and n.is_async:
                return
            if isinstance(n, ast.Call):
                d = dotted(n.func)
                if d.rsplit(".", 1)[-1] in LOOP_APIS:
                    return
        self.add("AIO017", func, f"`async def {func.name}` never awaits; make it a plain def (run blocking work "
                 "with asyncio.to_thread) unless it must satisfy an async interface")

    def check_sync_route(self, func):
        is_route = False
        for d in func.decorator_list:
            if isinstance(d, ast.Call) and isinstance(d.func, ast.Attribute) and d.func.attr in ROUTE_ATTRS:
                is_route = True
        if not is_route:
            return
        for n in walk_local(func.body):
            if isinstance(n, ast.Call) and (self.resolve(n.func) == "asyncio.run"
                                            or (isinstance(n.func, ast.Attribute)
                                                and n.func.attr == "run_until_complete")):
                self.add("AIO023", n, f"sync route `{func.name}` starts a new event loop per request; declare it "
                         "`async def` and await the coroutine")

    def check_unused_tasks(self, func):
        for n in walk_local(func.body):
            if not (isinstance(n, ast.Assign) and len(n.targets) == 1 and isinstance(n.targets[0], ast.Name)):
                continue
            v = n.value
            spawn = self.is_task_spawn(v) or (isinstance(v, ast.ListComp) and self.is_task_spawn(v.elt))
            if not spawn:
                continue
            var = n.targets[0].id
            loads = sum(1 for x in ast.walk(func) if isinstance(x, ast.Name) and x.id == var
                        and isinstance(x.ctx, ast.Load))
            if not loads:
                self.add("AIO020", n, f"task(s) stored in `{var}` are never awaited, gathered, cancelled or kept: "
                         "exceptions are lost and the task can outlive (or be collected before) its owner")

    # ------------------------------------------------------------ versions
    def check_versions(self):
        t = self.target
        for n in ast.walk(self.tree):
            if isinstance(n, (ast.Attribute, ast.Name)) and not isinstance(parent(n), ast.Attribute):
                name = self.resolve(n)
                need = VERSION_FEATURES.get(name)
                if need and t < need:
                    self.add("AIO024", n, f"`{name}` needs Python {fmt_version(need)}+ but --target-version is "
                             f"{fmt_version(t)}")
            elif hasattr(ast, "TryStar") and isinstance(n, ast.TryStar) and t < (3, 11):
                self.add("AIO024", n, f"`except*` needs Python 3.11+ but --target-version is {fmt_version(t)}")
            elif isinstance(n, ast.AsyncFor) and isinstance(n.iter, ast.Call) \
                    and self.resolve(n.iter.func) == "asyncio.as_completed" and t < (3, 13):
                self.add("AIO024", n, "`async for` over asyncio.as_completed needs Python 3.13+ but "
                         f"--target-version is {fmt_version(t)}")
            elif isinstance(n, ast.Call) and self.resolve(n.func) == "asyncio.run" and t < (3, 12) \
                    and any(k.arg == "loop_factory" for k in n.keywords):
                self.add("AIO024", n, f"asyncio.run(loop_factory=...) needs Python 3.12+ but --target-version is "
                         f"{fmt_version(t)}")


# ---------------------------------------------------------------- CLI

def lint_source(source, path, target):
    tree = ast.parse(source, filename=path)
    return Analyzer(path, source, tree, target).run()


def collect(paths):
    """Yield (path, explicit) pairs; explicit=False for files found by walking a directory."""
    skip = {".git", ".venv", "venv", "__pycache__", "node_modules", ".tox", ".mypy_cache", "build", "dist"}
    for p in paths:
        if p != "-" and os.path.isdir(p):
            for root, dirs, files in os.walk(p):
                dirs[:] = sorted(d for d in dirs if d not in skip)
                for f in sorted(files):
                    if f.endswith(".py"):
                        yield os.path.join(root, f), False
        else:
            yield p, True


def parse_rule_list(values):
    out = set()
    for v in values or []:
        for tok in v.split(","):
            tok = tok.strip()
            if not tok:
                continue
            matched = [rid for rid, info in RULES.items() if rid.startswith(tok.upper()) or info["name"] == tok]
            if not matched:
                raise ValueError(f"unknown rule {tok!r}; run --list-rules")
            out.update(matched)
    return out


def parse_version(text):
    m = re.match(r"^3\.(\d+)$", text.strip())
    if not m or not 9 <= int(m.group(1)) <= 15:
        raise ValueError(f"--target-version must look like 3.9 ... 3.15, got {text!r}")
    return (3, int(m.group(1)))


def print_rules():
    for rid, info in RULES.items():
        print(f"{rid} [{info['severity']}] {info['name']}: {info['summary']}")
        print(f"    why: {info['why']}")
        print("    fix: " + info["fix"].replace("\n", "\n         "))
        print()


def main(argv=None):
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("paths", nargs="*", help="Python files or directories, or - for stdin")
    ap.add_argument("--json", action="store_true", help="emit machine-readable JSON")
    ap.add_argument("--strict", action="store_true", help="exit 1 if any finding is reported")
    ap.add_argument("--select", action="append", metavar="IDS",
                    help="only these rules (ids, id prefixes or names; comma list)")
    ap.add_argument("--ignore", action="append", metavar="IDS",
                    help="skip these rules (ids, id prefixes or names; comma list)")
    ap.add_argument("--min-severity", choices=["low", "medium", "high"], default="low",
                    help="hide findings below this severity (default: low)")
    ap.add_argument("--target-version", default="3.11", metavar="3.N",
                    help="lowest Python version the code must run on (default 3.11); drives AIO015 and AIO024")
    ap.add_argument("--list-rules", action="store_true", help="print every rule with why and fix, then exit")
    args = ap.parse_args(argv)

    if args.list_rules:
        print_rules()
        return 0
    if not args.paths:
        ap.print_usage(sys.stderr)
        print("error: give at least one file, directory or - for stdin", file=sys.stderr)
        return 2
    try:
        select = parse_rule_list(args.select) or set(RULES)
        select -= parse_rule_list(args.ignore)
        target = parse_version(args.target_version)
    except ValueError as err:
        ap.print_usage(sys.stderr)
        print(f"error: {err}", file=sys.stderr)
        return 2

    findings, errors, warnings, files = [], [], [], 0
    for path, explicit in collect(args.paths):
        label = "<stdin>" if path == "-" else path
        try:
            if path == "-":
                source = sys.stdin.read()
            else:
                with open(path, encoding="utf-8") as fh:
                    source = fh.read()
            findings.extend(lint_source(source, label, target))
            files += 1
        except SyntaxError as err:
            msg = f"{label}:{err.lineno}: cannot parse: {err.msg}"
            (errors if explicit else warnings).append(msg)
        except (OSError, UnicodeDecodeError, ValueError) as err:
            msg = f"{label}: cannot read: {err}"
            (errors if explicit else warnings).append(msg)

    floor = SEVERITY_ORDER[args.min_severity]
    findings = [f for f in findings if f["rule"] in select and SEVERITY_ORDER[f["severity"]] <= floor]
    findings.sort(key=lambda f: (f["path"], f["line"], f["col"], f["rule"]))
    counts = {s: sum(f["severity"] == s for f in findings) for s in SEVERITY_ORDER}

    if args.json:
        print(json.dumps({"files": files, "target_version": fmt_version(target), "summary": counts,
                          "errors": errors, "warnings": warnings, "findings": findings}, indent=2))
    else:
        for w in warnings:
            print(f"warning: skipped {w}", file=sys.stderr)
        for e in errors:
            print(f"error: {e}", file=sys.stderr)
        for f in findings:
            print(f"{f['path']}:{f['line']}:{f['col']}: {f['rule']} [{f['severity']}] {f['name']}: {f['message']}")
            if f["code"]:
                print(f"    > {f['code']}")
            print("    fix: " + f["fix"].replace("\n", "\n         "))
            print()
        if findings:
            print(f"{len(findings)} finding(s) in {files} file(s): {counts['high']} high, "
                  f"{counts['medium']} medium, {counts['low']} low (target Python {fmt_version(target)})")
        elif not errors:
            print(f"No asyncio issues found in {files} file(s) (target Python {fmt_version(target)}).")
    if errors:
        return 2
    return 1 if args.strict and findings else 0


if __name__ == "__main__":
    sys.exit(main())