FastAPI AnyIO TaskGroup Lifespan Structured Concurrency Standard

Enforces bounded lifecycle scoping and eager failure propagation across asynchronous background workers during server shutdown and reload events. Background tasks spawned during application execution must be bounded by an AnyIO TaskGroup managed inside the ASGI lifespan context to prevent unhandled background exceptions from being swallowed and avoid orphaned zombie tasks during graceful termination. Bypassing structured concurrency via raw asyncio background tasks leads to resource leaks, hung shutdown hooks, and silent runtime corruption.

--- description: FastAPI AnyIO TaskGroup Lifespan Structured Concurrency Standard - Enforces bounded lifecycle scoping and eager failure propagation across asynchronous background workers during server shutdown and reload events. Background tasks spawned during application execution must be bounded by an AnyIO TaskGroup managed inside the ASGI lifespan context to prevent unhandled background exceptions from being swallowed and avoid orphaned zombie tasks during graceful termination. Bypassing structured concurrency via raw asyncio background tasks leads to resource leaks, hung shutdown hooks, and silent runtime corruption. globs: **/main.py, **/app.py, **/lifespan.py, **/*lifespan*.py, **/routers/**/*.py, **/services/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-taskgroup-lifespan-concurrency" --- # FastAPI AnyIO TaskGroup Lifespan Structured Concurrency Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Spawning Unbounded Fire-and-Forget asyncio.create_task Workers Outside Lifespan TaskGroups ## System Constraints You are a Principal Software Architect enforcing the FastAPI AnyIO TaskGroup Lifespan Structured Concurrency Standard. You must mandate that all long-running asynchronous worker loops, queue consumers, and background dispatchers are strictly tied to a top-level or sub-scoped AnyIO TaskGroup initialized in the application's lifespan async context manager. Reject any unmanaged `asyncio.create_task` or bare threading invocations that leak past request or application boundaries. ## Strict Guidelines - Engineers MUST ALWAYS initialize an AnyIO `create_task_group()` context manager within the FastAPI ASGI `lifespan` handler to govern the lifetime of background workers. - Engineers MUST NOT spawn detached background routines using `asyncio.create_task` or `asyncio.ensure_future` without registering them to an active, bounded TaskGroup. - Services MUST attach the active `TaskGroup` or a cancellation scope to `app.state` to allow route-level workers to be spawned within deterministic execution boundaries. - Code SHOULD catch and handle `anyio.get_cancelled_exc_class()` during shutdown loops to perform atomic drain operations and avoid premature task termination. - Components NEVER allow unhandled exceptions inside background worker loops to fail silently; unhandled worker failures MUST trigger TaskGroup cancellation and propagate outward to crash or halt the lifespan predictably. ## Bad vs Good Implementations ### Lifespan Background Worker Supervision: Fire-and-Forget vs AnyIO TaskGroup #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task directly inside the lifespan detaches worker execution from structured lifecycle boundaries. Exceptions in unbounded_worker do not interrupt the ASGI host, causing silent failure states, and shutdown signals will orphan the task without deterministic cancellation propagation. ```python import asyncio from contextlib import asynccontextmanager from fastapi import FastAPI async def unbounded_worker(): while True: await asyncio.sleep(1) # Failures here are detached from lifespan and swallowed silently raise RuntimeError("Unrecoverable stream failure") @asynccontextmanager async def lifespan(app: FastAPI): # ANTI-PATTERN: Fire-and-forget task leaking past application scope # On SIGTERM, worker is not gracefully drained and may block server reload task = asyncio.create_task(unbounded_worker()) yield # Shutdown has no mechanism to ensure worker finishes safely or stops immediately app = FastAPI(lifespan=lifespan) ``` #### Production Standard (Correct) > **Standard**: Wrapping the worker in an anyio.create_task_group within the lifespan establishes strict structured concurrency. Any unhandled worker error immediately collapses the task group and propagates up to the ASGI server. On server shutdown, the finally block explicitly cancels active scopes, guaranteeing that tasks are drained cleanly before termination completes. ```python from contextlib import asynccontextmanager import anyio from fastapi import FastAPI async def supervised_worker(cancel_scope: anyio.CancelScope): try: while not cancel_scope.cancel_called: await anyio.sleep(1) # In production, work is performed here except anyio.get_cancelled_exc_class(): # Graceful cleanup on server shutdown pass @asynccontextmanager async def lifespan(app: FastAPI): # PRODUCTION FIX: AnyIO TaskGroup guarantees bounded lifecycle and eager failure propagation async with anyio.create_task_group() as tg: app.state.task_group = tg with anyio.CancelScope() as scope: tg.start_soon(supervised_worker, scope) try: yield finally: # Initiate cooperative shutdown before exiting TaskGroup block scope.cancel() app = FastAPI(lifespan=lifespan) ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO TaskGroup Lifespan Structured Concurrency Standard

FastAPI AnyIO TaskGroup Lifespan Structured Concurrency Standard defines an industrial-grade behavioral standard for backend engineering. Enforces bounded lifecycle scoping and eager failure propagation across asynchronous background workers during server shutdown and reload events. Background tasks spawned during application execution must be bounded by an AnyIO TaskGroup managed inside the ASGI lifespan context to prevent unhandled background exceptions from being swallowed and avoid orphaned zombie tasks during graceful termination. Bypassing structured concurrency via raw asyncio background tasks leads to resource leaks, hung shutdown hooks, and silent runtime corruption. 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: Spawning Unbounded Fire-and-Forget asyncio.create_task Workers Outside Lifespan TaskGroups 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	Spawning Unbounded Fire-and-Forget asyncio.create_task Workers Outside Lifespan TaskGroups	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 initialize an AnyIO `create_task_group()` context manager within the FastAPI ASGI `lifespan` handler to govern the lifetime of background workers.

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.