Race Conditions in AI-Generated Concurrent Python

The prompt said: “make this endpoint handle concurrent requests efficiently.” The AI assistant delivered async def functions, asyncio.gather, and a shared dict that every coroutine writes to without a lock. The code compiles. Tests pass. Under load, it corrupts itself.

This is the race condition pattern AI assistants generate most often in Python — and it’s worth understanding precisely where the danger lives, because the fix is simple once you know the mechanism.

The Async Performance Pattern AI Generates

BrassCoders, the scanner that catches what AI assistants structurally miss, regularly surfaces a specific async antipattern: shared mutable state written by multiple coroutines with no synchronization between them. The shape is always the same — a dict, list, or counter defined at module or class scope, modified inside an async def function, and called via asyncio.gather or asyncio.create_task.

Here’s what the generated code looks like:

# AI-generated: shared counter, no lock
import asyncio

request_counts = {}  # shared dict

async def handle_request(user_id: str):
    current = request_counts.get(user_id, 0)
    await asyncio.sleep(0)          # simulates I/O
    request_counts[user_id] = current + 1   # write after yield

async def main():
    await asyncio.gather(
        handle_request("alice"),
        handle_request("alice"),
        handle_request("alice"),
    )

asyncio.run(main())
print(request_counts)  # often {"alice": 1}, not {"alice": 3}

The AI wrote idiomatic async Python. The bug is that await asyncio.sleep(0) yields control back to the event loop, and another coroutine runs before the first one writes its result. Three coroutines all read 0, all compute 1, and all write 1. Three increments. One net change. The Python documentation for asyncio is explicit about this: coroutines are cooperatively scheduled, not preempted, so every await is a potential race window on shared state.

Where the Race Window Lives

Python’s asyncio event loop is single-threaded. That’s the key architectural fact. A coroutine runs until it hits an await expression, at which point the event loop may schedule a different coroutine. Between any two await points, a coroutine holds the thread exclusively — no other coroutine can interleave.

The race window, then, is any read-modify-write sequence that spans an await:

# Race: read happens before await, write happens after
value = shared_state["key"]    # read
await some_io_call()           # yield — other coroutines run here
shared_state["key"] = value + 1  # write — stale value

If the state never crosses an await, asyncio’s single-threaded model protects it. If it does, you have a race.

Threading makes this worse. Mix asyncio with threading.Thread or concurrent.futures.ThreadPoolExecutor and you lose the cooperative-scheduling safety net entirely. In a threaded context, any access to shared mutable state without a lock is a race — no await required. The Python threading module documentation covers threading.Lock as the standard guard.

The risk profile depends on what the shared state controls. A corrupted request counter is a metrics bug. A corrupted session token map or permission cache is a security bug. Auth code and rate-limiter state are the two spots where AI-generated async patterns most commonly introduce real exposure.

What BrassCoders Can Catch

BrassCoders runs 12 static-analysis scanners — Bandit, Pylint, Pyre/Pysa, Semgrep, ast-grep, detect-secrets, and six custom detectors — and reports patterns it can identify deterministically from the source code.

For async concurrency, the deterministic signals BrassCoders catches include:

• threading.Thread calls without a paired threading.Lock — Pylint flags bare thread creation over functions that modify globals.
• Global mutable collections modified inside async def functions — Semgrep and ast-grep can match this structural pattern with a custom rule.
• asyncio.gather over functions that write a shared outer-scope variable — catchable via AST pattern matching when the shared variable is in scope.

What BrassCoders does not catch: whether two specific coroutines are logically independent. If coroutine A writes results["user_a"] and coroutine B writes results["user_b"], they share the same dict object but never write the same key. That’s safe. Inferring logical independence from variable names and key expressions is context inference — the AI consumer’s job, not the pattern reporter’s.

Bandit 1.8.6 (the version BrassCoders bundles) has no threading-specific plugin for race conditions. Its plugins cover injection, crypto, file permissions, and subprocess safety. Bandit B351 is hashlib; B411 is xmlrpc. Neither touches shared-state concurrency. This is a documented gap in the Bandit rule set, not a BrassCoders limitation — it’s why the custom detectors and the AI triage layer exist.

What the AI Triage Layer Handles

BrassCoders emits raw scanner findings as YAML for your AI assistant — Claude Code, Cursor, or whichever tool you use. The AI triage layer reads the YAML and applies context inference that no static scanner can do reliably.

For a finding flagged as “async function modifies shared dict without lock,” the AI triage layer can:

• Trace the actual call sites. Are the callers always passing disjoint keys? Safe. Are they passing user-controlled keys that could collide? Not safe.
• Check whether the write spans an await. If the dict write happens before any await in the function, no race window exists in asyncio. The scanner doesn’t know this; the AI can read the function body.
• Evaluate the security impact. Is this dict a metrics store or a permission cache? The answer changes whether this is a severity-low correctness issue or a severity-high security finding.

The division of labor is intentional. BrassCoders flags the structural pattern reliably and directly. The AI assigns severity based on context. That sequence produces fewer false positives than a scanner trying to infer security impact from variable names — and it never suppresses a real bug because a variable name looked benign.

The Fix: Locking Shared State

The standard fix for shared mutable state in asyncio is asyncio.Lock. The Python standard library has shipped it since Python 3.4, and the asyncio synchronization primitives documentation covers it directly.

For the shared counter example:

import asyncio

request_counts = {}
lock = asyncio.Lock()   # one lock per shared resource

async def handle_request(user_id: str):
    async with lock:
        current = request_counts.get(user_id, 0)
        request_counts[user_id] = current + 1
        # no await inside the lock — keeps the critical section short

async def main():
    await asyncio.gather(
        handle_request("alice"),
        handle_request("alice"),
        handle_request("alice"),
    )

asyncio.run(main())
print(request_counts)  # {"alice": 3} — correct

Key points on the fix:

The lock must be created before the event loop starts. Creating an asyncio.Lock inside the function being gathered is a common mistake — each coroutine gets its own lock object, which provides no mutual exclusion.

Avoid awaiting inside the critical section. If you hold the lock and yield to the event loop, other coroutines block on the lock for the full duration of your I/O call. Keep the read-modify-write sequence atomic and yield outside.

For counters specifically, asyncio.Queue is sometimes a cleaner alternative. Producer coroutines put increments into the queue; a single consumer coroutine drains it. No shared state, no lock, no race.

For threaded code that calls async functions via asyncio.run_coroutine_threadsafe, use threading.Lock — not asyncio.Lock. The asyncio primitives only work within a running event loop on the same thread.

---

BrassCoders flags the structural patterns that indicate missing synchronization. Your AI assistant triages whether the specific instance is a correctness bug, a security bug, or a false positive. Neither layer alone gives you the full picture. Both together do.

Install BrassCoders on your codebase:

pip install brasscoders
brasscoders scan .

The OSS core is Apache 2.0 — no account, no telemetry, no network calls. The BrassCoders Paid plan adds AI-powered enrichment for $12/dev/month when you want the full signal reduction pass on top.