FastAPI AnyIO Structured TaskGroup Exception Shielding Invariant

--- description: FastAPI AnyIO Structured TaskGroup Exception Shielding Invariant - Enforces strict structured concurrency via AnyIO TaskGroups within FastAPI asynchronous execution contexts to guarantee deterministic cancellation propagation. Prevents orphaned background tasks and resource leaks by mandating explicit cancellation scopes and shielding critical cleanup blocks during task group failure. globs: src/**/*.py, app/**/*.py, routers/**/*.py, services/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-taskgroup-shielding-invariant" --- # FastAPI AnyIO Structured TaskGroup Exception Shielding Invariant > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unstructured coroutine spawning via asyncio.create_task or unshielded cancellation blocks leaking zombie background workers. ## System Constraints You are a Principal Software Architect enforcing the FastAPI AnyIO Structured TaskGroup Exception Shielding Invariant. You must reject unstructured coroutine scheduling (e.g., asyncio.create_task, asyncio.gather without strict TaskGroups) inside FastAPI endpoints and background service calls. You must ensure all concurrent subtasks run inside structured `anyio.create_task_group()` contexts, propagate BaseException/CancelledError faithfully, and wrap all transactional rollbacks or cleanup inside `anyio.CancelScope(shield=True)` to prevent orphaned background tasks and partial state mutations. ## Strict Guidelines - Engineers MUST ALWAYS use `anyio.create_task_group()` instead of `asyncio.create_task()` or naked `asyncio.gather()` when fanning out asynchronous subtasks in FastAPI dependencies, routes, or services. - Engineers MUST wrap all critical cleanup, compensation, or database rollback logic inside an `anyio.CancelScope(shield=True)` context when executing within or around an active TaskGroup cancellation. - Engineers MUST NOT catch `anyio.get_cancelled_exc_class()` or `asyncio.CancelledError` without immediately re-raising, unless explicitly completing an idempotent terminal teardown within a shielded scope. - Code SHOULD catch `ExceptionGroup` (or `BaseExceptionGroup` in Python 3.11+) when unwinding multi-task failures from a TaskGroup exit, preserving the causal exception chain. - Services MUST ensure that if any concurrent subtask fails, all sibling subtasks in the task group are cancelled immediately and awaited deterministically before the request lifecycle terminates. ## Bad vs Good Implementations ### Unstructured Fire-and-Forget vs Structured AnyIO TaskGroup with Shielded Cleanup #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task or unconstrained asyncio.gather leaves sibling coroutines running as detached background zombies when one task encounters an error. If the client disconnects or billing fails, inventory reservation keeps running to completion without cancellation, causing data inconsistency and resource leaks. ```python import asyncio from fastapi import APIRouter, HTTPException router = APIRouter() async def bill_customer(user_id: str) -> None: await asyncio.sleep(0.5) raise RuntimeError("Payment gateway timeout") async def reserve_inventory(item_id: str) -> None: await asyncio.sleep(2.0) # This continues running orphaned in the background if billing fails print("Inventory reserved permanently") @router.post("/checkout") async def checkout(user_id: str, item_id: str): # ANTI-PATTERN: Raw asyncio.gather does not cancel lingering tasks on failure # and can leak running coroutines when a client disconnects. try: task1 = asyncio.create_task(bill_customer(user_id)) task2 = asyncio.create_task(reserve_inventory(item_id)) await asyncio.gather(task1, task2) except Exception as exc: # Failed to cancel task2; task2 continues executing as a zombie task raise HTTPException(status_code=500, detail=str(exc)) ``` #### Production Standard (Correct) > **Standard**: Using anyio.create_task_group guarantees structured concurrency: if bill_customer raises a RuntimeError, reserve_inventory is immediately cancelled and awaited. The cleanup operation in reserve_inventory is shielded via anyio.CancelScope(shield=True), guaranteeing that compensating transactions finish cleanly without leaking zombie tasks. ```python import anyio from fastapi import APIRouter, HTTPException router = APIRouter() async def bill_customer(user_id: str) -> None: await anyio.sleep(0.5) raise RuntimeError("Payment gateway timeout") async def reserve_inventory(item_id: str) -> None: try: await anyio.sleep(2.0) except anyio.get_cancelled_exc_class(): # Must use shielded scope for rollback to avoid interruption during cancellation with anyio.CancelScope(shield=True): await release_inventory_reservation(item_id) raise async def release_inventory_reservation(item_id: str) -> None: # Guaranteed cleanup under shield await anyio.sleep(0.1) @router.post("/checkout") async def checkout(user_id: str, item_id: str): try: async with anyio.create_task_group() as tg: tg.start_soon(bill_customer, user_id) tg.start_soon(reserve_inventory, item_id) except* RuntimeError as eg: # Structured cancellation propagated; reserve_inventory was cancelled and cleaned up raise HTTPException(status_code=502, detail=f"Checkout failed: {eg.exceptions}") ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured TaskGroup Exception Shielding Invariant

FastAPI AnyIO Structured TaskGroup Exception Shielding Invariant defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency via AnyIO TaskGroups within FastAPI asynchronous execution contexts to guarantee deterministic cancellation propagation. Prevents orphaned background tasks and resource leaks by mandating explicit cancellation scopes and shielding critical cleanup blocks during task group failure. 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: Unstructured coroutine spawning via asyncio.create_task or unshielded cancellation blocks leaking zombie background workers. 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	Unstructured coroutine spawning via asyncio.create_task or unshielded cancellation blocks leaking zombie background workers.	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 use `anyio.create_task_group()` instead of `asyncio.create_task()` or naked `asyncio.gather()` when fanning out asynchronous subtasks in FastAPI dependencies, routes, or services.

Engineers MUST ALWAYS use `anyio.create_task_group()` instead of `asyncio.create_task()` or naked `asyncio.gather()` when fanning out asynchronous subtasks in FastAPI dependencies, routes, or services.

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.