Unify the teardown-on-shutdown attach behind one guarded seam

CLAUDE.md

@@ -85,7 +85,7 @@ Design docs and implementation plans live under `planning/` (not under `docs/`,
 - **Optional-import guard pattern**: top-level conditional imports (`if import_checker.is_X_installed: import X`) keep optional dependencies actually optional. Code that references `X` is only reached when `check_dependencies()` has already returned True; the runtime invariant is maintained by the inline `is_configured → check_dependencies → instantiate` flow in `BaseBootstrapper.__init__`. See "Type checking" below.
 - **`from_dict` vs `from_object` accept different shapes for `None`**: `BaseConfig.from_dict({"service_name": None})` succeeds and explicitly overrides the default with `None`. `BaseConfig.from_object(obj)` where `obj.service_name is None` filters the attribute out and the dataclass default takes over. The asymmetry is documented in both methods' docstrings (`instruments/base.py:17, 23`) and pinned by tests in `tests/test_config.py:54-94`. Pick `from_dict` if explicit-None override is the load-bearing semantic.
 - **`__post_init__` cascade invariant**: every config-class `__post_init__` must call `super().__post_init__()` so MRO chains terminate cleanly. `BaseConfig` has a no-op `__post_init__` as the chain terminator. Required because `OpenTelemetryConfig.__post_init__` (SEC-2 warning), `CorsConfig.__post_init__` (SEC-3 validation), and `FastAPIConfig.__post_init__` (UnsetType app construction) all sit on the same MRO for `FastAPIConfig`/`LitestarConfig`/`FastStreamConfig`/`FreeConfig`; without the cascade, a class that returns early before `super()` blocks the rest of the chain. `FastAPIConfig` uses the explicit `super(FastAPIConfig, self).__post_init__()` form because `@dataclass(slots=True)` replaces the class object after the body compiles and breaks bare `super()`.
-- **`_lite_bootstrap_*` prefix for sentinels on user-supplied app instances**: when the bootstrapper needs to tag a user-supplied framework app (FastAPI, FastMCP, Litestar, FastStream) with internal state — e.g., the `_lite_bootstrap_lifespan_attached` marker that prevents double-wrap on FastAPI — store it as a direct attribute on the app instance with a `_lite_bootstrap_` prefix. Read with `getattr(application, "_lite_bootstrap_<name>", False)` (no SLF violation); write with `application._lite_bootstrap_<name> = value  # noqa: SLF001`. Don't squat in framework-provided user namespaces like Starlette's `application.state`.
+- **`_lite_bootstrap_*` prefix for sentinels on user-supplied app instances**: when the bootstrapper needs to tag a user-supplied framework app (FastAPI, FastMCP, Litestar, FastStream) with internal state, store it as a direct attribute on the instance with a `_lite_bootstrap_` prefix. The canonical example is the `_lite_bootstrap_teardown_attached` marker that gates the double-attach guard, set once by the shared `BaseBootstrapper._attach_teardown_once(target, attach)` and applied uniformly across all four app-bearing bootstrappers (FastAPI tags the app, Litestar tags its `AppConfig`, FastStream/FastMCP tag the app). Read with `getattr(target, "_lite_bootstrap_<name>", False)` (no SLF violation); write with `setattr`/`target._lite_bootstrap_<name> = value  # noqa: SLF001`. Don't squat in framework-provided user namespaces like Starlette's `application.state`. See `architecture/bootstrappers.md`.
 
 ### Type checking
 

architecture/bootstrappers.md

@@ -65,12 +65,34 @@ The same string is produced by the public `build_summary()` method, callable at
 any later point (REPL, health endpoint) regardless of log-level filtering. To
 inspect skips programmatically, iterate `bootstrapper.skipped_instruments`.
 
+## Teardown-on-shutdown attach
+
+Each app-bearing bootstrapper wires its `teardown` into the framework's shutdown
+lifecycle from `__init__`, through one shared seam:
+`BaseBootstrapper._attach_teardown_once(target, attach)`. That method owns the
+double-attach guard — it tags the attach target with a
+`_lite_bootstrap_teardown_attached` marker, so a second bootstrapper on the same
+app warns and skips rather than stacking a second teardown hook. Only the `attach`
+thunk is framework-specific:
+
+- **FastAPI** — merge a lifespan context manager (`_wrap_lifespan`); target is the app.
+- **Litestar** — append to `application_config.on_shutdown`; target is the
+  `AppConfig` (the built `Litestar` app is slotted and is never tagged).
+- **FastStream** — register via `application.on_shutdown(...)`; target is the app.
+- **FastMCP** — add a `_TeardownProvider` whose async lifespan runs teardown; target
+  is the app (FastMCP exposes no `on_shutdown` API).
+- **Free** — no app, no shutdown lifecycle; not wired.
+
+The guard is uniform: the same marker and warning apply to all four app-bearing
+frameworks. `attach` is typed `Callable[[], object]` because some hooks (FastStream's
+`on_shutdown`) return the callback.
+
 ## App-tagging sentinel convention
 
 When a bootstrapper must tag a user-supplied framework app (FastAPI, FastMCP,
 Litestar, FastStream) with internal state, it stores a direct attribute prefixed
 `_lite_bootstrap_` rather than squatting in framework namespaces like Starlette's
-`application.state`. Example: FastAPI's lifespan double-wrap guard reads
-`getattr(application, "_lite_bootstrap_lifespan_attached", False)` (no SLF
-violation) and writes
-`application._lite_bootstrap_lifespan_attached = True  # noqa: SLF001`.
+`application.state`. The canonical example is the teardown guard's
+`_lite_bootstrap_teardown_attached` marker, read via
+`getattr(target, BaseBootstrapper._TEARDOWN_MARKER, False)` (no SLF violation) and
+written via `setattr` inside `_attach_teardown_once`.

lite_bootstrap/bootstrappers/base.py

@@ -24,6 +24,31 @@ class BaseBootstrapper(abc.ABC, typing.Generic[ApplicationT]):
     skipped_instruments: list[tuple[type[BaseInstrument], str]]
     bootstrap_config: BaseConfig
 
+    # Marker tagged on a user-supplied app (or its config) once this bootstrapper has
+    # attached its teardown to the framework's shutdown lifecycle. Prevents a second
+    # bootstrapper on the same app from re-attaching. See architecture/bootstrappers.md.
+    _TEARDOWN_MARKER: typing.ClassVar[str] = "_lite_bootstrap_teardown_attached"
+
+    def _attach_teardown_once(self, target: object, attach: typing.Callable[[], object]) -> None:
+        """Run ``attach`` (which wires ``teardown`` into the framework's shutdown) once per target.
+
+        Idempotent across bootstrapper instances: if ``target`` is already tagged, warn
+        and skip rather than stacking a second teardown hook. ``attach`` is the only
+        framework-specific part; detection + warning live here.
+        """
+        if getattr(target, self._TEARDOWN_MARKER, False):
+            warnings.warn(
+                f"{type(self).__name__} already has a lite-bootstrap teardown hook attached to this "
+                f"application or its configuration; skipping. This {type(self).__name__}'s teardown "
+                f"will not run on shutdown — construct one {type(self).__name__} per application.",
+                stacklevel=3,
+            )
+            return
+        # Mark only after a successful attach: if attach() raises, the target stays untagged
+        # so a retry can re-attach rather than silently warning-and-skipping forever.
+        attach()
+        setattr(target, self._TEARDOWN_MARKER, True)
+
     def build_summary(self) -> str:
         """Return a multi-line human-readable summary of configured + skipped instruments.
 

lite_bootstrap/bootstrappers/fastapi_bootstrapper.py

@@ -197,19 +197,10 @@ async def lifespan_manager(self, _: "fastapi.FastAPI") -> typing.AsyncIterator[d
 
     def __init__(self, bootstrap_config: FastAPIConfig) -> None:
         super().__init__(bootstrap_config)
-
         application = _narrow_app(self.bootstrap_config)
-        # FastAPI's lifespan_context is opaque after wrap; tag the app instance directly
-        # rather than squatting in Starlette's user-facing application.state namespace.
-        if getattr(application, "_lite_bootstrap_lifespan_attached", False):
-            warnings.warn(
-                "FastAPI application already has a lite-bootstrap lifespan wrapper attached; "
-                "skipping re-wrap. This FastAPIBootstrapper's teardown will not be invoked on "
-                "ASGI shutdown — construct one FastAPIBootstrapper per application.",
-                stacklevel=2,
-            )
-            return
-        application._lite_bootstrap_lifespan_attached = True  # noqa: SLF001  # ty: ignore[unresolved-attribute]
+        self._attach_teardown_once(application, lambda: self._wrap_lifespan(application))
+
+    def _wrap_lifespan(self, application: "fastapi.FastAPI") -> None:
         old_lifespan_manager = application.router.lifespan_context
         application.router.lifespan_context = _merge_lifespan_context(
             old_lifespan_manager,

lite_bootstrap/bootstrappers/fastmcp_bootstrapper.py

@@ -2,7 +2,6 @@
 import dataclasses
 import time
 import typing
-import warnings
 
 from lite_bootstrap import import_checker
 from lite_bootstrap.bootstrappers.base import BaseBootstrapper
@@ -152,15 +151,8 @@ def is_ready(self) -> bool:
 
     def __init__(self, bootstrap_config: FastMcpConfig) -> None:
         super().__init__(bootstrap_config)
-        if any(isinstance(p, _TeardownProvider) for p in self.bootstrap_config.application.providers):
-            warnings.warn(
-                "FastMCP application already has a _TeardownProvider attached; skipping re-attachment. "
-                "This FastMcpBootstrapper's teardown will not be invoked on ASGI shutdown — "
-                "construct one FastMcpBootstrapper per application.",
-                stacklevel=2,
-            )
-            return
-        self.bootstrap_config.application.add_provider(_TeardownProvider(self.teardown))
+        application = self.bootstrap_config.application
+        self._attach_teardown_once(application, lambda: application.add_provider(_TeardownProvider(self.teardown)))
 
     def _prepare_application(self) -> "FastMCP[typing.Any]":
         return self.bootstrap_config.application

lite_bootstrap/bootstrappers/faststream_bootstrapper.py

@@ -205,7 +205,8 @@ def is_ready(self) -> bool:
 
     def __init__(self, bootstrap_config: FastStreamConfig) -> None:
         super().__init__(bootstrap_config)
-        self.bootstrap_config.application.on_shutdown(self.teardown)
+        application = self.bootstrap_config.application
+        self._attach_teardown_once(application, lambda: application.on_shutdown(self.teardown))
 
     def _prepare_application(self) -> "AsgiFastStream":
         return self.bootstrap_config.application

lite_bootstrap/bootstrappers/litestar_bootstrapper.py

@@ -277,8 +277,12 @@ class LitestarBootstrapper(BaseBootstrapper["litestar.Litestar"]):
 
     def __init__(self, bootstrap_config: LitestarConfig) -> None:
         super().__init__(bootstrap_config)
-        self.bootstrap_config.application_config.debug = bootstrap_config.service_debug
-        self.bootstrap_config.application_config.on_shutdown.append(self.teardown)
+        application_config = self.bootstrap_config.application_config
+        self._attach_teardown_once(application_config, lambda: self._apply_config(application_config))
+
+    def _apply_config(self, application_config: "AppConfig") -> None:
+        application_config.debug = self.bootstrap_config.service_debug
+        application_config.on_shutdown.append(self.teardown)
 
     def is_ready(self) -> bool:
         return import_checker.is_litestar_installed