scripts.check_async_await

Reference › scripts › check_async_await

An async-correctness gate that fails on un-awaited coroutines, resolving them by type through Pyright. No module or test imports it; it takes no options and is run by a gate or by hand.

main first looks for the pyright executable. If it is absent the gate prints a SKIPPED notice pointing at the dev dependencies and returns 0. Otherwise it runs Pyright with JSON output over the configured target directories, under a 600-second budget (a measured cold run took up to 380 seconds on the whole codebase). A timeout, or output that cannot be parsed, returns 2. It then filters the general diagnostics with _is_async_misuse, which keeps the coroutine-misuse diagnostics, and drops paths matched by _excluded. Any remaining hit is printed and fails the gate.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/check_async_await.py

#!/usr/bin/env python3
"""Async-correctness gate: fail on un-awaited coroutines (resolve-by-type via Pyright).

Detects the bug class where an ``async def`` is called WITHOUT ``await`` — the
coroutine is created and silently discarded (the work never runs, no exception is
raised), or its un-awaited result is used as if it were the resolved value.

Pyright is the detector because it ignores the ``# mypy: ignore-errors`` directive
present on ~228 source files (so it sees code mypy skips), AND because it is
*deterministic*: a runtime gate (erroring on the "coroutine never awaited"
RuntimeWarning) was evaluated and rejected — that warning fires at GC time via the
unraisable-exception hook, so under the free-threaded *batched* test runner it
fires non-deterministically and is mis-attributed to an innocent module, causing
flaky red builds. Static analysis has none of that timing dependence. See
``docs/sync-vs-async.md``.

Scope: both ``src/orchestrata_ultimate`` (production) and ``tests/`` (test
infrastructure can leak un-awaited coroutines too — e.g. an async driver fixture
whose teardown discards ``driver.close()``). The suite's ``--ignore``'d dirs
(``generated_recovery``, ``_generated_broken``) are excluded to match collection.

The gate is scoped to the TRUE-POSITIVE signatures only:
  * ``reportUnusedCoroutine``        — a bare ``async_fn()`` whose result is discarded.
  * operations on a ``CoroutineType`` — attribute/index/iteration/operator access on
                                        the un-awaited result of an ``async def``.

It deliberately EXCLUDES:
  * redis-py ``Awaitable[T] | T`` (``ResponseT``) stub noise on the *sync* client —
    Pyright emits ``Awaitable[...]`` (a union) for those; a real un-awaited async
    function is ``CoroutineType[...]`` (no union). See docs/sync-vs-async.md §1.
  * interface/argument signature mismatches (``reportArgumentType`` /
    ``reportIncompatibleMethodOverride``) — not un-awaited-coroutine bugs.

Exit 0 when clean, 1 when an offender is found. Skips (exit 0, warning) when
pyright is not installed, so a contributor without pyright is not blocked
locally; CI installs pyright (a declared dev dependency) and enforces the gate.
"""

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys

# Production code plus test infrastructure.
TARGETS = ["src/orchestrata_ultimate", "tests"]

# Mirror the suite's collection ignores (pyproject addopts: --ignore=...). These
# dirs never run, so their broken/generated contents are out of scope.
_EXCLUDE_DIRS = ("/generated_recovery/", "/_generated_broken/")

# Pyright rules that, combined with a CoroutineType in the message, mean an
# un-awaited async result is being *operated on* (attribute/index/iter/operator).
_OP_RULES = frozenset(
    {
        "reportAttributeAccessIssue",
        "reportIndexIssue",
        "reportOperatorIssue",
        "reportGeneralTypeIssues",
    }
)


def _excluded(path: str) -> bool:
    return any(d in path for d in _EXCLUDE_DIRS)


def _is_async_misuse(diag: dict) -> bool:
    """True iff the diagnostic is a genuine un-awaited-coroutine defect."""
    if diag.get("rule") == "reportUnusedCoroutine":
        return True
    message = diag.get("message", "")
    return "CoroutineType[" in message and diag.get("rule") in _OP_RULES


def main() -> int:
    if shutil.which("pyright") is None:
        print(
            "[async-gate] SKIPPED: pyright is not installed. Install the 'pyright' "
            "dev dependency (uv sync --dev) to enable this gate locally.",
            file=sys.stderr,
        )
        return 0

    # Pyright on this codebase measured 162s–380s (2026-09-10, cold cache
    # vs. warm cache, src/+tests/ ~122K LOC). Budget 600s — 1.5x the
    # worst measured run — so a genuine regression is caught without
    # false-timeout on legitimate runs. The old external `timeout 90`
    # wrapper simply killed a correct analysis mid-run.
    PYRIGHT_TIMEOUT = 600
    print(
        f"[async-gate] Running pyright on {', '.join(TARGETS)} "
        f"(timeout {PYRIGHT_TIMEOUT}s)...",
        file=sys.stderr,
        flush=True,
    )
    try:
        proc = subprocess.run(
            [sys.executable, "-m", "pyright", "--outputjson", *TARGETS],
            capture_output=True,
            text=True,
            timeout=PYRIGHT_TIMEOUT,
        )
    except subprocess.TimeoutExpired:
        print(
            f"[async-gate] TIMEOUT: pyright exceeded {PYRIGHT_TIMEOUT}s budget.",
            file=sys.stderr,
        )
        return 2
    try:
        data = json.loads(proc.stdout)
    except json.JSONDecodeError:
        print("[async-gate] ERROR: could not parse pyright output.", file=sys.stderr)
        print(proc.stdout[-2000:], file=sys.stderr)
        print(proc.stderr[-2000:], file=sys.stderr)
        return 2

    diags = data.get("generalDiagnostics", [])
    hits = [
        d for d in diags if _is_async_misuse(d) and not _excluded(d.get("file", ""))
    ]

    if not hits:
        print(
            f"[async-gate] OK: no un-awaited coroutines in {', '.join(TARGETS)} "
            f"({len(diags)} pyright diagnostics scanned)."
        )
        return 0

    print(
        f"[async-gate] BLOCKED: {len(hits)} un-awaited coroutine(s) found — an async "
        f"function is called without `await`, so the work never runs. "
        f"See docs/sync-vs-async.md.\n",
        file=sys.stderr,
    )
    for d in hits:
        try:
            rel = os.path.relpath(d["file"])
        except ValueError:
            rel = d["file"]
        line = d["range"]["start"]["line"] + 1
        first = (d.get("message", "") or "").splitlines()[0]
        print(f"  {rel}:{line}  [{d.get('rule')}]  {first}", file=sys.stderr)
    return 1


if __name__ == "__main__":
    raise SystemExit(main())