FastAPI 0.115+ AnyIO Structured Concurrency TaskGroup Lifespan Standard

--- description: FastAPI 0.115+ AnyIO Structured Concurrency TaskGroup Lifespan Standard - Enforces deterministic cancellation and resource lifecycle management in FastAPI 0.115+ using AnyIO TaskGroups within lifespan contexts. Eliminates orphaned background worker coroutines, dangling database connections, and uncollected socket handles during server SIGTERM and SIGINT events. globs: app/**/*.py, src/**/*.py, api/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-structured-concurrency-lifespan" --- # FastAPI 0.115+ AnyIO Structured Concurrency TaskGroup Lifespan Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Fire-and-Forget asyncio.create_task Invocations During Lifespan Causing Silent Zombie Tasks and Resource Leaks ## System Constraints You are a Principal Software Architect enforcing structured concurrency invariants in FastAPI 0.115+ applications. You must reject unstructured fire-and-forget concurrency patterns such as asyncio.create_task or raw threading in application startup hooks. Mandate anyio.create_task_group() within asynccontextmanager lifespan handlers to guarantee deterministic teardown, task supervision, and clean cancellation scope propagation on SIGTERM. ## Strict Guidelines - Engineers MUST ALWAYS manage long-running application background tasks within an `anyio.create_task_group()` context inside the FastAPI lifespan handler. - Engineers MUST NOT invoke `asyncio.create_task()`, `asyncio.ensure_future()`, or legacy `@app.on_event('startup')` / `@app.on_event('shutdown')` decorators. - Code MUST explicitly cancel the `anyio.TaskGroup` using `tg.cancel_scope.cancel()` inside the `finally` block or post-yield teardown phase of the lifespan context. - Services MUST yield an application state dictionary or typed context holding references to shared resources, allowing endpoints to cleanly access monitored infrastructure. - Components NEVER start infinite background consumption loops without checking for cooperative cancellation or handling `anyio.get_cancelled_exc_class()`. ## Bad vs Good Implementations ### Migrating Unstructured Background Coroutines to AnyIO Structured TaskGroup Lifespans #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task inside the lifespan detaches coroutine execution from the ASGI server cancellation scope. When the server receives SIGTERM, the coroutine is not cleanly cancelled, causing socket leaks, unclosed client sessions, and hung worker processes. ```python import asyncio from contextlib import asynccontextmanager from fastapi import FastAPI async def unbounded_metric_flusher(): while True: await asyncio.sleep(5) # Orphaned loop that ignores ASGI shutdown print("Flushing metrics...") @asynccontextmanager async def lifespan(app: FastAPI): # ANTI-PATTERN: Fire-and-forget task escapes lifecycle supervision asyncio.create_task(unbounded_metric_flusher()) yield # Task is never cancelled, holding sockets open indefinitely during reload app = FastAPI(lifespan=lifespan) ``` #### Production Standard (Correct) > **Standard**: Encapsulating the lifespan within anyio.create_task_group() guarantees structured concurrency. Yielding within the block suspends the lifespan until the ASGI server triggers teardown. The finally block cancels all child tasks, ensuring complete resource reclamation before process termination. ```python from contextlib import asynccontextmanager from typing import AsyncIterator, TypedDict import anyio from anyio.abc import TaskGroup from fastapi import FastAPI class AppState(TypedDict): task_group: TaskGroup async def bounded_metric_flusher() -> None: try: while True: await anyio.sleep(5) # Monitored work payload except anyio.get_cancelled_exc_class(): # Guaranteed cooperative teardown with anyio.CancelScope(shield=True): # Flushes pending metrics during graceful shutdown await anyio.sleep(0.1) raise @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[AppState]: async with anyio.create_task_group() as tg: tg.start_soon(bounded_metric_flusher) try: yield {"task_group": tg} finally: # Deterministically trigger cancellation of all tasks in the group tg.cancel_scope.cancel() app = FastAPI(lifespan=lifespan) ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI 0.115+ AnyIO Structured Concurrency TaskGroup Lifespan Standard

FastAPI 0.115+ AnyIO Structured Concurrency TaskGroup Lifespan Standard defines an industrial-grade behavioral standard for backend engineering. Enforces deterministic cancellation and resource lifecycle management in FastAPI 0.115+ using AnyIO TaskGroups within lifespan contexts. Eliminates orphaned background worker coroutines, dangling database connections, and uncollected socket handles during server SIGTERM and SIGINT events. 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 Invocations During Lifespan Causing Silent Zombie Tasks and Resource Leaks 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 Invocations During Lifespan Causing Silent Zombie Tasks and Resource Leaks	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 long-running application background tasks within an `anyio.create_task_group()` context inside the FastAPI lifespan handler.

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.