FastAPI Lifespan Structured TaskGroup Teardown Invariant

--- description: FastAPI Lifespan Structured TaskGroup Teardown Invariant - Enforces structured concurrency inside ASGI lifespan context managers using anyio or asyncio TaskGroups. Prevents unmanaged fire-and-forget background tasks from leaking connections, corrupting state, or abruptly terminating upon SIGTERM by guaranteeing deterministic cancellation and resource draining. globs: app/**/*.py, src/**/*.py, main.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-lifespan-taskgroup-teardown-invariant" --- # FastAPI Lifespan Structured TaskGroup Teardown Invariant > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Spawning Unbounded Fire-and-Forget asyncio.create_task Inside Lifespans ## System Constraints You are a Principal Backend Architect enforcing strict structured concurrency in ASGI applications. Mandate that all asynchronous background workers, telemetry daemons, and consumer loops initiated during FastAPI application startup are scoped strictly within an anyio or asyncio.TaskGroup inside the lifespan generator. Prohibit bare asyncio.create_task calls that escape lifecycle management. ## Strict Guidelines - Engineers MUST ALWAYS encapsulate concurrent background tasks within an explicit anyio.create_task_group() or Python 3.11+ asyncio.TaskGroup() block in the ASGI lifespan context manager. - Engineers MUST NOT spawn detached background tasks using asyncio.create_task() or loop.create_task() inside lifespan handlers or API routes without an explicit lifecycle owner. - Code MUST use a cooperative cancellation mechanism (such as anyio.Event or asyncio.Event) to signal workers to drain before exiting the lifespan context block. - Services MUST ensure that the lifespan generator yields execution to FastAPI strictly while the task group context remains open, awaiting complete task completion upon teardown. - Components NEVER swallow or suppress asyncio.CancelledError or anyio.get_cancelled_exc_class() without re-raising or executing required cleanup. ## Bad vs Good Implementations ### Unbounded Background Task vs. Structured Lifespan TaskGroup #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Spawning tasks with asyncio.create_task inside the lifespan creates detached coroutines that are not tracked or drained when the application receives SIGTERM. The ASGI container shuts down immediately after yield, causing abrupt task truncation, unclosed database transactions, and leaked connections. ```python import asyncio from contextlib import asynccontextmanager from fastapi import FastAPI async def background_worker(): while True: await asyncio.sleep(1) # Process queue items @asynccontextmanager async def lifespan(app: FastAPI): # ANTI-PATTERN: Fire-and-forget task escapes lifecycle control # On SIGTERM, Uvicorn terminates the event loop before worker cleans up, # causing leaked DB connections or half-committed writes. asyncio.create_task(background_worker()) yield app = FastAPI(lifespan=lifespan) ``` #### Production Standard (Correct) > **Standard**: Encapsulating the background worker inside an anyio.create_task_group() within the lifespan generator guarantees structured concurrency. When the ASGI host initiates shutdown, the finally block signals the worker and cancels the task scope, ensuring all coroutines drain and complete cleanup before the lifespan exit returns. ```python from contextlib import asynccontextmanager import anyio from fastapi import FastAPI async def background_worker(stop_event: anyio.Event): try: while not stop_event.is_set(): # Simulate cooperative batch processing with anyio.move_on_after(1.0): await stop_event.wait() finally: # Deterministic teardown and state flushing await anyio.to_thread.run_sync(lambda: None) @asynccontextmanager async def lifespan(app: FastAPI): stop_event = anyio.Event() async with anyio.create_task_group() as tg: tg.start_soon(background_worker, stop_event) try: yield finally: # Signal cooperative termination and allow tg to drain and cancel stop_event.set() tg.cancel_scope.cancel() app = FastAPI(lifespan=lifespan) ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Lifespan Structured TaskGroup Teardown Invariant

FastAPI Lifespan Structured TaskGroup Teardown Invariant defines an industrial-grade behavioral standard for backend engineering. Enforces structured concurrency inside ASGI lifespan context managers using anyio or asyncio TaskGroups. Prevents unmanaged fire-and-forget background tasks from leaking connections, corrupting state, or abruptly terminating upon SIGTERM by guaranteeing deterministic cancellation and resource draining. 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 Inside Lifespans 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 Inside Lifespans	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 encapsulate concurrent background tasks within an explicit anyio.create_task_group() or Python 3.11+ asyncio.TaskGroup() block in the ASGI lifespan context manager.

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.