API Reference¶

class BaseApp[FactoryT = None](ABC):
    """Subclass and override ``wire`` to open resources and include resources/endpoints.

    The app owns the two exit stacks. Parameterize with a factory class —
    ``class MyApp(BaseApp[MyFactory])`` — and the app builds it at construction,
    injecting the stacks the factory's ``__init__`` names (``es`` for the
    ExitStack, ``aes`` for the AsyncExitStack). The built factory is then
    ``self._factory`` (typed as ``MyFactory``) inside ``wire``, and any resource
    it registers on those stacks is closed at shutdown.

    Pass ``factory=`` to supply a prebuilt factory instead of building one — the
    seam for tests, which inject a ``create_autospec`` stand-in
    (``MyApp(factory=mock_factory)``) so the real services are never constructed.

    Reverse-routed ``Location`` / ``Link`` URLs are relative unless the environment sets
    ``JERO_BASE_URL`` (a static public origin) or ``JERO_TRUST_FORWARDED`` (rebuild the
    origin per request from ``X-Forwarded-*``); see :func:`_forwarded_config_from_env`.
    """

    def __init__(self, *, factory: FactoryT | None = None) -> None:
        self.__static: _StaticRoutes = {}
        self.__dynamic: _DynamicRoutes = {}
        self.__websocket_static: dict[str, _WebSocketHandler] = {}
        self.__websocket_dynamic: dict[int, list[_WebSocketPattern]] = {}
        self.__allowed: _AllowedMethods = {}
        self.__allow_cache: dict[str, bytes] = {}
        self.__decoders: dict[type[Struct], Decoder[Struct]] = {}
        self.__operations: list[OperationSpec] = []  # captured for the OpenAPI document
        self.__openapi: _OpenAPIConfig | None = None  # set by _include_openapi, built at __finalize
        base_url, trust_forwarded = _forwarded_config_from_env()
        self.__reverser = _Reverser(base_url=base_url, trust_forwarded=trust_forwarded)
        self.__exceptions = _ExceptionHandlers(self.__reverser)
        # The tail for responses no route owns (unrouted 404, 405): the app-wide CORS
        # default and global middleware headers, filled at __finalize.
        self.__app_tail = _RouteTail()
        self.__includes: list[_IncludeRecord] = []
        self.__cors_default: CompiledCORS | None = None  # set by _include_cors
        # route handler -> its resolved CORS policy, for answering preflights; built
        # at __finalize once inheritance is resolved.
        self.__route_cors: dict[_Handler, CompiledCORS] = {}
        self.__middleware: list[CompiledMiddleware] = []  # global, in registration order
        # The pre-routing middleware machinery (global intercept table + observes);
        # stays None unless global middleware defines those hooks, so the disabled
        # per-request cost is one attribute load.
        self.__pre: _GlobalMiddleware | None = None
        self.__stack = ExitStack()
        self.__astack = AsyncExitStack()
        self.__factory: FactoryT = factory if factory is not None else self.__make_factory()

    @property
    def _factory(self) -> FactoryT:
        """The built factory, read inside ``wire``. Set once at construction (read-only)."""
        return self.__factory

    def _enter[T](self, cm: AbstractContextManager[T]) -> T:
        """Open a sync context manager, closed at shutdown in reverse order."""
        return self.__stack.enter_context(cm)

    async def _aenter[T](self, cm: AbstractAsyncContextManager[T]) -> T:
        """Open an async context manager, closed at shutdown in reverse order."""
        return await self.__astack.enter_async_context(cm)

    def __decoder(self, struct_type: type[Struct]) -> Decoder[Struct]:
        """The reusable typed JSON decoder for ``struct_type``, built once per app.

        Decoders are keyed by type, so models shared across handlers (a ``WidgetIn``
        used by both ``create`` and ``update_full``) share one decoder. Populated only at
        wiring time; the binder holds the resolved decoder, so the request path does
        no lookup.
        """
        if struct_type not in self.__decoders:
            self.__decoders[struct_type] = Decoder(struct_type)
        return self.__decoders[struct_type]

    def __resolve_factory_type(self) -> type | None:
        """The factory class from ``BaseApp[...]``, or None if unparameterized."""
        for base in get_original_bases(type(self)):
            if get_origin(base) is BaseApp:
                args = get_args(base)
                if args and isinstance(args[0], type) and args[0] is not type(None):
                    return args[0]
        return None

    def __make_factory(self) -> FactoryT:
        factory_type = self.__resolve_factory_type()
        if factory_type is None:
            return cast("FactoryT", None)
        return cast("FactoryT", _instantiate_factory(factory_type, self.__stack, self.__astack))

    @abstractmethod
    async def wire(self) -> None:
        """Override to open resources (via ``_enter`` / ``_aenter``) and include them.

        Runs once at startup. Anything entered via the helpers is torn
        down (in reverse order) at shutdown.

        Abstract: every ``BaseApp`` subclass must implement it. A subclass that
        omits it is flagged at its instantiation site by the type checker.
        """

    def __register(self, method: HTTPMethod, segments: list[_Segment], handler: _Handler) -> None:
        params = tuple((i, value) for i, (is_param, value) in enumerate(segments) if is_param)
        if not params:
            route_path = "/".join(value for _, value in segments)
            if (method, route_path) in self.__static:
                raise WiringError(f"{method} {route_path} is already registered")
            self.__static[(method, route_path)] = handler
            self.__allowed.setdefault(route_path, []).append(method)
            return

        statics = tuple((i, value) for i, (is_param, value) in enumerate(segments) if not is_param)
        bucket = self.__dynamic.setdefault((method, len(segments)), [])
        if any(pattern.statics == statics for pattern in bucket):
            raise WiringError(f"{method} {_template_str(segments)} is already registered")
        bucket.append(_Pattern(statics, params, handler))

    def __register_websocket(self, segments: list[_Segment], handler: _WebSocketHandler) -> None:
        params = tuple((i, value) for i, (is_param, value) in enumerate(segments) if is_param)
        if not params:
            route_path = "/".join(value for _, value in segments)
            if route_path in self.__websocket_static:
                raise WiringError(f"WebSocket {route_path} is already registered")
            self.__websocket_static[route_path] = handler
            return
        statics = tuple((i, value) for i, (is_param, value) in enumerate(segments) if not is_param)
        bucket = self.__websocket_dynamic.setdefault(len(segments), [])
        if any(pattern.statics == statics for pattern in bucket):
            raise WiringError(f"WebSocket {_template_str(segments)} is already registered")
        bucket.append(_WebSocketPattern(statics, params, handler))

    @staticmethod
    def __check_user_source(
        resource_cls: type,
        name: str,
        sources: Sources,
        auth: _CompiledAuth | None,
    ) -> None:
        """Validate a handler's ``user`` annotation against the route's authenticator — that
        there is one, that the Struct matches, and that its optionality agrees.

        A handler on an anonymous-accepting route must declare ``user``: with nothing to
        check, a handler that ignores the auth result would serve anonymous callers with no
        sign of it at the mount or in the signature. Behind a gating authenticator, omitting
        ``user`` stays fine — the gate has already run."""
        user_type = sources.user
        if user_type is None:
            if auth is not None and auth.reports_absence:
                raise WiringError(
                    f"{resource_cls.__name__}.{name} declares no 'user', but "
                    f"{auth.owner}.authenticate returns {auth.returns.__name__} | None, so "
                    f"this route serves anonymous callers — declare "
                    f"'user: {auth.returns.__name__} | None' and handle None, or mount it "
                    f"behind an authenticator that returns {auth.returns.__name__} to gate it",
                )
            return
        if auth is None:
            raise WiringError(
                f"{resource_cls.__name__}.{name} declares 'user' but no auth was given",
            )
        if not issubclass(auth.returns, user_type):
            raise WiringError(
                f"{resource_cls.__name__}.{name}: 'user' expects {user_type.__name__} "
                f"but {auth.owner}.authenticate returns {auth.returns.__name__}",
            )
        if auth.reports_absence and not sources.user_optional:
            raise WiringError(
                f"{resource_cls.__name__}.{name}: 'user' must be annotated "
                f"'{user_type.__name__} | None' — {auth.owner}.authenticate returns "
                f"'{user_type.__name__} | None', so a caller may arrive anonymous",
            )
        if sources.user_optional and not auth.reports_absence:
            raise WiringError(
                f"{resource_cls.__name__}.{name}: 'user' must be annotated "
                f"'{user_type.__name__}' — {auth.owner}.authenticate returns "
                f"'{user_type.__name__}', so an unauthenticated caller never reaches the "
                f"handler; return '{user_type.__name__} | None' from it to accept anonymous "
                f"callers",
            )

    def __include(
        self,
        obj: Resource | Endpoint,
        methods: dict[str, _Verb],
        *,
        auth: "Auth[Any, Any] | CookieAuth[Any, Any] | HybridAuth[Any, Any, Any] | None",
        cors: CORS | None,
        middleware: Sequence[object],
    ) -> None:
        cls = type(obj)
        if cors is not None and not isinstance(cast("object", cors), CORS):
            raise WiringError(
                f"{cls.__name__}: cors= must be a CORS policy (or CORS.OFF), "
                f"got {type(cors).__name__}",
            )
        # Explicit policies are validated and compiled here, at the include call —
        # only what depends on other registrations (inheriting an omitted cors=, the
        # global middleware every route picks up) waits for __finalize.
        cors_off = cors is CORS.OFF
        compiled_cors = CompiledCORS(cors) if cors is not None and not cors_off else None
        compiled_middleware = tuple(CompiledMiddleware(m) for m in middleware)
        # One shared tail per include: every route (and sender) the include registers
        # holds this instance, and __finalize fills it in place once coverage is known.
        tail = _RouteTail()
        record = _IncludeRecord(
            tail=tail,
            routes=[],
            cors=compiled_cors,
            cors_off=cors_off,
            middleware=compiled_middleware,
        )
        path = getattr(cls, "path", None)
        if path is None:
            raise WiringError(
                f"{cls.__name__}: no path — declare it on the class, "
                f"e.g. `class {cls.__name__}(..., path='/...')`.",
            )
        template = _parse_template(path)
        # The authenticator's declared return type is the policy: `-> TUser | None` accepts
        # anonymous callers, `-> TUser` gates. Never inferred from anything else.
        compiled_auth = _CompiledAuth(auth) if auth is not None else None
        auth_mode: AuthMode = None
        if compiled_auth is not None:
            auth_mode = "optional" if compiled_auth.reports_absence else "required"
        # An authed route with no declared scheme derives one from what authenticate()
        # binds — see _derive_security_scheme; it may derive nothing, which is only a
        # problem for an app that calls _include_openapi (operation_input raises there).
        security_scheme: SecurityScheme | None = None
        if compiled_auth is not None:
            declared: object = getattr(type(auth), "openapi_security", None)
            if declared is None:
                security_scheme = _derive_security_scheme(compiled_auth)
            elif isinstance(declared, SecurityScheme):
                security_scheme = declared
            else:
                raise WiringError(
                    f"{type(auth).__name__}: openapi_security must be SecurityScheme, "
                    f"got {type(declared).__name__}",
                )

        registered = False
        for name, verb in methods.items():
            fn = getattr(obj, name, None)
            if fn is None:
                continue
            sources = _bind_sources(cls, name, fn, verb, self.__decoder)
            self.__check_user_source(cls, name, sources, compiled_auth)
            segments = _route_segments(
                cls, name, template, sources.path, extends_path=verb.extends_path
            )
            # A NoContent/Created/Accepted return fixes its own status regardless of the
            # verb's default (204/201/202); a union's members carry their own already.
            status = _effective_status(sources.return_kind, verb.success_status)
            handler = _Route(
                fn,
                status,
                sources=sources,
                auth=compiled_auth,
                reverser=self.__reverser,
                exceptions=self.__exceptions,
                tail=tail,
            )
            self.__register(verb.method, segments, handler)
            record.routes.append((verb.method, handler))
            self.__reverser.register(
                fn.__func__, cls.ref, name, _RouteRef(tuple(segments), sources.path)
            )
            self.__operations.append(
                OperationSpec(
                    path=_template_str(segments),
                    method=verb.method.lower(),
                    success_status=status,
                    sources=sources,
                    auth_mode=auth_mode,
                    security_scheme=security_scheme,
                    class_meta=cls.meta,
                    op_meta=getattr(cls, f"meta_{name}", None),
                    operation_id_default=f"{cls.__name__}_{_camel(name)}",
                    auth_owner=compiled_auth.owner if compiled_auth is not None else None,
                )
            )
            registered = True

        if not registered:
            raise WiringError(f"{cls.__name__} defines none of: {', '.join(methods)}")
        self.__includes.append(record)

    def _include_exception_handler[
        E: Exception,
    ](self, handler: ExceptionHandler[E]) -> None:
        """Register a structurally typed custom exception handler.

        The exception, JSON body, and typed-header types are inferred from the concrete
        ``handle_exception`` signature and validated once during wiring.
        """
        self.__exceptions.register(handler)

    def _include_error_adapter(self, adapter: ErrorBodyAdapter[Any]) -> None:
        """Replace the Problem family's wire body app-wide with ``adapter``'s composition.

        Call inside ``wire``, at most once. Every Problem-family error — the framework's
        built-ins (404/405/422/500, …) and your own ``HTTPError`` subclasses, including
        those returned by exception handlers — is rendered through ``adapter.compose``
        instead of RFC 9457 Problem Details, and the derived OpenAPI error responses
        document the adapter's body. ``StructHTTPError``\\ s render themselves.
        """
        # The isinstance guards untyped callers; cast first so it isn't statically vacuous.
        if not isinstance(cast("object", adapter), ErrorBodyAdapter):
            raise WiringError(
                "_include_error_adapter requires an ErrorBodyAdapter instance, "
                f"got {type(adapter).__name__}",
            )
        if getattr(type(adapter), "body_type", None) is None:
            raise WiringError(
                f"{type(adapter).__name__} never bound a concrete body Struct; "
                "parameterize the class: ErrorBodyAdapter[YourBody]",
            )
        if self.__exceptions.adapter is not None:
            existing = type(self.__exceptions.adapter).__name__
            raise WiringError(
                f"an error body adapter ({existing}) is already registered; an app has at most one",
            )
        self.__exceptions.adapter = adapter

    def _include_cors(self, cors: CORS) -> None:
        """Serve cross-origin browser callers app-wide with one default :class:`CORS` policy.

        Call inside ``wire``, at most once (order among the ``include_*`` calls doesn't
        matter). Every include inherits the policy unless it passes its own ``cors=`` —
        a different policy overrides it, ``CORS.OFF`` removes it. Skipping
        ``_include_cors`` entirely means no CORS anywhere except includes that opt in
        with their own ``cors=``.

        The policy is compiled at wiring: a wildcard origin becomes constant header
        pairs in covered routes' responses (free per request), an origin allow-list one
        set lookup + origin echo. Preflights (``OPTIONS`` with
        ``Access-Control-Request-Method``) are answered per (path, requested method) on
        the existing OPTIONS branch, so two verbs on one path may carry two policies.
        Error responses carry the failing route's pairs — a browser page must be able
        to *read* the 401/422 problem body; unrouted 404s carry this app default.
        """
        if not isinstance(cast("object", cors), CORS):
            raise WiringError(f"_include_cors requires a CORS policy, got {type(cors).__name__}")
        if cors is CORS.OFF:
            raise WiringError(
                "CORS.OFF is the per-include opt-out; an app that wants no CORS default "
                "simply does not call _include_cors",
            )
        if self.__cors_default is not None:
            raise WiringError("a CORS default is already registered; an app has at most one")
        self.__cors_default = CompiledCORS(cors)

    def _include_middleware(self, middleware: object) -> None:
        """Register one middleware app-wide: every route (current and later includes) is
        covered, and its intercepts run *pre-routing* — they can answer requests no
        route serves, which is how an OPTIONS-scoped intercept answers preflights for
        paths that would 404.

        The middleware is a structurally typed object — no base class; its hooks
        (``response_headers`` attribute or method, ``intercept`` + ``intercept_methods``,
        ``observe``) are introspected and validated here, fail-loud, and compiled into
        the covered routes at wiring (see :class:`~jero.CORS` for the same idea as a
        built-in). Register order is run order, and globals run before include-scoped
        middleware. Only what a middleware defines costs anything: a constant
        ``response_headers`` is baked into route header blocks for free, an off-scope
        verb never reaches an ``intercept``.
        """
        self.__middleware.append(CompiledMiddleware(middleware))

    # Three overloads rather than one three-way-union signature: a Union of several
    # generic Protocols, each binding its own subset of type parameters, is more than
    # mypy's structural-match inference can solve in one shot (it falls back to `Never`
    # for every parameter and rejects any concrete auth class); overloads give each
    # shape its own clean inference, and every other checker (pyright/ty/zuban/pyrefly)
    # handles both forms equally well. Runtime dispatch is unaffected — the
    # undecorated implementation below is what actually runs.
    @overload
    def _include_resource[THeaders: Struct, TUser: Struct](
        self,
        resource: Resource,
        *,
        auth: Auth[THeaders, TUser] | None = None,
        cors: CORS | None = None,
        middleware: Sequence[object] = (),
    ) -> None: ...
    @overload
    def _include_resource[TCookies: Struct, TUser: Struct](
        self,
        resource: Resource,
        *,
        auth: CookieAuth[TCookies, TUser] | None = None,
        cors: CORS | None = None,
        middleware: Sequence[object] = (),
    ) -> None: ...
    @overload
    def _include_resource[THeaders: Struct, TCookies: Struct, TUser: Struct](
        self,
        resource: Resource,
        *,
        auth: HybridAuth[THeaders, TCookies, TUser] | None = None,
        cors: CORS | None = None,
        middleware: Sequence[object] = (),
    ) -> None: ...
    def _include_resource(
        self,
        resource: Resource,
        *,
        auth: "Auth[Any, Any] | CookieAuth[Any, Any] | HybridAuth[Any, Any, Any] | None" = None,
        cors: CORS | None = None,
        middleware: Sequence[object] = (),
    ) -> None:
        """Register a ``Resource``'s CRUD methods as routes, optionally behind ``auth``.

        An authenticator returning ``TUser`` gates every method: no valid credentials, no
        handler. One returning ``TUser | None`` accepts anonymous callers instead — see
        :meth:`_include_endpoint`.

        ``cors=`` sets this include's :class:`CORS` policy: omitted inherits the
        ``_include_cors`` default, a policy overrides it, ``CORS.OFF`` opts out.
        ``middleware=`` adds include-scoped middleware on top of any registered with
        :meth:`_include_middleware` — scope is deployment policy, so it lives here at
        the mount, not on the class. Scoped intercepts run post-resolve, pre-auth.
        """
        self.__include(resource, Resource.METHODS, auth=auth, cors=cors, middleware=middleware)

    @overload
    def _include_endpoint[THeaders: Struct, TUser: Struct](
        self,
        endpoint: Endpoint,
        *,
        auth: Auth[THeaders, TUser] | None = None,
        cors: CORS | None = None,
        middleware: Sequence[object] = (),
    ) -> None: ...
    @overload
    def _include_endpoint[TCookies: Struct, TUser: Struct](
        self,
        endpoint: Endpoint,
        *,
        auth: CookieAuth[TCookies, TUser] | None = None,
        cors: CORS | None = None,
        middleware: Sequence[object] = (),
    ) -> None: ...
    @overload
    def _include_endpoint[THeaders: Struct, TCookies: Struct, TUser: Struct](
        self,
        endpoint: Endpoint,
        *,
        auth: HybridAuth[THeaders, TCookies, TUser] | None = None,
        cors: CORS | None = None,
        middleware: Sequence[object] = (),
    ) -> None: ...
    def _include_endpoint(
        self,
        endpoint: Endpoint,
        *,
        auth: "Auth[Any, Any] | CookieAuth[Any, Any] | HybridAuth[Any, Any, Any] | None" = None,
        cors: CORS | None = None,
        middleware: Sequence[object] = (),
    ) -> None:
        """Register an ``Endpoint``'s verb methods as routes, optionally behind ``auth``.

        An authenticator returning ``TUser`` gates every verb: no valid credentials, no
        handler. One returning ``TUser | None`` makes credentials an *input* — a caller
        presenting none is served with ``user=None``, *invalid* credentials are still a 401,
        and every handler on the route must declare ``user: TUser | None`` (all checked at
        startup).

        ``cors=`` sets this include's :class:`CORS` policy: omitted inherits the
        ``_include_cors`` default, a policy overrides it, ``CORS.OFF`` opts out.
        ``middleware=`` adds include-scoped middleware on top of any registered with
        :meth:`_include_middleware` — scope is deployment policy, so it lives here at
        the mount, not on the class. Scoped intercepts run post-resolve, pre-auth.
        """
        self.__include(endpoint, Endpoint.METHODS, auth=auth, cors=cors, middleware=middleware)

    @overload
    def _include_websocket[THeaders: Struct, TUser: Struct](
        self,
        endpoint: WebSocketEndpoint,
        *,
        auth: Auth[THeaders, TUser] | None = None,
        max_frame_size: int = 1024 * 1024,
        middleware: Sequence[object] = (),
    ) -> None: ...
    @overload
    def _include_websocket[TCookies: Struct, TUser: Struct](
        self,
        endpoint: WebSocketEndpoint,
        *,
        auth: CookieAuth[TCookies, TUser] | None = None,
        max_frame_size: int = 1024 * 1024,
        middleware: Sequence[object] = (),
    ) -> None: ...
    @overload
    def _include_websocket[THeaders: Struct, TCookies: Struct, TUser: Struct](
        self,
        endpoint: WebSocketEndpoint,
        *,
        auth: HybridAuth[THeaders, TCookies, TUser] | None = None,
        max_frame_size: int = 1024 * 1024,
        middleware: Sequence[object] = (),
    ) -> None: ...
    def _include_websocket(
        self,
        endpoint: WebSocketEndpoint,
        *,
        auth: "Auth[Any, Any] | CookieAuth[Any, Any] | HybridAuth[Any, Any, Any] | None" = None,
        max_frame_size: int = 1024 * 1024,
        middleware: Sequence[object] = (),
    ) -> None:
        """Register one typed WebSocket protocol and compile its handshake contract."""
        if not isinstance(max_frame_size, int) or isinstance(max_frame_size, bool):
            raise WiringError("max_frame_size must be a positive integer")
        if max_frame_size < 1:
            raise WiringError("max_frame_size must be a positive integer")
        cls = type(endpoint)
        path = getattr(cls, "path", None)
        if path is None:
            raise WiringError(
                f"{cls.__name__}: no path — declare it on the class, "
                f"e.g. `class {cls.__name__}(WebSocketEndpoint, path='/...')`.",
            )
        fn = getattr(endpoint, "handle", None)
        if fn is None:
            raise WiringError(f"{cls.__name__} must define handle")
        sources, inbound, outbound = _bind_websocket_sources(cls, fn)
        compiled_auth = _CompiledAuth(auth) if auth is not None else None
        self.__check_user_source(cls, "handle", sources, compiled_auth)
        tail = _RouteTail()
        intercepts: list[_WebSocketInterceptRunner] = []
        for item in middleware:
            compiled = CompiledMiddleware(item)
            if compiled.intercept is None:
                continue
            if "GET" not in compiled.intercept_methods:
                raise WiringError(
                    f"{compiled.owner}.intercept can never run on a WebSocket handshake: "
                    f"intercept_methods is {compiled.intercept_methods!r}, expected GET",
                )
            intercepts.append(
                _InterceptRunner(
                    compiled.intercept,
                    _intercept_sender(
                        compiled.owner,
                        compiled.intercept,
                        self.__reverser,
                        self.__exceptions,
                        tail,
                    ),
                )
            )
        segments = _route_segments(
            cls, "handle", _parse_template(path), sources.path, extends_path=False
        )
        route = _WebSocketRoute(
            fn,
            sources=sources,
            inbound=inbound,
            outbound=outbound,
            auth=compiled_auth,
            exceptions=self.__exceptions,
            intercepts=tuple(intercepts),
            tail=tail,
            max_frame_size=max_frame_size,
        )
        self.__register_websocket(segments, route)

    def _include_openapi(
        self,
        *,
        title: str,
        version: str,
        description: str | None = None,
        openapi_path: str = "/openapi.json",
        docs_path: str | None = "/docs",
        servers: Sequence[str] = (),
        tags: Sequence[Tag] = (),
        docs_html: str | None = None,
        favicon: Path | str | None = None,
        scalar_config: ScalarConfig | None = None,
    ) -> None:
        """Serve an auto-generated OpenAPI 3.1 document and a docs UI.

        Call inside ``wire`` (order among the ``include_*`` calls doesn't matter — the
        document is built once after wiring completes). ``openapi_path`` serves the JSON
        spec; ``docs_path`` serves a Scalar UI pointed at it (pass ``None`` to omit the
        UI, or ``docs_html`` to replace the page — e.g. for offline / strict-CSP hosting).

        ``scalar_config`` is a typed :class:`~jero.ScalarConfig` tuning the Scalar UI — e.g.
        ``scalar_config=ScalarConfig(hide_models=True)`` drops the global Models list, or set a
        ``theme`` / ``layout``. Only its set fields are sent, so Scalar's own defaults apply
        otherwise. For options ``ScalarConfig`` doesn't model, supply a full ``docs_html``
        page instead (``scalar_config`` is ignored when ``docs_html`` is given).

        ``favicon`` gives the docs page an icon. A ``Path`` (the primary case) is read
        once at wiring — a missing/unreadable file or an unsupported suffix
        (``.ico``/``.png``/``.svg``) is a ``WiringError`` — and served as a precomputed
        response at ``/favicon.ico``; no runtime file I/O. A ``str`` is a URL (a
        ``data:`` URI works too), emitted verbatim in the page's ``<link rel="icon">``
        with nothing served. Like the spec routes, ``/favicon.ico`` never appears in the
        generated document. A custom ``docs_html`` page is never modified — reference
        the favicon yourself there.

        ``tags`` declares document-level ``Tag``\\ s to describe operation groups and pin the
        order they appear in the docs UI. Operations may also define/use tags via their
        ``meta`` (a bare name, or a ``Tag`` with a description); all are merged here. The one
        rule: describing the same tag name two different ways is a startup ``WiringError``.

        The spec is derived from your wired resources/endpoints: their typed sources
        (path/query/header params, request bodies), return types (responses), auth
        (security), ``msgspec.Meta`` field constraints, and the metadata you declare —
        ``OperationMeta`` (summary/description/tags/responses) and a model's ``ModelMeta``.
        Docstrings are never published; public prose is always explicit.
        """
        self.__openapi = _OpenAPIConfig(
            title=title,
            version=version,
            description=description,
            servers=tuple(servers),
            tags=tuple(tags),
            openapi_path=openapi_path,
            docs_path=docs_path,
        )
        # The framework routes register through an include record like any other
        # include, so they are covered by the app's CORS default and middleware (a
        # cross-origin tool must be able to fetch the spec; a global security-headers
        # middleware must decorate the docs page). They just never appear in the
        # generated document.
        tail = _RouteTail()
        record = _IncludeRecord(tail=tail, routes=[], cors=None, cors_off=False, middleware=())
        favicon_href: str | None = None
        if isinstance(favicon, Path):
            body, content_type = _favicon_payload(favicon)
            favicon_handler = _static_bytes_handler(body, content_type, tail)
            self.__register("GET", _parse_template("/favicon.ico"), favicon_handler)
            record.routes.append(("GET", favicon_handler))
            favicon_href = "/favicon.ico"
        elif favicon is not None:
            favicon_href = favicon
        doc_handler = _json_doc_handler(self.__openapi, tail)
        self.__register("GET", _parse_template(openapi_path), doc_handler)
        record.routes.append(("GET", doc_handler))
        if docs_path is not None:
            page = (
                docs_html
                if docs_html is not None
                else _scalar_html(title, openapi_path, favicon_href, scalar_config)
            )
            page_handler = _static_bytes_handler(page.encode(), b"text/html; charset=utf-8", tail)
            self.__register("GET", _parse_template(docs_path), page_handler)
            record.routes.append(("GET", page_handler))
        self.__includes.append(record)

    async def _create_background_tasks(
        self,
        *,
        maxsize: int = 1024,
        drain_timeout: float | None = 30.0,
        allow_one_to_many: bool = False,
    ) -> BackgroundTasks:
        """Build a :class:`BackgroundTasks` queue bound to the app's lifecycle.

        Sugar for ``await self._aenter(BackgroundTasks(...))``: the worker starts at
        startup and drains/stops at shutdown. Call inside ``wire`` *after* the services
        its handlers use, so reverse-order shutdown drains the queue before those
        services are torn down.
        """
        return await self._aenter(
            BackgroundTasks(
                maxsize=maxsize,
                drain_timeout=drain_timeout,
                allow_one_to_many=allow_one_to_many,
            )
        )

    def __build_openapi_document(self, config: _OpenAPIConfig) -> bytes:
        """Translate the captured operations into the OpenAPI document, encoded as JSON."""
        operations = tuple(
            operation_input(spec, self.__exceptions.adapter) for spec in self.__operations
        )
        schemes: dict[str, SecurityScheme] = {}
        for spec in self.__operations:
            scheme = spec.security_scheme
            if scheme is None:
                continue
            existing = schemes.get(scheme.scheme_name)
            if existing is not None and existing != scheme:
                raise WiringError(
                    f"two different OpenAPI security schemes share the name "
                    f"{scheme.scheme_name!r}; give each a distinct scheme_name",
                )
            schemes[scheme.scheme_name] = scheme
        info = Info(
            title=config.title,
            version=config.version,
            description=config.description,
            servers=config.servers,
            tags=_assemble_openapi_tags(config.tags, operations),
        )
        try:
            document = build_openapi(info, operations, schemes)
        except OpenAPINameConflictError as exc:
            # A ModelMeta(name=...) override collided with another component's name.
            raise WiringError(str(exc)) from exc
        except KeyError as exc:
            # msgspec.json.schema_components keys components by type name and raises KeyError
            # when two distinct Structs share a name in the same module.
            raise WiringError(
                f"could not generate the OpenAPI schema for {exc}; two different msgspec "
                f"Structs likely share a name in the same module — rename one",
            ) from exc
        return msgspec_encoder.encode(document)

    def __resolve_dynamic(self, method: str, path: str) -> tuple[_Handler, dict[str, str]] | None:
        # Static routes never reach here: __call__ resolves them with an inlined dict
        # lookup. The cast is paid only on this, the dynamic path.
        segments = path.split("/")
        verb = cast("HTTPMethod", method)
        for pattern in self.__dynamic.get((verb, len(segments)), ()):
            # Inlines pattern.matches (kept for the cold Allow path): a genexpr per
            # candidate is measurable here. unquote only when a segment is escaped.
            for i, value in pattern.statics:
                if segments[i] != value:
                    break
            else:
                values: dict[str, str] = {}
                for i, name in pattern.params:
                    segment = segments[i]
                    values[name] = unquote(segment) if "%" in segment else segment
                return pattern.handler, values
        return None

    def __resolve_websocket_dynamic(
        self, path: str
    ) -> tuple[_WebSocketHandler, dict[str, str]] | None:
        segments = path.split("/")
        for pattern in self.__websocket_dynamic.get(len(segments), ()):
            for index, value in pattern.statics:
                if segments[index] != value:
                    break
            else:
                values: dict[str, str] = {}
                for index, name in pattern.params:
                    segment = segments[index]
                    values[name] = unquote(segment) if "%" in segment else segment
                return pattern.handler, values
        return None

    def __allowed_methods(self, path: str) -> tuple[HTTPMethod, ...]:
        allowed = list(self.__allowed.get(path, ()))
        segments = path.split("/")
        for (method, count), bucket in self.__dynamic.items():
            if (
                count == len(segments)
                and method not in allowed
                and any(pattern.matches(segments) for pattern in bucket)
            ):
                allowed.append(method)
        return tuple(allowed)

    def __allow_for(self, path: str) -> bytes | None:
        """The Allow header for a path, or None if no route shape matches it."""
        cached = self.__allow_cache.get(path)
        if cached is not None:
            return cached
        allowed = self.__allowed_methods(path)
        return _allow_header(allowed) if allowed else None

    def __preflight_pairs(self, scope: Scope, path: str) -> list[tuple[bytes, bytes]] | None:
        """The CORS pairs for a preflight OPTIONS, or None when the request isn't one
        (no ``Access-Control-Request-Method``) or nothing answers it.

        The requested method selects *which route's* policy replies — the answer is per
        (path, requested method), so ``GET`` public / ``POST`` restricted on one path
        works. Preflights carry no credentials, so this runs before any auth would."""
        requested = requested_method(scope)
        if requested is None:
            return None
        # HEAD is served from GET routes, but the *policy* check still sees "HEAD".
        verb = "GET" if requested == "HEAD" else requested
        handler = self.__static.get((verb, path))
        if handler is None:
            resolved = self.__resolve_dynamic(verb, path)
            if resolved is None:
                return None
            handler = resolved[0]
        cors = self.__route_cors.get(handler)
        if cors is None:
            return None
        return cors.preflight_pairs(scope, requested)

    def __log_openapi_docs(self, config: _OpenAPIConfig) -> None:
        """Announce where the docs/spec are served, once, at startup.

        jero is the ASGI app, not the server, so it can't know the bound host/port — the
        URL is absolute only when ``JERO_BASE_URL`` names the public origin, otherwise the
        path is relative (the server prints its own ``Listening at`` line with the host).
        """
        base = (os.environ.get("JERO_BASE_URL") or "").rstrip("/")
        if config.docs_path is not None:
            logger.info("Serving API docs at %s%s", base, config.docs_path)
        else:
            logger.info("Serving OpenAPI spec at %s%s", base, config.openapi_path)

    def __fill_tail(
        self,
        tail: _RouteTail,
        cors: CompiledCORS | None,
        middlewares: Sequence[CompiledMiddleware],
    ) -> None:
        """Fill one response-header tail: the CORS policy first, then each middleware's
        constant pairs and ``response_headers`` method hooks (globals before scoped, in
        registration order). Duplicate *constant* names across contributors fail loud."""
        claims: dict[str, str] = {}
        if cors is not None:
            _claim_header_names(claims, "CORS", cors.constant_pairs)
            tail.pairs += cors.constant_pairs
            if cors.dynamic is not None:
                tail.dynamic += (cors.dynamic,)
        for mw in middlewares:
            if mw.constant_headers is not None:
                pairs = _constant_middleware_pairs(mw.constant_headers)
                _claim_header_names(claims, mw.owner, pairs)
                tail.pairs += pairs
            if mw.headers_hook is not None:
                tail.dynamic += (_HeadersTailHook(mw.headers_hook),)
        tail.active = bool(tail.pairs or tail.dynamic)

    def __build_global_middleware(self) -> None:
        """Build the pre-routing machinery: the verb-keyed global intercept table and
        the global observes. ``__pre`` stays None when neither exists — a middleware
        with only header tiers has no pre-routing presence at all."""
        table: dict[str, list[_InterceptRunner]] = {}
        for mw in self.__middleware:
            if mw.intercept is None:
                continue
            runner = _InterceptRunner(
                mw.intercept,
                _intercept_sender(
                    mw.owner, mw.intercept, self.__reverser, self.__exceptions, self.__app_tail
                ),
            )
            for method in mw.intercept_methods:
                table.setdefault(method, []).append(runner)
        observes = tuple(mw.observe for mw in self.__middleware if mw.observe is not None)
        if table or observes:
            self.__pre = _GlobalMiddleware(
                {method: tuple(runners) for method, runners in table.items()}, observes
            )

    @staticmethod
    def __check_scoped_intercepts(record: _IncludeRecord) -> None:
        """A scoped intercept none of the include's wire methods can ever trigger is a
        dead registration — fail loud rather than silently never running.

        A *partial* overlap stays legal: one middleware class declaring
        ``("GET", "POST")`` is reusable across a GET-only and a POST-only include, each
        picking up the verbs it serves. The exception is ``OPTIONS``, which is answered
        before routing on every path — an OPTIONS entry is dead on *any* include, so it
        is rejected outright rather than left to overlap luck."""
        served = {method for method, _ in record.routes}
        if "GET" in served:
            served.add("HEAD")
        for mw in record.middleware:
            if mw.intercept is None:
                continue
            if "OPTIONS" in mw.intercept_methods:
                raise WiringError(
                    f"{mw.owner}.intercept declares OPTIONS in intercept_methods, but "
                    f"OPTIONS never reaches include-scoped middleware — intercept it "
                    f"globally via _include_middleware",
                )
            if not served.intersection(mw.intercept_methods):
                raise WiringError(
                    f"{mw.owner}.intercept can never run on this include: it declares "
                    f"intercept_methods {mw.intercept_methods!r} but the include serves "
                    f"{', '.join(sorted(served))}",
                )

    def __swap_handler(self, old: _Handler, new: _Handler) -> None:
        """Replace one compiled route in the routing tables (wiring-time only) — how a
        covered route's ``_CoveredRoute`` wrapper takes its place, so uncovered routes
        never pay so much as a branch for middleware."""
        for key, handler in self.__static.items():
            if handler is old:
                self.__static[key] = new
        for bucket in self.__dynamic.values():
            for index, pattern in enumerate(bucket):
                if pattern.handler is old:
                    bucket[index] = _Pattern(pattern.statics, pattern.params, new)

    def __cover_route(
        self,
        record: _IncludeRecord,
        verb: HTTPMethod,
        route: _Handler,
        scoped_runners: Sequence[tuple[CompiledMiddleware, _InterceptRunner]],
        observes: tuple[ObserveHook, ...],
        *,
        headers_hooks: bool,
    ) -> _Handler:
        """Wrap one route in a ``_CoveredRoute`` when any middleware hook covers it
        (an intercept scoped to one of its wire methods, an observe, or a dynamic
        ``response_headers`` needing the ``received_at`` stamp); uncovered routes are
        returned untouched."""
        wire_methods = ("GET", "HEAD") if verb == "GET" else (verb,)
        table: dict[str, list[_InterceptRunner]] = {}
        for mw, runner in scoped_runners:
            for method in mw.intercept_methods:
                if method in wire_methods:
                    table.setdefault(method, []).append(runner)
        if not table and not observes and not headers_hooks:
            return route
        compiled = _RouteMiddleware(
            {method: tuple(runners) for method, runners in table.items()} if table else None,
            observes,
        )
        covered = _CoveredRoute(route, compiled, self.__exceptions, record.tail)
        self.__swap_handler(route, covered)
        return covered

    def __finalize_tails(self) -> None:
        """Resolve every include's response-header tail (the CORS policy it declares or
        inherits, plus middleware header tiers), the middleware coverage of each route,
        and the app-level tail + pre-routing table for unrouted responses. One pass
        after wiring, so ``_include_cors`` / ``_include_middleware`` and the include
        calls compose in any order."""
        default = self.__cors_default
        # Unrouted responses (404, 405) carry the app default and global middleware only.
        self.__fill_tail(self.__app_tail, default, self.__middleware)
        self.__build_global_middleware()
        global_observes = self.__pre.observes if self.__pre is not None else ()
        for record in self.__includes:
            self.__check_scoped_intercepts(record)
            cors = None if record.cors_off else (record.cors or default)
            middlewares = (*self.__middleware, *record.middleware)
            self.__fill_tail(record.tail, cors, middlewares)
            headers_hooks = any(mw.headers_hook is not None for mw in middlewares)
            scoped_runners = [
                (
                    mw,
                    _InterceptRunner(
                        mw.intercept,
                        _intercept_sender(
                            mw.owner, mw.intercept, self.__reverser, self.__exceptions, record.tail
                        ),
                    ),
                )
                for mw in record.middleware
                if mw.intercept is not None
            ]
            observes = global_observes + tuple(
                mw.observe for mw in record.middleware if mw.observe is not None
            )
            for verb, route in record.routes:
                handler = self.__cover_route(
                    record, verb, route, scoped_runners, observes, headers_hooks=headers_hooks
                )
                if cors is not None:
                    self.__route_cors[handler] = cors

    def __finalize(self) -> None:
        """Precompute Allow headers, resolve response-header tails, and build the OpenAPI
        document; runs once after wiring."""
        self.__finalize_tails()
        self.__allow_cache = {
            path: _allow_header(self.__allowed_methods(path)) for path in self.__allowed
        }
        if self.__openapi is not None:
            self.__openapi.payload = self.__build_openapi_document(self.__openapi)
            self.__log_openapi_docs(self.__openapi)

    async def __close_resources(self) -> None:
        await self.__astack.aclose()
        self.__stack.close()

    async def __handle_lifespan(self, receive: Receive, send: Send) -> None:
        await receive()  # lifespan.startup
        try:
            await self.wire()
            self.__finalize()  # builds the OpenAPI doc; can raise WiringError (e.g. tag conflict)
        except BaseException as exc:
            await self.__close_resources()  # release anything entered before the failure
            await send(
                {
                    "type": "lifespan.startup.failed",
                    "message": f"{type(exc).__name__}: {exc}",
                },
            )
            raise
        await send({"type": "lifespan.startup.complete"})

        await receive()  # lifespan.shutdown
        try:
            await self.__close_resources()
        except BaseException as exc:
            await send(
                {
                    "type": "lifespan.shutdown.failed",
                    "message": f"{type(exc).__name__}: {exc}",
                },
            )
            raise
        await send({"type": "lifespan.shutdown.complete"})

    async def __intercept_global(
        self,
        pre: _GlobalMiddleware,
        runners: tuple[_InterceptRunner, ...],
        scope: Scope,
        receive: Receive,
        send: Send,
    ) -> bool:
        """Run one verb's global intercepts, pre-routing (they can answer requests no
        route serves — preflights for paths that would 404). True when one answered:
        the response (or its error) has left and the global observes have seen it;
        False falls through to routing untouched."""
        received_at = perf_counter()
        scope["jero.received_at"] = received_at
        if scope["method"] == "HEAD":
            send = _SuppressBody(send)
        capture: _StatusCapture | None = None
        if pre.observes:
            send = capture = _StatusCapture(send)
        answered = await _run_intercepts(
            runners, scope, receive, send, exceptions=self.__exceptions, tail=self.__app_tail
        )
        if answered and capture is not None:
            duration = capture.started_at - received_at if capture.started_at else 0.0
            for observe in pre.observes:
                await observe(scope, capture.status, duration)
        return answered

    async def __fallthrough(self, scope: Scope, send: Send, method: str, path: str) -> None:
        """Answer a request no route serves. No route owns these responses, so the 404
        and 405 problems carry the app-level tail (the CORS default and global
        middleware headers); the OPTIONS answer adds the CORS preflight block when the
        requested method's route has a policy."""
        allow = self.__allow_for(path)
        error: HTTPError
        if allow is None:
            error = NotFoundError()
            await _send_json(
                send,
                error.status,
                self.__exceptions.encode_error(error),
                self.__app_tail.contained_extra(scope),
            )
        elif method == "OPTIONS":
            headers = [(b"allow", allow)]
            if self.__route_cors:
                preflight = self.__preflight_pairs(scope, path)
                if preflight is not None:
                    headers += preflight
            await send({"type": "http.response.start", "status": 204, "headers": headers})
            await send({"type": "http.response.body", "body": b""})
        else:
            error = MethodNotAllowedError()
            extra = self.__app_tail.contained_extra(scope)
            await _send_json(
                send,
                error.status,
                self.__exceptions.encode_error(error),
                [(b"allow", allow)] + (extra if extra is not None else []),
            )

    async def __handle_websocket(self, scope: Scope, receive: Receive, send: Send) -> None:
        connect = await receive()
        if connect["type"] != "websocket.connect":
            await send({"type": "websocket.close", "code": 1008, "reason": "bad handshake"})
            return
        # WebSocket handshakes are GET requests for middleware verb scoping. Global
        # intercepts run before routing/auth exactly as on the HTTP path; observe is
        # deliberately absent for sockets in v1.
        scope["method"] = "GET"
        pre = self.__pre
        if pre is not None:
            runners = pre.intercepts.get("GET")
            if runners is not None and await _run_intercepts(
                runners,
                scope,
                receive,
                _WebSocketDenialSend(send, _supports_websocket_denial(scope)),
                exceptions=self.__exceptions,
                tail=self.__app_tail,
            ):
                return
        path: str = scope["path"]
        handler = self.__websocket_static.get(path)
        path_values: dict[str, str] = {}
        if handler is None:
            resolved = self.__resolve_websocket_dynamic(path)
            if resolved is not None:
                handler, path_values = resolved
        if handler is not None:
            await handler(scope, receive, send, path_values)
            return
        error = NotFoundError()
        payload = self.__exceptions.encode_error(error)
        await _send_websocket_rejection(scope, send, error.status, payload)

    async def __handle_non_http(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] == "lifespan":
            await self.__handle_lifespan(receive, send)
            return
        if scope["type"] == "websocket":
            await self.__handle_websocket(scope, receive, send)
            return
        raise RuntimeError(f"unsupported scope type {scope['type']!r}")

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.__handle_non_http(scope, receive, send)
            return

        # HTTP is the hot path; inlined here (was _handle_http) to save a coroutine hop.
        method: str = scope["method"]
        path: str = scope["path"]
        # Global intercepts run pre-routing; the table is verb-keyed, so an off-scope
        # method costs one dict hit and apps without them one attribute load.
        pre = self.__pre
        if pre is not None:
            runners = pre.intercepts.get(method)
            if runners is not None and await self.__intercept_global(
                pre, runners, scope, receive, send
            ):
                return
        verb = "GET" if method == "HEAD" else method
        # A static hit is the hottest path of all: one dict lookup, inlined here to skip
        # the resolver call (a non-route verb simply misses).
        handler = self.__static.get((verb, path))
        path_values: dict[str, str] = {}
        if handler is None:
            resolved = self.__resolve_dynamic(verb, path)
            if resolved is not None:
                handler, path_values = resolved
        if handler is not None:
            await handler(
                scope, receive, _SuppressBody(send) if method == "HEAD" else send, path_values
            )
            return
        if pre is None or not pre.observes:
            await self.__fallthrough(scope, send, method, path)
            return
        # Global observes see fallthrough answers too — capture the outcome around it.
        received_at = perf_counter()
        scope["jero.received_at"] = received_at
        capture = _StatusCapture(send)
        await self.__fallthrough(scope, capture, method, path)
        duration = capture.started_at - received_at if capture.started_at else 0.0
        for observe in pre.observes:
            await observe(scope, capture.status, duration)