FastAPI Lifespan Asyncio TaskGroup Structured Concurrency Guardrail

--- description: FastAPI Lifespan Asyncio TaskGroup Structured Concurrency Guardrail - Enforces Python 3.12+ structured concurrency patterns in FastAPI applications using asyncio.TaskGroup within the lifespan context manager. Prevents dangling background coroutines, ContextVar cross-request contamination, and unhandled task crashes during ASGI graceful shutdown sequences. globs: src/**/main.py, src/**/lifespan.py, src/**/app.py, app/**/main.py, app/**/lifespan.py, app/**/app.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-lifespan-taskgroup-concurrency-guardrail" --- # FastAPI Lifespan Asyncio TaskGroup Structured Concurrency Guardrail > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Fire-and-forget asyncio.create_task() spawned during application lifespan without TaskGroup bounds or cooperative cancellation ## System Constraints You are a Principal Software Architect enforcing Python 3.12+ structured concurrency invariants across FastAPI applications. You must mandate that background coroutines initiated during the lifespan cycle are strictly managed through `asyncio.TaskGroup` context managers, preventing orphan task leaks, context variable bleed, and non-deterministic application teardowns. ## Strict Guidelines - Engineers MUST ALWAYS manage background worker coroutines using an `asyncio.TaskGroup` bound to the `@asynccontextmanager` lifespan of the FastAPI instance. - Engineers MUST NOT invoke unbound `asyncio.create_task()` or rely on deprecated `on_event('startup')` / `on_event('shutdown')` hooks. - Code MUST isolate context variables per worker loop using `contextvars.copy_context().run` when task contexts must remain pristine across scheduled cycles. - Services MUST implement cooperative cancellation via `asyncio.CancelledError` trapping to ensure external handles (connections, file descriptors, channels) flush and close properly. - Components NEVER swallow `asyncio.CancelledError` or `BaseExceptionGroup` without completing critical teardown sequences or re-raising appropriately to notify the ASGI server. ## Bad vs Good Implementations ### FastAPI Lifespan Background Task Orchestration #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using unbound asyncio.create_task inside the lifespan function creates a dangling coroutine. During ASGI server shutdown, Uvicorn/Hypercorn terminates the main loop without waiting for the task to exit or clean up network sockets, potentially leading to 'Task was destroyed but it is still pending!' warnings and corrupted state. ```python import asyncio from contextlib import asynccontextmanager from fastapi import FastAPI # ANTI-PATTERN: Fire-and-forget task without lifecycle management # Leaks resources, ignores cancellation during SIGTERM, and causes context bleeding @asynccontextmanager async def lifespan(app: FastAPI): # Unreferenced task may be garbage collected prematurely or outlive app shutdown asyncio.create_task(run_cache_refresh()) yield # Shutdown phase does not wait for run_cache_refresh to finish or cancel cleanly async def run_cache_refresh(): while True: await asyncio.sleep(5) print("Refreshed cache") app = FastAPI(lifespan=lifespan) ``` #### Production Standard (Correct) > **Standard**: Wrapping the worker within an asyncio.TaskGroup in the lifespan context guarantees that child coroutines are bound to the lifecycle of the FastAPI instance. Using an asyncio.Event alongside cooperative cancellation ensures predictable teardown before the server completely terminates. ```python import asyncio from collections.abc import AsyncIterator from contextlib import asynccontextmanager from fastapi import FastAPI async def run_cache_refresh(stop_event: asyncio.Event) -> None: try: while not stop_event.is_set(): # Implement cooperative cancellation or check stop_event try: async with asyncio.timeout(5.0): await stop_event.wait() except* TimeoutError: # Perform refresh work on tick pass except asyncio.CancelledError: # Flush buffers and gracefully teardown worker resources raise @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: stop_event = asyncio.Event() async with asyncio.TaskGroup() as tg: # TaskGroup ensures child task exceptions are captured in ExceptionGroup tg.create_task(run_cache_refresh(stop_event)) try: yield finally: # Initiate cooperative shutdown before exiting TaskGroup block stop_event.set() app = FastAPI(lifespan=lifespan) ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Lifespan Asyncio TaskGroup Structured Concurrency Guardrail

FastAPI Lifespan Asyncio TaskGroup Structured Concurrency Guardrail defines an industrial-grade behavioral standard for backend engineering. Enforces Python 3.12+ structured concurrency patterns in FastAPI applications using asyncio.TaskGroup within the lifespan context manager. Prevents dangling background coroutines, ContextVar cross-request contamination, and unhandled task crashes during ASGI graceful shutdown sequences. Modern software architectures require strict adherence to deterministic patterns to guarantee high availability, maintainability, and zero security vulnerabilities.

Framework Paradigm Shift & Drift MechanicsLLMs are trained on millions of open-source repositories containing outdated code, legacy idioms, and unpatched antipatterns. When coding assistants generate implementations for this stack, they default to common statistical tokens: Fire-and-forget asyncio.create_task() spawned during application lifespan without TaskGroup bounds or cooperative cancellation Without an explicit system instruction file, developers spend hours debugging silent failures.

Failure Modes & Production Blast Radius

Hazard Analysis: What Breaks Without This Rule

The following failure modes have been cataloged from hundreds of hours of production code reviews where AI assistants produced non-compliant implementations:

Failure Mode	AI Symptom (The Anti-Pattern)	Severity	Production Impact & Remedy
AI Hallucination & Deprecation Drift	Fire-and-forget asyncio.create_task() spawned during application lifespan without TaskGroup bounds or cooperative cancellation	Critical	Violating this industrial standard introduces regression bugs, security risks, and unexpected runtime failures in production. Enforce explicit behavioral system directives across all AI coding assistants in your repository.
Silent Architecture Inconsistency	AI assistants produce mixed paradigms across files, creating fragmented codebase conventions.	Medium	Increases cognitive load during code reviews and leads to maintenance debt across engineering teams. Pin project conventions in .cursor/rules, CLAUDE.md, and AGENTS.md so all team agents follow identical guardrails.

Architectural Directives in Detail

Step-by-Step Production Directives

Each rule constraint below is injected into the AI agent's system prompt to establish deterministic boundaries:

Engineers MUST ALWAYS manage background worker coroutines using an `asyncio.TaskGroup` bound to the `@asynccontextmanager` lifespan of the FastAPI instance.

Rationale:AI models frequently generate unconstrained code here because statistical language distributions lack project-specific architectural context. Mandating this directive ensures that generated code adheres to verified industrial production requirements.