FastAPI AnyIO TaskGroup Structured Concurrency & Deterministic Cancellation Standard

--- description: FastAPI AnyIO TaskGroup Structured Concurrency & Deterministic Cancellation Standard - Enforces strict structured concurrency in FastAPI backends using AnyIO TaskGroups to prevent orphaned coroutines, resource exhaustion, and unhandled exception swallowing. Unhandled child errors must cascade deterministic cancellation to all siblings within the failure domain and cleanly unwind via shielded cancellation scopes. globs: app/**/*.py, src/**/*.py, routers/**/*.py, services/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-taskgroup-structured-concurrency-protocol" --- # FastAPI AnyIO TaskGroup Structured Concurrency & Deterministic Cancellation Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unmanaged asyncio.create_task Fire-and-Forget Pipelines Causing Zombie Coroutines and Silent Exception Swallowing ## System Constraints You are a Principal Backend Architect enforcing structured concurrency invariants in FastAPI applications powered by AnyIO. You mandate deterministic async task lifecycles: no raw asyncio.create_task invocations, mandatory TaskGroup boundaries for fan-out execution, explicit exception group unwinding, and shielded cancel scopes for non-abortable resource teardown. ## Strict Guidelines - Engineers MUST ALWAYS encapsulate concurrent worker fan-outs inside an 'anyio.create_task_group()' context manager to ensure parent-child lifecycle coupling. - Engineers MUST NOT invoke 'asyncio.create_task' or 'asyncio.gather(..., return_exceptions=True)' for background request processing, as they leak coroutines across request boundaries. - Code MUST use 'with anyio.CancelScope(shield=True):' exclusively when executing critical teardown logic (e.g., socket release, transaction rollback) within cancellation handlers. - Services MUST capture and handle 'ExceptionGroup' (or use PEP 654 'except*') when catching failures from task group exits to avoid masking sibling diagnostic traces. - Components NEVER catch 'BaseException' or AnyIO cancellation exceptions without immediately re-raising, ensuring cancellation propagation is never suppressed. ## Bad vs Good Implementations ### Unstructured Task Spawning vs AnyIO Structured TaskGroup Teardown #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task or unbounded asyncio.gather leaves pending background coroutines orphaned when a sibling raises an error. The runtime leaks memory, active sockets, and database connections while dropping exception traces. ```python import asyncio from fastapi import APIRouter, HTTPException router = APIRouter() async def process_item(item_id: str): if item_id == "bad": raise ValueError("Processing failed") await asyncio.sleep(2) @router.post("/pipeline/bad") async def trigger_pipeline(items: list[str]): # ANTI-PATTERN: Raw asyncio.create_task creates unmanaged, dangling tasks. # If one task fails, the other runs forever; exceptions are dropped or logged as unretrievable. tasks = [asyncio.create_task(process_item(item)) for item in items] try: results = await asyncio.gather(*tasks) return {"status": "ok", "results": results} except Exception as exc: # Leaks pending tasks in 'tasks' without cancellation! raise HTTPException(status_code=500, detail=str(exc)) ``` #### Production Standard (Correct) > **Standard**: anyio.create_task_group guarantees that any unhandled child exception immediately signals cancellation to all sibling tasks. Cleanups are protected with shielded CancelScopes, and PEP 654 exception group filtering isolates failures deterministically. ```python import anyio from fastapi import APIRouter, HTTPException, status router = APIRouter() async def process_item(item_id: str, cancel_scope: anyio.CancelScope) -> str: try: if item_id == "bad": raise ValueError(f"Invalid item payload: {item_id}") await anyio.sleep(2) return f"processed_{item_id}" except anyio.get_cancelled_exc_class(): # Execute atomic resource cleanup under a shielded scope with anyio.CancelScope(shield=True): # Teardown logic guaranteed to run to completion despite parent cancellation pass raise @router.post("/pipeline/good") async def trigger_pipeline(items: list[str]): results: list[str] = [] try: async with anyio.create_task_group() as tg: for item in items: tg.start_soon(process_item, item, tg.cancel_scope) except* ValueError as eg: # PEP 654 ExceptionGroup handling guarantees clean propagation and introspection raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"Batch processing aborted: {eg.exceptions[0]}" ) from eg return {"status": "ok", "processed_count": len(items)} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO TaskGroup Structured Concurrency & Deterministic Cancellation Standard

FastAPI AnyIO TaskGroup Structured Concurrency & Deterministic Cancellation Standard defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency in FastAPI backends using AnyIO TaskGroups to prevent orphaned coroutines, resource exhaustion, and unhandled exception swallowing. Unhandled child errors must cascade deterministic cancellation to all siblings within the failure domain and cleanly unwind via shielded cancellation scopes. 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: Unmanaged asyncio.create_task Fire-and-Forget Pipelines Causing Zombie Coroutines and Silent Exception Swallowing 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	Unmanaged asyncio.create_task Fire-and-Forget Pipelines Causing Zombie Coroutines and Silent Exception Swallowing	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 worker fan-outs inside an 'anyio.create_task_group()' context manager to ensure parent-child lifecycle coupling.

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.