Run backend as module (python -m) in devserver instead of PATH-installed CLI.

run() gains access_log (default on) and log_config parameters.  When
enabled, uvicorn's own access log is disabled (access_log=False) and
replaced by an ASGI middleware (adapted from the uvicorn access-log
branch) that logs colored request lines, WebSocket open/close pairs,
and pre-accept rejections in open format.

log_config dicts (uvicorn's default or user-supplied) are patched
best-effort via a registered patch pipeline in fastapi_vue.logging:
swapping in our AccessFormatter, routing our output to a private
fastapi_vue.access logger (uvicorn gates its protocol-level access
logging on uvicorn.access.hasHandlers(), so that logger must stay
handlerless), and filtering stock WebSocket handshake chatter from
uvicorn.error.  Non-dict or structurally unexpected configs pass
through untouched.

The middleware is installed by patching Config.load/load_app, hooked
into AccessFormatter instantiation so reload/worker subprocesses
(which re-apply log_config without re-calling run()) are covered.

README.md

@@ -84,7 +84,7 @@ my-app/
 └── scripts/
     ├── devserver.py             # Run Vite and FastAPI together in dev mode
     └── fastapi-vue/             # Dev utilities (only on the source tree)
-        ├── build-frontend.py
+        ├── buildhook.py
         ├── buildutil.py
         └── devutil.py
 ```

fastapi-vue/fastapi_vue/__init__.py

@@ -1,3 +1,5 @@
+"""FastAPI Vue integration - serve Vue frontend from FastAPI."""
+
 from .staticfiles import Frontend
 
 __all__ = ["Frontend"]

fastapi-vue/fastapi_vue/accesslog.py

@@ -0,0 +1,418 @@
+"""HTTP/WebSocket access logging ASGI middleware."""
+
+from __future__ import annotations
+
+import http
+import itertools
+import logging
+import time
+from ipaddress import IPv6Address
+from typing import TYPE_CHECKING, cast
+
+from .termwidth import pad_display
+
+if TYPE_CHECKING:
+    from uvicorn._types import (
+        ASGI3Application,
+        ASGIReceiveCallable,
+        ASGIReceiveEvent,
+        ASGISendCallable,
+        ASGISendEvent,
+        Scope,
+        WWWScope,
+    )
+
+logger = logging.getLogger("fastapi_vue.access")
+
+# Terminal color codes
+_RESET = "\033[0m"
+_STATUS_INFO = "\033[32m"  # 1xx (green)
+_STATUS_OK = "\033[1;92m"  # 2xx (bright green)
+_STATUS_REDIRECT = "\033[32m"  # 3xx (green)
+_STATUS_CLIENT_ERR = "\033[0;31m"  # 4xx (red)
+_STATUS_SERVER_ERR = "\033[1;91m"  # 5xx (bold bright red)
+_METHOD_READ = "\033[0;34m"  # GET, HEAD, OPTIONS (blue)
+_METHOD_WRITE = "\033[1;94m"  # POST, PUT, DELETE, PATCH (bold bright blue)
+_HOST = "\033[38;5;242m"  # hostname (dark grey)
+_PATH = "\033[38;5;250m"  # path (white)
+_TIMING = "\033[38;5;242m"  # timing/devmode (dark grey)
+_WS_OPEN = "\033[38;5;226m"  # WebSocket connect (brightest yellow)
+_WS_CLOSE = "\033[38;5;142m"  # WebSocket disconnect (dimmer yellow)
+
+
+def _format_duration(duration: float) -> str:
+    ms = int(duration * 1000)
+    if ms < 2000:
+        return f"{ms}ms"
+
+    total_seconds = ms // 1000
+    if total_seconds < 60:
+        return f"{total_seconds}s"
+
+    if total_seconds < 3600:
+        minutes, seconds = divmod(total_seconds, 60)
+        return f"{minutes}m{seconds}s"
+
+    hours, remainder = divmod(total_seconds, 3600)
+    minutes = remainder // 60
+    return f"{hours}h{minutes}m"
+
+
+def _status_color(status: int) -> str:
+    if status < 200:
+        return _STATUS_INFO
+    if status < 300:
+        return _STATUS_OK
+    if status < 400:
+        return _STATUS_REDIRECT
+    if status < 500:
+        return _STATUS_CLIENT_ERR
+    return _STATUS_SERVER_ERR
+
+
+def _method_color(method: str) -> str:
+    return _METHOD_READ if method in ("GET", "HEAD", "OPTIONS") else _METHOD_WRITE
+
+
+def _format_extra_timing(extra: str = "", duration: float | None = None) -> tuple[str, str]:
+    timing = _format_duration(duration) if duration is not None else ""
+    return (f"{extra} " if extra else "", f"{_TIMING}{timing}{_RESET}" if timing else "")
+
+
+def _format_ipv6_network(ip: str) -> str:
+    try:
+        ip = ip.strip("[]")
+        if "%" in ip:
+            ip = ip.split("%")[0]
+        addr = IPv6Address(ip)
+
+        if addr.is_loopback:
+            return "::1"
+        if addr.is_unspecified:
+            return "::"
+        if addr.ipv4_mapped:
+            return str(addr.ipv4_mapped)
+        if addr.is_link_local:
+            return str(addr)
+
+        network_int = int(addr) >> 64
+        groups: list[str] = []
+        for _ in range(4):
+            groups.insert(0, format(network_int & 0xFFFF, "x"))
+            network_int >>= 16
+        result = ":".join(groups) + "::"
+        return str(IPv6Address(result + "0")).removesuffix("::")
+    except ValueError:
+        return ip
+
+
+def _format_client_ip(ip: str) -> str:
+    if not ip or ip == "-":
+        return "-"
+    stripped = ip.strip("[]")
+    if ":" in stripped:
+        return _format_ipv6_network(ip)
+    return ip
+
+
+def _header(scope: WWWScope, name: str) -> str | None:
+    name_bytes = name.lower().encode("latin-1")
+    for key, value in scope["headers"]:
+        if key.lower() == name_bytes:
+            return value.decode("latin-1")
+    return None
+
+
+def _client_host(scope: WWWScope) -> str:
+    client = scope["client"]
+    return client[0] if client else "-"
+
+
+def _path(scope: WWWScope) -> str:
+    path = scope["path"]
+    query = scope["query_string"]
+    if query:
+        return f"{path}?{query.decode('latin-1')}"
+    return path
+
+
+# WebSocket connection counter (mod 100)
+_ws_counter = itertools.count()
+
+
+def _next_ws_id() -> str:
+    return f"{next(_ws_counter) % 100:02d}"
+
+
+WS_CLOSE_CODES = {
+    1000: "ok",
+    1001: "going away",
+    1002: "protocol error",
+    1003: "unsupported",
+    1005: "no status",
+    1006: "abnormal",
+    1007: "invalid data",
+    1008: "policy violation",
+    1009: "too large",
+    1010: "extension required",
+    1011: "server error",
+    1012: "restarting",
+    1013: "try again",
+    1014: "bad gateway",
+    1015: "tls error",
+}
+
+
+def _http_access_log_extra(
+    scope: WWWScope,
+    status: int,
+    duration: float,
+    extra: str = "",
+    method: str | None = None,
+) -> dict[str, object]:
+    client_addr = _client_host(scope)
+    full_path = _path(scope)
+    method = method if method is not None else cast("str", scope.get("method", "-"))
+    method = cast("str", scope.get("state", {}).get("access_log_method") or method)
+
+    try:
+        status_phrase = http.HTTPStatus(status).phrase
+    except ValueError:
+        status_phrase = ""
+
+    extra, timing = _format_extra_timing(extra, duration)
+
+    return {
+        "client": _format_client_ip(client_addr).ljust(19),
+        "status": f"{_status_color(status)}{str(status).rjust(3)}{_RESET}",
+        "method": (
+            f"{_METHOD_READ}{pad_display('🔌', 7)}{_RESET}"
+            if method == "🔌"
+            else f"{_method_color(method)}{pad_display(method, 7)}{_RESET}"
+        ),
+        "host": f"{_HOST}{_header(scope, 'host') or '-'}{_RESET}",
+        "path": f"{_PATH}{full_path}{_RESET}",
+        "extra": extra,
+        "timing": timing,
+        "client_addr": client_addr,
+        "status_code": f"{status} {status_phrase}",
+        "request_line": f"{method} {full_path} HTTP/{scope.get('http_version', '-')}",
+        "http_version": scope.get("http_version", "-"),
+        "full_path": full_path,
+    }
+
+
+def _ws_open_extra(
+    scope: WWWScope,
+    ws_id: str,
+    origin: str | None,
+    extra: str = "",
+) -> dict[str, object]:
+    client_addr = _client_host(scope)
+    path = scope.get("path", "")
+    full_path = _path(scope)
+
+    origin_host = origin.split("://", 1)[-1] if origin else None
+    extra, timing = _format_extra_timing(extra)
+
+    host = _header(scope, "host")
+    path = f"{_PATH}{path}{_RESET}"
+    if origin_host and origin_host != host:
+        path += f" {_RESET}from {_HOST}{origin_host}{_RESET}"
+    return {
+        "client": _format_client_ip(client_addr).ljust(19),
+        "status": f"{_WS_OPEN} {ws_id}{_RESET}",
+        "method": f"{_METHOD_READ}{pad_display('🔌', 7)}{_RESET}",
+        "host": f"{_HOST}{host}{_RESET}" if host else "",
+        "path": path,
+        "extra": extra,
+        "timing": timing,
+        "client_addr": client_addr,
+        "status_code": "",
+        "request_line": f"WebSocket {path}",
+        "http_version": scope.get("http_version", "-"),
+        "full_path": full_path,
+    }
+
+
+def _ws_close_extra(
+    scope: WWWScope,
+    ws_id: str,
+    close_code: int | None,
+    duration: float,
+    extra: str = "",
+) -> dict[str, object]:
+    client_addr = _client_host(scope)
+    path = scope.get("path", "-")
+    full_path = _path(scope)
+
+    if close_code is None:
+        code, status_text = "----", "unknown"
+    else:
+        code = str(close_code)
+        status_text = WS_CLOSE_CODES.get(close_code, f"code {close_code}")
+
+    extra, timing = _format_extra_timing(extra, duration)
+
+    return {
+        "client": " " * 19,
+        "status": f"{_WS_CLOSE} {ws_id}{_RESET}",
+        "method": f"{_TIMING}{pad_display('closed', 7)}{_RESET}",
+        "host": "",
+        "path": f"{code} {status_text}",
+        "extra": extra,
+        "timing": timing,
+        "client_addr": client_addr,
+        "status_code": f"{code} {status_text}",
+        "request_line": f"WebSocket {path}",
+        "http_version": scope.get("http_version", "-"),
+        "full_path": full_path,
+    }
+
+
+def _ws_reject_extra(
+    scope: WWWScope,
+    close_code: int | None,
+    duration: float,
+    extra: str = "",
+) -> dict[str, object]:
+    """Open-format line for a connection closed before accept.
+
+    Replaces the open line (which never happened), so the client IP, host and
+    path stay visible.  No connection id is printed (ids are only assigned on
+    accept); the status column shows a dim ``--`` and the close reason rides
+    in the extra column.
+    """
+    if close_code is None:
+        code, status_text = "----", "unknown"
+    else:
+        code = str(close_code)
+        status_text = WS_CLOSE_CODES.get(close_code, f"code {close_code}")
+
+    fields = _ws_open_extra(scope, "--", _header(scope, "origin"))
+    reason = f"closed {code} {status_text}"
+    extra, timing = _format_extra_timing(f"{reason} {extra}" if extra else reason, duration)
+    fields.update(
+        {
+            "status": f"{_WS_CLOSE} --{_RESET}",
+            "extra": extra,
+            "timing": timing,
+            "status_code": f"{code} {status_text}",
+        }
+    )
+    return fields
+
+
+def _assemble_access_log(fields: dict[str, object]) -> str:
+    return (
+        f"{fields['client']} {fields['status']} {fields['method']}"
+        f"{fields['host']}{fields['path']}{fields['extra']}{fields['timing']}"
+    )
+
+
+class AccessLogMiddleware:
+    """ASGI middleware logging HTTP and WebSocket access with colored fields."""
+
+    def __init__(self, app: ASGI3Application) -> None:
+        """Store the wrapped app."""
+        self.app = app
+
+    async def __call__(
+        self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
+    ) -> None:
+        """Dispatch by scope type to HTTP/WebSocket access logging."""
+        if scope["type"] == "http":
+            return await self._handle_http(scope, receive, send)
+        if scope["type"] == "websocket":
+            return await self._handle_websocket(scope, receive, send)
+        return await self.app(scope, receive, send)
+
+    async def _handle_http(
+        self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
+    ) -> None:
+        start = time.perf_counter()
+        www_scope = cast("WWWScope", scope)
+
+        async def wrapped_send(message: ASGISendEvent) -> None:
+            if message["type"] == "http.response.start":
+                fields = _http_access_log_extra(
+                    www_scope,
+                    status=message["status"],
+                    duration=time.perf_counter() - start,
+                    extra=www_scope.get("state", {}).get("log_extra", ""),
+                )
+                logger.info(
+                    '%s - "%s" %s',
+                    fields["client_addr"],
+                    fields["request_line"],
+                    fields["status_code"],
+                    extra=fields,
+                )
+            await send(message)
+
+        return await self.app(scope, receive, wrapped_send)
+
+    async def _handle_websocket(
+        self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
+    ) -> None:
+        start = time.perf_counter()
+        ws_id: str | None = None  # assigned on accept; rejects print "--"
+        accepted = False
+        closed = False
+
+        www_scope = cast("WWWScope", scope)
+        origin = _header(www_scope, "origin")
+
+        def _extra() -> str:
+            return www_scope.get("state", {}).get("log_extra", "")
+
+        def _close_fields(message: ASGIReceiveEvent | ASGISendEvent) -> dict[str, object]:
+            if accepted:
+                assert ws_id is not None  # noqa: S101  # guaranteed once accepted
+                return _ws_close_extra(
+                    www_scope,
+                    ws_id,
+                    message.get("code"),
+                    time.perf_counter() - start,
+                    _extra(),
+                )
+            return _ws_reject_extra(
+                www_scope,
+                message.get("code"),
+                time.perf_counter() - start,
+                _extra(),
+            )
+
+        async def wrapped_send(message: ASGISendEvent) -> None:
+            nonlocal accepted, closed, ws_id
+            if message["type"] == "websocket.accept" and not accepted:
+                accepted = True
+                ws_id = _next_ws_id()
+                fields = _ws_open_extra(www_scope, ws_id, origin, _extra())
+                logger.info(_assemble_access_log(fields), extra=fields)
+            elif message["type"] == "websocket.http.response.start" and not closed:
+                closed = True
+                fields = _http_access_log_extra(
+                    www_scope,
+                    status=message["status"],
+                    duration=time.perf_counter() - start,
+                    extra=_extra(),
+                    method="🔌",
+                )
+                logger.info(_assemble_access_log(fields), extra=fields)
+            elif message["type"] == "websocket.close" and not closed:
+                closed = True
+                fields = _close_fields(message)
+                logger.info(_assemble_access_log(fields), extra=fields)
+            await send(message)
+
+        async def wrapped_receive() -> ASGIReceiveEvent:
+            nonlocal closed
+            message = await receive()
+            if message["type"] == "websocket.disconnect" and not closed:
+                closed = True
+                fields = _close_fields(message)
+                logger.info(_assemble_access_log(fields), extra=fields)
+            return message
+
+        return await self.app(scope, wrapped_receive, wrapped_send)

fastapi-vue/fastapi_vue/hostutil.py

@@ -1,8 +1,60 @@
+"""Parse endpoint strings for uvicorn server configuration."""
+
 import contextlib
 import ipaddress
 from urllib.parse import urlparse
 
 
+def _parse_all_interfaces(value: str) -> list[dict] | None:
+    """Parse ':port' format to bind all interfaces."""
+    if not (value.startswith(":") and value != ":"):
+        return None
+    port_part = value[1:]
+    if not port_part.isdigit():
+        msg = f"Invalid port in '{value}'"
+        raise SystemExit(msg)
+    port = int(port_part)
+    return [{"host": "0.0.0.0", "port": port}, {"host": "::", "port": port}]  # noqa: S104
+
+
+def _parse_unix_socket(value: str) -> list[dict] | None:
+    """Parse UNIX domain socket paths."""
+    if value.startswith("/"):
+        return [{"uds": value}]
+    if value.startswith("unix:"):
+        uds_path = value[5:] or None
+        if uds_path is None:
+            msg = "unix: path must not be empty"
+            raise SystemExit(msg)
+        return [{"uds": uds_path}]
+    return None
+
+
+def _parse_unbracketed_ipv6(value: str, default_port: int) -> list[dict] | None:
+    """Parse unbracketed IPv6 addresses."""
+    if value.count(":") <= 1 or value.startswith("["):
+        return None
+    try:
+        ipaddress.IPv6Address(value)
+    except ValueError as e:
+        msg = f"Invalid IPv6 address '{value}': {e}"
+        raise SystemExit(msg) from e
+    return [{"host": value, "port": default_port}]
+
+
+def _parse_host_port(value: str, default_port: int) -> list[dict]:
+    """Parse host[:port] or [ipv6][:port] using urlparse."""
+    parsed = urlparse(f"//{value}")  # // prefix lets urlparse treat it as netloc
+    host = parsed.hostname or "localhost"
+    port = parsed.port or default_port
+
+    # Validate IP literals (optional; hostname passes through)
+    with contextlib.suppress(ValueError):
+        ipaddress.ip_address(host)
+
+    return [{"host": host, "port": port}]
+
+
 def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
     """Parse an endpoint string into uvicorn bind configurations.
 
@@ -12,6 +64,7 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
 
     Returns:
         List of dicts with uvicorn bind kwargs (host/port or uds).
+        Two entries may be returned for IPv4 and IPv6 (all interaces).
 
     Supported forms:
     - None or empty -> [{host: "localhost", port: default_port}]
@@ -22,6 +75,7 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
     - [ipv6]:port -> [{host: ipv6, port}]
     - ipv6 (unbracketed) -> [{host: ipv6, port: default_port}]
     - /path or unix:/path -> [{uds: path}]
+
     """
     if not value:
         return [{"host": "localhost", "port": default_port}]
@@ -30,38 +84,36 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
     if value.isdigit():
         return [{"host": "localhost", "port": int(value)}]
 
-    # Leading colon :port -> bind all interfaces (0.0.0.0 + ::)
+    # Try specialized parsers in order
-    if value.startswith(":") and value != ":":
+    result = _parse_all_interfaces(value)
-        port_part = value[1:]
+    if result is not None:
-        if not port_part.isdigit():
+        return result
-            raise SystemExit(f"Invalid port in '{value}'")
-        port = int(port_part)
-        return [{"host": "0.0.0.0", "port": port}, {"host": "::", "port": port}]  # noqa: S104
 
-    # UNIX domain socket (unix:/path or just /path)
+    result = _parse_unix_socket(value)
-    if value.startswith("/"):
+    if result is not None:
-        return [{"uds": value}]
+        return result
-    if value.startswith("unix:"):
-        uds_path = value[5:] or None
-        if uds_path is None:
-            raise SystemExit("unix: path must not be empty")
-        return [{"uds": uds_path}]
 
-    # Unbracketed IPv6 (cannot safely contain a port) -> detect by multiple colons
+    result = _parse_unbracketed_ipv6(value, default_port)
-    if value.count(":") > 1 and not value.startswith("["):
+    if result is not None:
-        try:
+        return result
-            ipaddress.IPv6Address(value)
-        except ValueError as e:
-            raise SystemExit(f"Invalid IPv6 address '{value}': {e}") from e
-        return [{"host": value, "port": default_port}]
 
-    # Use urllib.parse for everything else (host[:port], [ipv6][:port])
+    # Fallback: host[:port], [ipv6][:port]
-    parsed = urlparse(f"//{value}")  # // prefix lets urlparse treat it as netloc
+    return _parse_host_port(value, default_port)
-    host = parsed.hostname or "localhost"
-    port = parsed.port or default_port
 
-    # Validate IP literals (optional; hostname passes through)
-    with contextlib.suppress(ValueError):
-        ipaddress.ip_address(host)
 
-    return [{"host": host, "port": port}]
+def parse_endpoints(
+    listen: str | list[str] | None = None,
+    default_port: int = 8000,
+) -> list[dict]:
+    """Parse listen strings into a list of endpoint dicts.
+
+    Args:
+        listen: Endpoint string(s) (see parse_endpoint for formats).
+        default_port: Port to use when not specified in listen args.
+
+    """
+    if listen is None:
+        listen = [f"localhost:{default_port}"]
+    elif isinstance(listen, str):
+        listen = [listen]
+    return [ep for s in listen for ep in parse_endpoint(s, default_port)]

fastapi-vue/fastapi_vue/logging.py

@@ -0,0 +1,379 @@
+"""Logging integration: tracerite loading and colored access log formatting.
+
+The access log middleware supplies colored fields (``client``, ``status``,
+``method``, ``host``, ``path``, ``extra``, ``timing``) via ``extra=``.  When
+colors are disabled the ANSI escape codes are stripped from the assembled
+output so the same formatting code path produces plain text.
+"""
+
+from __future__ import annotations
+
+import io
+import logging
+import re
+import sys
+from contextlib import suppress
+from copy import deepcopy
+from typing import TYPE_CHECKING, Literal
+
+import tracerite
+from starlette.middleware.errors import ServerErrorMiddleware
+from starlette.responses import HTMLResponse, JSONResponse, PlainTextResponse, Response
+from uvicorn.config import Config
+from uvicorn.lifespan.on import LifespanOn
+
+if TYPE_CHECKING:
+    from typing import Any
+
+    from starlette.requests import Request
+    from uvicorn._types import LifespanScope
+    from uvicorn.lifespan.on import LifespanSendMessage
+
+from .accesslog import AccessLogMiddleware
+
+ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*m")
+
+ACCESS_LOG_FMT = "%(client)s %(status)s %(method)s %(host)s%(path)s %(extra)s%(timing)s"
+
+ACCESS_LOGGER = "fastapi_vue.access"
+
+
+def strip_ansi(text: str) -> str:
+    """Remove ANSI escape codes from text."""
+    return ANSI_ESCAPE_RE.sub("", text)
+
+
+_LEVEL_EMOJI = {
+    logging.DEBUG: "🐛",
+    logging.INFO: "🔷",
+    logging.WARNING: "❗",
+    logging.ERROR: "🛑",
+    logging.CRITICAL: "🚨",
+}
+
+
+def _level_prefix(record: logging.LogRecord) -> str:
+    emoji = _LEVEL_EMOJI.get(record.levelno)
+    return f"{emoji} " if emoji else f"{record.levelname}: "
+
+
+class Formatter(logging.Formatter):
+    """Formatter for both access records and ordinary log messages.
+
+    Records with the middleware's access fields (``client`` etc.) are
+    formatted from those; anything else gets an emoji level prefix
+    (``LEVEL: `` fallback for unknown levels) in place of uvicorn's
+    ``levelprefix``.
+
+    Instantiation always loads tracerite, and with ``access=True`` also
+    installs the access-log middleware: ``dictConfig`` builds formatters while
+    uvicorn applies ``log_config``, which happens before the app is loaded —
+    including in reload/worker subprocesses that re-import the config without
+    calling ``fastapi_vue.server.run()`` again.  Patching the server error
+    middleware here likewise propagates it to those subprocesses.
+    """
+
+    def __init__(
+        self,
+        fmt: str | None = None,
+        datefmt: str | None = None,
+        style: Literal["%", "{", "$"] = "%",
+        use_colors: bool | None = None,  # noqa: FBT001  # mirrors logging.Formatter
+        *,
+        access: bool = False,
+    ) -> None:
+        """Load tracerite, optionally install the access log, detect color support."""
+        tracerite.load()
+        tracerite.load_suppressions(
+            extra={"starlette.routing": "until", "fastapi.routing": "until"}
+        )
+        patch_lifespan_logging()
+        patch_server_error_middleware()
+        if access:
+            install_access_log()
+        if use_colors in (True, False):
+            self.use_colors = use_colors
+        else:
+            self.use_colors = sys.stdout.isatty()
+        super().__init__(fmt=fmt, datefmt=datefmt, style=style)
+
+    def formatMessage(self, record: logging.LogRecord) -> str:  # noqa: N802
+        """Format access records via middleware fields, others with an emoji prefix."""
+        if "client" not in record.__dict__:
+            return _level_prefix(record) + record.getMessage()
+        formatted = super().formatMessage(record)
+        if not self.use_colors:
+            formatted = strip_ansi(formatted)
+        return formatted
+
+
+class WebSocketChatterFilter(logging.Filter):
+    """Drop stock uvicorn WebSocket handshake/chatter records.
+
+    Stock uvicorn logs WS handshakes (``'%s - "WebSocket %s" ...'``) and the
+    websockets library's "connection open/closed" chatter to ``uvicorn.error``,
+    ungated by ``access_log``.  Our middleware logs WebSockets itself.
+    """
+
+    _PREFIXES = ('%s - "WebSocket ', "connection open", "connection closed", "connection rejected")
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        """Keep records not matching stock WebSocket chatter prefixes."""
+        msg = record.msg
+        if not isinstance(msg, str):
+            return True
+        return not msg.startswith(self._PREFIXES)
+
+
+class UvicornQuietFilter(logging.Filter):
+    """Silence uvicorn's routine chatter (startup/shutdown lines, etc.).
+
+    Handler-side, not a logger level: uvicorn's ``configure_logging``
+    re-applies ``log_level`` to its loggers after ``dictConfig``, which would
+    override a level lifted in the config dict.
+    """
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        """Drop uvicorn records below WARNING."""
+        return not (record.name.startswith("uvicorn") and record.levelno < logging.WARNING)
+
+
+_installed = False
+
+
+def install_access_log() -> None:
+    """Wrap apps loaded by uvicorn with AccessLogMiddleware, once per process.
+
+    The guard is deliberately module-level: reload/worker subprocesses
+    re-import this module, resetting it so the patch is re-applied there.
+    """
+    global _installed  # noqa: PLW0603  # deliberately module-level, see docstring
+    if _installed:
+        return
+    _installed = True
+
+    original_load = Config.load
+
+    def load(self):  # noqa: ANN001, ANN202
+        original_load(self)
+        if not isinstance(self.loaded_app, AccessLogMiddleware):
+            self.loaded_app = AccessLogMiddleware(self.loaded_app)
+
+    Config.load = load  # type: ignore[method-assign]
+
+
+_lifespan_patched = False
+
+
+def patch_lifespan_logging() -> None:
+    """Patch uvicorn's LifespanOn to log lifespan failures with exc_info.
+
+    Starlette formats lifespan exceptions into a plain-text ASGI message,
+    which uvicorn logs as-is without exc_info, while the exc_info-carrying
+    log in ``LifespanOn.main()`` is skipped when a failure message was sent.
+    This suppresses the text message and always logs the exception with
+    exc_info, so tracerite (or any exc_info-aware handler) renders the
+    traceback. Monkeypatches uvicorn internals; written against uvicorn 0.52.
+    """
+    global _lifespan_patched  # noqa: PLW0603  # once-per-process, resets in subprocesses
+    if _lifespan_patched:
+        return
+    _lifespan_patched = True
+
+    original_send = LifespanOn.send
+
+    async def send(self: LifespanOn, message: LifespanSendMessage) -> None:
+        # Drop the pre-formatted traceback text; main() logs the exception itself.
+        if message["type"] in ("lifespan.startup.failed", "lifespan.shutdown.failed"):
+            message = dict(message)  # type: ignore[assignment]
+            message.pop("message", None)
+        await original_send(self, message)
+
+    async def main(self: LifespanOn) -> None:
+        """Mirror upstream LifespanOn.main, but always log failures with exc_info."""
+        try:
+            app = self.config.loaded_app
+            scope: LifespanScope = {
+                "type": "lifespan",
+                "asgi": {"version": self.config.asgi_version, "spec_version": "2.0"},
+                "state": self.state,
+            }
+            await app(scope, self.receive, self.send)
+        except BaseException:
+            self.asgi = None
+            self.error_occurred = True
+            if self.startup_failed or self.shutdown_failed or self.config.lifespan != "auto":
+                phase = "shutdown" if self.shutdown_failed else "startup"
+                self.logger.exception("Uncaught exception during application %s", phase)
+            else:
+                self.logger.info("ASGI 'lifespan' protocol appears unsupported.")
+        finally:
+            self.startup_event.set()
+            self.shutdown_event.set()
+
+    LifespanOn.send = send  # type: ignore[method-assign]
+    LifespanOn.main = main  # type: ignore[method-assign]
+
+
+_server_error_patched = False
+
+DEBUG_INGRESS = """This page is shown for your guidance because the application is \
+running in debug mode and has crashed handling this request."""
+
+
+def _generate_html(exc: Exception) -> str:
+    return tracerite.html_page(
+        exc,
+        title="FastAPI debugger",
+        heading="500 Server Error",
+        ingress=DEBUG_INGRESS,
+    )
+
+
+def _generate_plain_text(exc: Exception) -> str:
+    buffer = io.StringIO()
+    tracerite.tty_traceback(exc, file=buffer)
+    return buffer.getvalue()
+
+
+def _generate_json(exc: Exception) -> dict[str, Any]:
+    chain = tracerite.extract_chain(exc)
+    return {"detail": "Internal Server Error", "traceback": chain}
+
+
+def patch_server_error_middleware() -> None:
+    """Patch Starlette's ServerErrorMiddleware to format debug errors with tracerite.
+
+    Starlette's debug responses use its own static HTML traceback template.
+    This replaces ``debug_response`` with tracerite renderers (source
+    context, locals, chained exceptions), adds ``accept: application/json``
+    handling, and returns a JSON body also for non-debug errors when
+    requested. Only apps running with ``debug=True`` produce traceback
+    responses. Monkeypatches Starlette internals; written against
+    starlette 1.6.
+    """
+    global _server_error_patched  # noqa: PLW0603  # once-per-process, resets in subprocesses
+    if _server_error_patched:
+        return
+    _server_error_patched = True
+
+    def debug_response(
+        self: ServerErrorMiddleware,  # noqa: ARG001
+        request: Request,
+        exc: Exception,
+    ) -> Response:
+        accept = request.headers.get("accept", "")
+        if "text/html" in accept:
+            return HTMLResponse(_generate_html(exc), status_code=500)
+        if "application/json" in accept:
+            return JSONResponse(_generate_json(exc), status_code=500)
+        return PlainTextResponse(_generate_plain_text(exc), status_code=500)
+
+    def error_response(
+        self: ServerErrorMiddleware,  # noqa: ARG001
+        request: Request,
+        exc: Exception,  # noqa: ARG001  # signature mirrors Starlette's
+    ) -> Response:
+        if "application/json" in request.headers.get("accept", ""):
+            return JSONResponse({"detail": "Internal Server Error"}, status_code=500)
+        return PlainTextResponse("Internal Server Error", status_code=500)
+
+    ServerErrorMiddleware.debug_response = debug_response  # type: ignore[method-assign]
+    ServerErrorMiddleware.error_response = error_response  # type: ignore[method-assign]
+
+
+def patch_log_config(log_config, *, access_log: bool = True):  # noqa: ANN001, ANN201
+    """Patch a uvicorn log_config dict for our logging, best-effort.
+
+    Users presumably base their config on uvicorn's default dict, but any
+    shape is tolerated: pieces that do not fit the config's structure are
+    silently skipped.  Non-dict configs (e.g. an ini file path) pass through
+    untouched.
+
+    Always adds an unreferenced NullHandler whose Formatter instantiation
+    loads tracerite in every process uvicorn applies the config in, filters
+    on the default handler dropping stock uvicorn's WebSocket chatter and
+    routine INFO lines, an emoji-level-prefix Formatter in place of
+    uvicorn's stock ``default`` formatter (a user-supplied one wins), a root
+    logger entry so ``logging.info()`` et al. print through the default
+    handler, and a no-prefix ``kanta`` logger entry (likewise).
+    With ``access_log``, additionally rewires the ``access`` formatter to
+    our Formatter and attaches its handler to our ``fastapi_vue.access``
+    logger.  We must not
+    attach handlers to ``uvicorn.access``: uvicorn gates its own
+    protocol-level access logging on ``uvicorn.access.hasHandlers()``.
+    """
+    if not isinstance(log_config, dict):
+        return log_config
+    config = deepcopy(log_config)
+
+    with suppress(Exception):
+        config["formatters"]["fastapi_vue"] = {"()": "fastapi_vue.logging.Formatter"}
+        config["handlers"]["fastapi_vue"] = {
+            "class": "logging.NullHandler",
+            "formatter": "fastapi_vue",
+        }
+
+    with suppress(Exception):
+        filters = config.setdefault("filters", {})
+        filters["ws_chatter"] = {"()": "fastapi_vue.logging.WebSocketChatterFilter"}
+        filters["uvicorn_quiet"] = {"()": "fastapi_vue.logging.UvicornQuietFilter"}
+        handler_filters = config["handlers"]["default"].setdefault("filters", [])
+        for name in ("ws_chatter", "uvicorn_quiet"):
+            if name not in handler_filters:
+                handler_filters.append(name)
+
+    # Emoji level prefixes for ordinary logs, replacing uvicorn's stock
+    # default formatter; a user-supplied default formatter is left alone.
+    with suppress(Exception):
+        default = config["formatters"]["default"]
+        if default.get("()") in (None, "uvicorn.logging.DefaultFormatter"):
+            config["formatters"]["default"] = {
+                "()": "fastapi_vue.logging.Formatter",
+                "fmt": "%(message)s",
+                "use_colors": None,
+            }
+
+    # uvicorn's default config leaves the root logger handlerless, eating
+    # logging.info() et al.; route root through uvicorn's default handler.
+    with suppress(Exception):
+        root = config.setdefault("root", {})
+        root.setdefault("level", "INFO")
+        root_handlers = root.setdefault("handlers", [])
+        if "default" not in root_handlers:
+            root_handlers.append("default")
+
+    # kanta-style output (diffs, colored headers) prints without prefixes,
+    # like our access log.  A user-supplied "kanta" logger entry wins.
+    with suppress(Exception):
+        config["formatters"].setdefault("plain", {"fmt": "%(message)s"})
+        config["handlers"].setdefault(
+            "plain",
+            {
+                "class": "logging.StreamHandler",
+                "formatter": "plain",
+                "stream": "ext://sys.stderr",
+            },
+        )
+        config.setdefault("loggers", {}).setdefault(
+            "kanta",
+            {"handlers": ["plain"], "level": "INFO", "propagate": False},
+        )
+
+    if access_log:
+        with suppress(Exception):
+            config["formatters"]["access"] = {
+                "()": "fastapi_vue.logging.Formatter",
+                "fmt": ACCESS_LOG_FMT,
+                "use_colors": None,
+                "access": True,
+            }
+        with suppress(Exception):
+            if "access" in config["handlers"]:
+                config.setdefault("loggers", {})[ACCESS_LOGGER] = {
+                    "handlers": ["access"],
+                    "level": "INFO",
+                    "propagate": False,
+                }
+
+    return config

fastapi-vue/fastapi_vue/server.py

@@ -1,58 +1,143 @@
+"""Uvicorn server runner with multi-endpoint support."""
+
 import asyncio
+import importlib.metadata
 import logging
 import os
 from contextlib import suppress
+from pathlib import Path
+from typing import Any
 
+import tracerite
 import uvicorn
 from uvicorn import Config, Server
 
-from .hostutil import parse_endpoint
+from .hostutil import parse_endpoints
+from .logging import (
+    install_access_log,
+    patch_lifespan_logging,
+    patch_log_config,
+    patch_server_error_middleware,
+)
+from .startupbox import print_box
+
+tracerite.load()  # Early load on CLI load (import server); uvicorn workers reload via log config
 
 logger = logging.getLogger(__name__)
 
+_WILDCARD_HOSTS = frozenset({"0.0.0.0", "::"})  # noqa: S104
 
-def run(
+
+def _connect_url(endpoints: list[dict]) -> str:
+    """Return a URL the user can connect to for the first TCP endpoint.
+
+    Wildcard binds (0.0.0.0, ::) are shown as localhost, as that is the
+    address a user can actually open. Returns "" for unix-socket-only setups.
+    """
+    for endpoint in endpoints:
+        host = endpoint.get("host")
+        if host is None:
+            continue
+        if host in _WILDCARD_HOSTS:
+            host = "localhost"
+        elif ":" in host:  # IPv6 literal
+            host = f"[{host}]"
+        return f"http://{host}:{endpoint['port']}"
+    return ""
+
+
+def _print_startup_box(template: str, app: str, endpoints: list[dict]) -> None:
+    """Format the startup box template and print it.
+
+    Available fields: ``{module}`` (top-level package of the app path),
+    ``{name}`` (module with spaces instead of underscores), ``{Name}``
+    (also capitalized), ``{version}`` (from installed package metadata,
+    "dev" when not installed) and ``{url}``.
+    """
+    module = app.split(":", 1)[0].split(".", 1)[0]
+    name = module.replace("_", " ")
+    try:
+        version = importlib.metadata.version(module)
+    except importlib.metadata.PackageNotFoundError:
+        version = ""
+    values = {
+        "module": module,
+        "name": name,
+        "Name": name.title(),
+        "version": version,
+        "url": _connect_url(endpoints),
+    }
+    print_box(template.format_map(values))
+
+
+def run(  # noqa: PLR0913
     app: str,
     *,
     listen: str | list[str] | None = None,
     default_port: int = 8000,
-    reload: bool = False,
+    reload: bool | Path = False,
     workers: int | None = None,
-    **uvicorn_config,
+    access_log: bool = True,
-):
+    startup_box: str | None = "{Name} {version}\n{url}",
+    log_config: Any = uvicorn.config.LOGGING_CONFIG,  # noqa: ANN401
+    **uvicorn_config: Any,  # noqa: ANN401
+) -> None:
     """Run uvicorn server(s) for the given app.
 
     Args:
         app: The ASGI application path (e.g., "myapp.main:app")
         listen: Endpoint string(s) (see parse_endpoint for formats).
         default_port: Port to use when not specified in listen args.
-        reload: Enable auto-reload (requires uvicorn.run, single endpoint only).
+        reload: Enable auto-reload. If a Path is given, reload watches that
+            directory. True enables reload without setting a reload directory.
+            False disables reload and clears any reload_dirs.
         workers: Number of worker processes (requires uvicorn.run, single endpoint only).
+        access_log: Enable our colored HTTP/WebSocket access logging middleware
+            (uvicorn's own access logging is always bypassed).
+        startup_box: Template for the startup box printed to stderr before
+            serving (see _print_startup_box for fields), None to not print it.
+        log_config: Logging config passed to uvicorn. Dict configs are patched
+            best-effort (see fastapi_vue.logging.patch_log_config): tracerite
+            loading and WebSocket chatter filtering are always installed, and
+            when access_log is enabled the access formatting is rewired too.
         **uvicorn_config: Additional uvicorn config options (overrides all other settings).
-    """
-    if listen is None:
-        listen = [f"localhost:{default_port}"]
-    elif isinstance(listen, str):
-        listen = [listen]
-    endpoints: list[dict] = []
-    for ep in listen:
-        endpoints.extend(parse_endpoint(ep, default_port))
 
-    conf: dict[str, object] = {"app": app, "reload": reload, "workers": workers}
+    """
+    endpoints = parse_endpoints(listen, default_port)
+    if not endpoints:
+        msg = "No endpoints to serve; check listen configuration"
+        raise ValueError(msg)
+
+    if startup_box:
+        _print_startup_box(startup_box, app, endpoints)
+
+    if isinstance(reload, Path):
+        uvicorn_config["reload_dirs"] = [str(reload)]
+    elif not reload:
+        uvicorn_config.pop("reload_dirs", None)
+
+    if access_log:
+        install_access_log()
+    patch_lifespan_logging()
+    patch_server_error_middleware()
+    uvicorn_config["access_log"] = False  # We always bypass uvicorn's own access logging
+    uvicorn_config["log_config"] = patch_log_config(log_config, access_log=access_log)
+
+    conf: dict[str, object] = {"app": app, "reload": bool(reload), "workers": workers}
     proxy = os.getenv("FORWARDED_ALLOW_IPS", "127.0.0.1,::1")
     if proxy:
         conf["proxy_headers"] = True
         conf["forwarded_allow_ips"] = proxy
     conf.update(uvicorn_config)
 
-    with suppress(KeyboardInterrupt):
+    with suppress(KeyboardInterrupt, asyncio.CancelledError):
         if reload or workers:
             serve_multiprocess(endpoints, **conf)
         else:
             asyncio.run(serve(endpoints, **conf))
 
 
-async def serve(endpoints: list[dict], **kwargs) -> None:
+async def serve(endpoints: list[dict], **kwargs: Any) -> None:  # noqa: ANN401
     """Serve the given endpoints in current process/loop. Does not spawn extra processes."""
     forbidden = {"reload", "workers"} & {k for k, v in kwargs.items() if v}
     if forbidden:
@@ -63,13 +148,10 @@ async def serve(endpoints: list[dict], **kwargs) -> None:
     await asyncio.gather(*(Server(Config(**kwargs, **ep)).serve() for ep in endpoints))
 
 
-def serve_multiprocess(endpoints: list[dict], **kwargs) -> None:
+def serve_multiprocess(endpoints: list[dict], **kwargs: Any) -> None:  # noqa: ANN401
     """Serve using uvicorn.run() for reload/workers support. Only first endpoint is used."""
     if len(endpoints) > 1:
-        eps = [
+        eps = [ep["uds"] if "uds" in ep else f"{ep['host']}:{ep['port']}" for ep in endpoints]
-            ep["uds"] if "uds" in ep else f"{ep['host']}:{ep['port']}"
-            for ep in endpoints
-        ]
         logger.warning(
             "Current mode supports only one endpoint. Listening: %s, skipped: %s",
             eps[0],

fastapi-vue/fastapi_vue/startupbox.py

@@ -0,0 +1,20 @@
+"""Startup banner: a rounded unicode box printed to stderr (not via logging)."""
+
+import sys
+
+from .termwidth import display_width, pad_display
+
+
+def print_box(text: str) -> None:
+    """Print text in a rounded unicode box sized to its longest line.
+
+    The text is split on newlines as-is (not trimmed). Padding accounts
+    for ANSI colors and unicode display width (see termwidth.display_width).
+    """
+    lines = text.split("\n")
+    width = max(map(display_width, lines))
+    border = "─" * (width + 2)
+    out = [f"╭{border}╮"]
+    out.extend(f"│ {pad_display(line, width)} │" for line in lines)
+    out.append(f"╰{border}╯")
+    sys.stderr.write("\n".join(out) + "\n")

fastapi-vue/fastapi_vue/staticfiles.py

@@ -25,7 +25,7 @@ __all__ = ["Frontend"]
 
 
 class Assets:
-    """Default cached value to /assets/"""
+    """Default cached value to /assets/."""
 
     @staticmethod
     def parse(cached: str | list[str] | Assets) -> list[str]:
@@ -37,7 +37,8 @@ class Assets:
             case list():
                 return cached
             case _:
-                raise ValueError(f"Invalid cached value: {cached!r}")
+                msg = f"Invalid cached value: {cached!r}"
+                raise ValueError(msg)
 
 
 class Frontend:
@@ -55,27 +56,29 @@ class Frontend:
         index: Name of the index file (default: "index.html")
         spa: Enable SPA mode - serve index.html for unknown routes (default: False)
         cached: Path prefixes that are immutable (default: "/assets/")
-        favicon: May use wildcards of full path. E.g. /assets/logo*.png matches logo.hash.png created by Vite
+        favicon: Wildcard path to favicon. E.g. /assets/logo*.png matches Vite output
         zstdlevel: Zstd compression level (default: 18)
+
     """
 
-    def __init__(
+    def __init__(  # noqa: PLR0913
         self,
         directory: Path | str,
         *,
         index: str = "index.html",
         spa: bool = False,
         catch_all: bool | None = None,
-        cached: str | list[str] | Assets = Assets(),
+        cached: str | list[str] | Assets | None = None,
         favicon: str | None = None,
         zstdlevel: int = 18,
     ) -> None:
+        """Initialize Frontend with given configuration."""
         self.www: dict[str, tuple[bytes, bytes | None, dict]] = {}
         self.base: Path = Path(directory)
         self.index = index
         self.spa = spa
         self._catch_all = spa if catch_all is None else catch_all
-        self.cached_paths = Assets.parse(cached)
+        self.cached_paths = Assets.parse(cached if cached is not None else Assets())
         self.zstdlevel = zstdlevel
         self.favicon = favicon
         self._app: FastAPI | None = None
@@ -107,47 +110,48 @@ class Frontend:
                 paths.add("/favicon.ico")
         return paths
 
-    def _load(self):
+    def _load(self) -> dict[str, tuple[bytes, bytes | None, dict]]:
         """Load static files from disk with compression."""
         www: dict[str, tuple[bytes, bytes | None, dict]] = {}
         if not self.base.exists():
-            raise ValueError(f"Frontend folder {self.base} not found (try uv build)")
+            logger.error(
-        paths = [PurePath()]
+                "Missing %s - no frontend (try uv build)",
-        while paths:
+                self.base,
-            current = self.base / paths.pop(0)
+            )
-            for p in current.iterdir():
+        else:
-                rel = p.relative_to(self.base)
+            paths = [PurePath()]
-                if p.is_dir():
+            while paths:
-                    paths.append(rel)
+                current = self.base / paths.pop(0)
-                    continue
+                for p in current.iterdir():
-                # Read file
+                    rel = p.relative_to(self.base)
-                name = "/" + rel.as_posix()
+                    if p.is_dir():
-                mime = mimetypes.guess_type(name)[0] or "application/octet-stream"
+                        paths.append(rel)
-                name = name.removesuffix(self.index)
+                        continue
-                data = p.read_bytes()
+                    # Read file
-                etag = urlsafe_b64encode(blake3(data).digest(9)).decode()
+                    name = "/" + rel.as_posix()
-                if mime.startswith("text/"):
+                    mime = mimetypes.guess_type(name)[0] or "application/octet-stream"
-                    mime += "; charset=UTF-8"
+                    name = name.removesuffix(self.index)
-                mtime = p.stat().st_mtime
+                    data = p.read_bytes()
-                cached = any(name.startswith(prefix) for prefix in self.cached_paths)
+                    etag = urlsafe_b64encode(blake3(data).digest(9)).decode()
-                headers = {
+                    if mime.startswith("text/"):
-                    "etag": f'"{etag}"',
+                        mime += "; charset=UTF-8"
-                    "last-modified": format_date_time(mtime),
+                    mtime = p.stat().st_mtime
-                    "cache-control": (
+                    cached = any(name.startswith(prefix) for prefix in self.cached_paths)
-                        "max-age=31536000, immutable" if cached else "no-cache"
+                    headers = {
-                    ),
+                        "etag": f'"{etag}"',
-                    "content-type": mime,
+                        "last-modified": format_date_time(mtime),
-                }
+                        "cache-control": ("max-age=31536000, immutable" if cached else "no-cache"),
-                zstd = ZstdCompressor(self.zstdlevel).compress(data)
+                        "content-type": mime,
-                if len(zstd) >= len(data):
+                    }
-                    zstd = None
+                    zstd = ZstdCompressor(self.zstdlevel).compress(data)
-                www[name] = data, zstd, headers
+                    if len(zstd) >= len(data):
-        if self.favicon:
+                        zstd = None
-            if m := fnmatch.filter(www, self.favicon):
+                    www[name] = data, zstd, headers
-                data, zstd, headers = www[m[0]]
+        if self.favicon and (m := fnmatch.filter(www, self.favicon)):
-                if "immutable" in headers.get("cache-control", ""):
+            data, zstd, headers = www[m[0]]
-                    headers = {**headers, "cache-control": "max-age=86400"}
+            if "immutable" in headers.get("cache-control", ""):
-                www["/favicon.ico"] = data, zstd, headers
+                headers = {**headers, "cache-control": "max-age=86400"}
+            www["/favicon.ico"] = data, zstd, headers
         if not www:
             msg = "Frontend files missing, check your installation.\n"
             www["/"] = (
@@ -161,7 +165,7 @@ class Frontend:
             )
         return www
 
-    async def load(self, *, debug: bool | None = None, log: bool = True):
+    async def load(self, *, debug: bool | None = None, log: bool = True) -> None:
         """Load or reload static files from disk.
 
         In debug mode, returns 409 instead of files (avoid accidental use of stale builds)
@@ -187,13 +191,19 @@ class Frontend:
         ratio = comp / raw * 100 if raw else 100.0
         if log and self.www:
             logger.info(
-                f"{self.base.name}: {len(self.www)} files in {1000 * duration:.1f} ms | "
+                "%s: %d files in %.1f ms | zstd %d files %.2f->%.2f MB (%.0f %%)",
-                f"zstd {len(compfiles)} files {1e-6 * raw:.2f}->{1e-6 * comp:.2f} MB ({ratio:.0f} %)"
+                self.base.name,
+                len(self.www),
+                1000 * duration,
+                len(compfiles),
+                1e-6 * raw,
+                1e-6 * comp,
+                ratio,
             )
         if self.favicon and "/favicon.ico" not in self.www:
             logger.warning("Favicon not found: %s", self.favicon)
 
-    def route(self, app: FastAPI, mount_path="/"):
+    def route(self, app: FastAPI, mount_path: str = "/") -> None:
         """Register frontend routes with a FastAPI app.
 
         In SPA/catch-all mode, this must only be called only after all other routes.
@@ -204,6 +214,7 @@ class Frontend:
         Args:
             app: FastAPI application instance
             mount_path: Path where the frontend should be mounted (default: "/")
+
         """
         self._app = app
         self._mount_path = mount_path.rstrip("/")
@@ -212,9 +223,11 @@ class Frontend:
         if self._catch_all:
             # Register catch-all immediately (works without load)
             path = self._mount_path + "{path:path}"
-            app.api_route(path, methods=["GET", "HEAD"], name="frontend")(self.handle)
+            app.api_route(path, methods=["GET", "HEAD"], name="frontend", response_model=None)(
+                self.handle
+            )
 
-    def _register_routes(self):
+    def _register_routes(self) -> None:
         """Register individual routes for each loaded file (non-catch_all mode)."""
         if self._app is None or self._catch_all:
             return
@@ -240,18 +253,19 @@ class Frontend:
             for p in paths
         ]
 
-    def _respond(self, request: Request, name: str):
+    def _respond(self, request: Request, name: str) -> Response:
         """Serve a static file with ETag and compression support."""
         data, zstd, headers = self.www[name]
         if request.headers.get("if-none-match") == headers["etag"]:
             return Response(status_code=304, headers=headers)
         if zstd and "zstd" in request.headers.get("accept-encoding", ""):
             return Response(
-                content=zstd, headers={**headers, "content-encoding": "zstd"}
+                content=zstd,
+                headers={**headers, "content-encoding": "zstd"},
             )
         return Response(content=data, headers=headers)
 
-    def handle(self, request: Request, path: str):
+    def handle(self, request: Request, path: str) -> Response | RedirectResponse:
         """SPA catch-all handler with directory redirects and fallback to index."""
         name = path.removesuffix(self.index)
         debug = getattr(self._app, "debug", False)
@@ -271,11 +285,9 @@ class Frontend:
         return (_devmode_respond if debug else self._respond)(request, name)
 
 
-def _devmode_respond(request: Request, name=""):
+def _devmode_respond(_request: Request, _name: str = "") -> JSONResponse:
     """Return error response directing to Vite server."""
     return JSONResponse(
         status_code=409,
-        content={
+        content={"detail": "[devmode] Use Vite devserver instead."},
-            "detail": "Frontend assets served by Vite in debug mode. You are on backend, connect to frontend instead."
-        },
     )

fastapi-vue/fastapi_vue/termwidth.py

@@ -0,0 +1,37 @@
+"""Terminal display width calculation for unicode and ANSI-colored text."""
+
+import re
+import unicodedata
+
+ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;:]*[A-Za-z]")
+
+
+def _is_wide(char: str) -> bool:
+    """Return True for characters rendered as two terminal columns."""
+    if unicodedata.east_asian_width(char) in {"F", "W"}:
+        return True
+    cp = ord(char)
+    return unicodedata.category(char) == "So" and (
+        0x2600 <= cp <= 0x27BF or 0x1F300 <= cp <= 0x1F9FF or 0x1FA00 <= cp <= 0x1FAFF
+    )
+
+
+def display_width(text: str) -> int:
+    """Calculate the display width of a string in terminal columns.
+
+    ANSI escape codes are ignored. Wide characters (East Asian F/W and
+    emoji) count as two columns, combining marks and format characters
+    (e.g. emoji variation selectors) as zero.
+    """
+    plain = ANSI_ESCAPE_RE.sub("", text)
+    width = 0
+    for char in plain:
+        if unicodedata.category(char) in {"Mn", "Mc", "Me", "Cf"}:
+            continue
+        width += 2 if _is_wide(char) else 1
+    return width
+
+
+def pad_display(text: str, width: int) -> str:
+    """Pad text with trailing spaces to the given display width (no truncation)."""
+    return text + " " * max(width - display_width(text), 0)

fastapi-vue/pyproject.toml

@@ -8,11 +8,13 @@ dependencies = [
     "fastapi>=0.115.0",
     "zstandard>=0.23.0",
     "blake3>=1.0.8",
+    "tracerite>=2.6.5",
 ]
 
 [project.urls]
-Homepage = "https://git.zi.fi/LeoVasanko/fastapi-vue"
+Homepage = "https://vasanko.com/coders/fastapi-vue"
-Repository = "https://github.com/LeoVasanko/fastapi-vue"
+Repository = "https://git.zi.fi/LeoVasanko/fastapi-vue-setup"
+Issues = "https://github.com/LeoVasanko/fastapi-vue-setup"
 
 [build-system]
 requires = ["hatchling", "hatch-vcs"]

pyproject.toml

@@ -14,8 +14,9 @@ dependencies = [
 ]
 
 [project.urls]
-Homepage = "https://git.zi.fi/LeoVasanko/fastapi-vue-setup"
+Homepage = "https://vasanko.com/coders/fastapi-vue"
-Repository = "https://github.com/LeoVasanko/fastapi-vue-setup"
+Repository = "https://git.zi.fi/LeoVasanko/fastapi-vue-setup"
+Issues = "https://github.com/LeoVasanko/fastapi-vue-setup"
 
 [project.scripts]
 fastapi-vue-setup = "fastapi_vue_setup:main"
@@ -31,3 +32,20 @@ dev = ["ruff", "fastapi-vue"]
 
 [tool.uv.sources]
 fastapi-vue = { path = "fastapi-vue", editable = true }
+
+[tool.uv.workspace]
+members = [
+    "f",
+]
+
+[tool.ruff]
+line-length = 100
+
+[tool.ruff.lint]
+select = ["ALL"]
+ignore = ["CPY", "D203", "D213", "COM812", "PLR2004"]
+
+[tool.ruff.lint.per-file-ignores]
+"template/**" = ["F821"]  # Undefined names are template placeholders
+"template/scripts/devserver.py" = ["N806"]  # MODULE_NAME is a template variable
+"fastapi_vue_setup.py" = ["PLR", "C901", "T201", "RUF001"]

scripts/release.py

@@ -10,25 +10,18 @@ DIST = ROOT / "dist"
 FASTAPI_VUE = ROOT / "fastapi-vue"
 
 
-def main():
+def main() -> None:
+    """Build both packages to dist directory."""
     # Clear the dist directory
     if DIST.exists():
         shutil.rmtree(DIST)
     DIST.mkdir()
 
     # Build fastapi-vue (subdirectory) to root dist
-    subprocess.run(
+    subprocess.run(["uv", "build", "--out-dir", str(DIST)], cwd=FASTAPI_VUE, check=True)  # noqa: S603, S607
-        ["uv", "build", "--out-dir", str(DIST)],
-        cwd=FASTAPI_VUE,
-        check=True,
-    )
 
     # Build fastapi-vue-setup (root)
-    subprocess.run(
+    subprocess.run(["uv", "build", "--out-dir", str(DIST)], cwd=ROOT, check=True)  # noqa: S603, S607
-        ["uv", "build", "--out-dir", str(DIST)],
-        cwd=ROOT,
-        check=True,
-    )
 
 
 if __name__ == "__main__":

template/backend/__init__.py

@@ -0,0 +1 @@
+"""Backend package with FastAPI application and Vue frontend integration."""

template/backend/__main__.py

@@ -1,14 +1,18 @@
 # auto-upgrade@fastapi-vue-setup - remove this if you modify this file
+"""Command-line entry point for running the backend server."""
+
 import argparse
 import os
+from pathlib import Path
 
 from fastapi_vue import server
 
 DEFAULT_PORT = TEMPLATE_DEFAULT_PORT
-DEVMODE = bool(os.getenv("ENVPREFIX_DEV") == "1")
+DEVMODE = os.getenv("ENVPREFIX_DEV") == "1"
 
 
-def main():
+def main() -> None:
+    """Run the backend server with optional arguments."""
     parser = argparse.ArgumentParser(description="Run the MODULE_NAME server.")
     parser.add_argument(
         "-l",
@@ -17,12 +21,12 @@ def main():
         help=(f"Endpoint (default: localhost:{DEFAULT_PORT})."),
     )
     args = parser.parse_args()
-    dev = {"reload": True, "reload_dirs": ["MODULE_NAME"]}
     server.run(
-        "MODULE_NAME.APP_MODULE:APP_VAR",
+        "APP_MODULE:APP_VAR",
         listen=args.listen,
         default_port=DEFAULT_PORT,
-        **(dev if DEVMODE else {}),
+        server_header=False,
+        reload=Path(__file__).parent if DEVMODE else False,
     )
 
 

template/backend/app.py

@@ -1,16 +1,19 @@
+"""FastAPI application module with Vue frontend integration."""
+
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 from pathlib import Path
 
 from fastapi import FastAPI
 from fastapi_vue import Frontend
-from MODULE_NAME.__main__ import DEVMODE
+from MAIN_MODULE import DEVMODE
 
 # Vue Frontend static files
 frontend = Frontend(Path(__file__).with_name("frontend-build"))
 
 
 @asynccontextmanager
-async def lifespan(app: FastAPI):
+async def lifespan(_app: FastAPI) -> AsyncGenerator:
     """Manage app startup and shutdown resources."""
     await frontend.load()
     yield
@@ -24,7 +27,8 @@ app = FastAPI(title="PROJECT_TITLE", debug=DEVMODE, lifespan=lifespan)
 
 # Health check endpoint for the Vue demo app to verify the backend is running
 @app.get("/api/health")
-async def health_check():
+async def health_check() -> dict:
+    """Return backend status for health monitoring."""
     return {"status": "ok"}
 
 

template/frontend/vite-plugin-fastapi.js

@@ -5,13 +5,14 @@
  * Configures Vite for FastAPI backend integration:
  * - Proxies /api/* requests to the FastAPI backend
  * - Builds to the Python module's frontend-build directory
+ * - Disables Vite's screen clearing on startup
  *
  * Options:
  *   paths - Array of paths to proxy (default: ["/api"])
  */
 
 export default function fastapiVue({ paths = ["/api"] } = {}) {
-  const backendUrl = process.env.ENVPREFIX_BACKEND_URL || "http://localhost:TEMPLATE_DEFAULT_PORT"
+  const backendUrl = process.env.ENVPREFIX_BACKEND_URL || "http://localhost:TEMPLATE_DEV_PORT"
 
   // Build proxy configuration for each path
   const proxy = {}
@@ -26,6 +27,7 @@ export default function fastapiVue({ paths = ["/api"] } = {}) {
   return {
     name: "vite-plugin-fastapi-MODULE_NAME",
     config: () => ({
+      clearScreen: false,
       server: { proxy },
       build: {
         outDir: "../MODULE_NAME/frontend-build",

template/scripts/devserver.py

@@ -9,9 +9,11 @@ import sys
 from contextlib import suppress
 from pathlib import Path
 
+import tracerite
+
 # Import util.py from scripts/fastapi-vue (not a package, so we adjust sys.path)
 sys.path.insert(0, str(Path(__file__).with_name("fastapi-vue")))
-from devutil import (  # type: ignore
+from devutil import (
     ProcessGroup,
     check_ports_free,
     logger,
@@ -22,11 +24,15 @@ from devutil import (  # type: ignore
 
 DEFAULT_VITE_PORT = TEMPLATE_VITE_PORT
 DEFAULT_DEV_PORT = TEMPLATE_DEV_PORT
+HEALTH = TEMPLATE_HEALTH
 
 
 async def run_devserver(
-    listen: str, backend: str, extra_args: list[str] | None = None
+    listen: str,
+    backend: str,
+    extra_args: list[str] | None = None,
 ) -> None:
+    """Start Vite and FastAPI dev servers with hot reload."""
     reporoot = Path(__file__).parent.parent
     front = reporoot / "frontend"
     if not (front / "package.json").exists():
@@ -36,7 +42,7 @@ async def run_devserver(
     viteurl, npm_install, vite = setup_vite(listen, DEFAULT_VITE_PORT)
     backurl, MODULE_NAME = setup_cli("PROJECT_CLI", backend, DEFAULT_DEV_PORT)
 
-    # Tell the everyone by environment (vite proxy and backend devmode use these)
+    # Tell everyone via environment (vite proxy and backend devmode use these)
     os.environ["ENVPREFIX_VITE_URL"] = viteurl
     os.environ["ENVPREFIX_BACKEND_URL"] = backurl
     os.environ["ENVPREFIX_DEV"] = "1"
@@ -45,24 +51,27 @@ async def run_devserver(
         npm_i = await pg.spawn(*npm_install, cwd=front)
         await check_ports_free(viteurl, backurl)
         await pg.spawn(*MODULE_NAME, *(extra_args or []))
-        await pg.wait(npm_i, ready(backurl, path="/api/health?from=devserver.py"))
+        await pg.wait(npm_i, ready(backurl, path=HEALTH))
         await pg.spawn(*vite, cwd=front)
 
 
-def main():
+def main() -> None:
+    """Parse CLI arguments and run the devserver."""
+    tracerite.load()
     parser = argparse.ArgumentParser(
         description="Run Vite and FastAPI development servers",
         formatter_class=argparse.RawDescriptionHelpFormatter,
         epilog=HELP_EPILOG,
     )
     parser.add_argument(
-        "-l", "--listen",
+        "-l",
-        metavar="host:port",
+        "--listen",
+        metavar="addr",
         help=f"Vite (default: localhost:{DEFAULT_VITE_PORT})",
     )
     parser.add_argument(
         "--backend",
-        metavar="host:port",
+        metavar="addr",
         help=f"FastAPI (default: localhost:{DEFAULT_DEV_PORT})",
     )
     args, extra_args = parser.parse_known_args()
@@ -71,7 +80,7 @@ def main():
 
 
 HELP_EPILOG = """
-  scripts/devserver.py [args to PROJECT_CLI]
+  Other options are forwarded to PROJECT_CLI [args]
 
   JS_RUNTIME environment variable can be used to select the JS runtime:
   npm, deno, bun, or full path to the runtime executable (node maps to npm).

template/scripts/fastapi-vue/buildhook.py

@@ -1,15 +1,19 @@
+# ruff: noqa: INP001
 """Hatch build hook for building Vue frontend during package build."""
 
 import sys
 from pathlib import Path
 
-from hatchling.builders.hooks.plugin.interface import BuildHookInterface  # type: ignore
+from hatchling.builders.hooks.plugin.interface import BuildHookInterface
 
 sys.path.insert(0, str(Path(__file__).parent))
 from buildutil import build
 
 
-class CustomBuildHook(BuildHookInterface):
+class CustomBuildHook(BuildHookInterface):  # type: ignore[misc]
-    def initialize(self, version, build_data):
+    """Hatch build hook that builds Vue frontend during package build."""
+
+    def initialize(self, version: str, build_data: dict) -> None:  # type: ignore[override]
+        """Build frontend before package is built."""
         super().initialize(version, build_data)
         build("frontend")

template/scripts/fastapi-vue/buildutil.py

@@ -1,3 +1,4 @@
+# ruff: noqa: INP001
 """Utilities used at build time and in devserver script. No dependencies."""
 
 import logging
@@ -7,6 +8,8 @@ import shutil
 import subprocess
 from pathlib import Path
 
+MIN_NODE_VERSION = 20
+
 
 class _PrefixFormatter(logging.Formatter):
     """Formatter that adds prefix based on log level."""
@@ -30,82 +33,119 @@ def _check_node_version(node_path: str) -> None:
     Raises RuntimeError if version is too old or cannot be determined.
     """
     try:
-        result = subprocess.run(
+        result = subprocess.run(  # noqa: S603
-            [node_path, "--version"], capture_output=True, text=True, check=True
+            [node_path, "--version"],
+            capture_output=True,
+            text=True,
+            check=True,
         )
         version_str = result.stdout.strip()
         # Parse version like "v20.10.0" or "v18.17.1"
         match = re.match(r"v(\d+)", version_str)
         if match:
             major_version = int(match.group(1))
-            if major_version >= 20:
+            if major_version >= MIN_NODE_VERSION:
                 return
-            raise RuntimeError(
+            msg = f"Node.js {version_str} found, but v{MIN_NODE_VERSION}+ required"
-                f"Node.js {version_str} found, but v20+ required (install with nvm)"
+            raise RuntimeError(msg)
-            )
     except (subprocess.CalledProcessError, FileNotFoundError, ValueError):
         pass
-    raise RuntimeError("Could not determine Node.js version")
+    msg = "Could not determine Node.js version"
+    raise RuntimeError(msg)
+
+
+def _validate_npm_runtime(tool: str) -> bool:
+    """Validate npm runtime by checking Node.js version. Returns True if valid."""
+    node_path = shutil.which("node", path=str(Path(tool).parent))
+    if node_path is None:
+        return False
+    try:
+        _check_node_version(node_path)
+    except RuntimeError:
+        return False
+    return True
+
+
+def _find_runtime_from_env(options: list[str]) -> tuple[str, str] | None:
+    """Find runtime specified by JS_RUNTIME environment variable."""
+    js_runtime_env = os.environ.get("JS_RUNTIME")
+    if not js_runtime_env:
+        return None
+
+    js_runtime = js_runtime_env
+    js_path = Path(js_runtime)
+    runtime_name = js_path.name
+
+    # Map node to npm
+    if runtime_name == "node":
+        runtime_name = "npm"
+        js_runtime = str(js_path.parent / "npm") if js_path.parent.name else "npm"
+
+    for option in options:
+        if option != runtime_name and not runtime_name.startswith(option):
+            continue
+
+        tool = shutil.which(js_runtime)
+        if tool is None:
+            msg = f"JS_RUNTIME={js_runtime_env}: {option} not found"
+            raise RuntimeError(msg)
+
+        if option == "npm":
+            node_path = shutil.which("node", path=str(Path(tool).parent))
+            if node_path is None:
+                msg = f"JS_RUNTIME={js_runtime_env}: node not found"
+                raise RuntimeError(msg)
+            _check_node_version(node_path)
+
+        return tool, option
+
+    msg = f"JS_RUNTIME={js_runtime_env} not recognized"
+    raise RuntimeError(msg)
+
+
+def _auto_detect_runtime(options: list[str]) -> tuple[str, str]:
+    """Auto-detect JavaScript runtime from available options."""
+    node_version_error: RuntimeError | None = None
+
+    for option in options:
+        tool = shutil.which(option)
+        if not tool:
+            continue
+
+        if option == "npm" and not _validate_npm_runtime(tool):
+            try:
+                node_path = shutil.which("node", path=str(Path(tool).parent))
+                if node_path:
+                    _check_node_version(node_path)
+            except RuntimeError as e:
+                node_version_error = e
+            continue
+
+        return tool, option
+
+    if node_version_error:
+        raise node_version_error
+    msg = "Node.js (v20+), Deno or Bun is required but none was found"
+    raise RuntimeError(msg)
 
 
 def find_js_runtime() -> tuple[str, str]:
     """Find a JavaScript runtime from JS_RUNTIME env or auto-detect.
 
     Returns (tool_path, tool_name) where tool_name is "deno", "npm", or "bun".
-    Raises JSRuntimeError if no suitable runtime is found.
+    Raises RuntimeError if no suitable runtime is found.
     """
     options = ["npm", "deno", "bun"]
-    node_version_error: RuntimeError | None = None
 
     # Check for JS_RUNTIME environment variable
-    if js_runtime_env := os.environ.get("JS_RUNTIME"):
+    if result := _find_runtime_from_env(options):
-        js_runtime = js_runtime_env
+        return result
-        js_path = Path(js_runtime)
-        runtime_name = js_path.name
-        # Map node to npm
-        if runtime_name == "node":
-            runtime_name = "npm"
-            js_runtime = str(js_path.parent / "npm") if js_path.parent.name else "npm"
-        for option in options:
-            if option == runtime_name or runtime_name.startswith(option):
-                tool = shutil.which(js_runtime)
-                if tool is None:
-                    raise RuntimeError(
-                        f"JS_RUNTIME={js_runtime_env}: {option} not found"
-                    )
-                # Check Node.js version if using npm
-                if option == "npm":
-                    node_path = shutil.which("node", path=str(Path(tool).parent))
-                    if node_path is None:
-                        raise RuntimeError(
-                            f"JS_RUNTIME={js_runtime_env}: node not found"
-                        )
-                    _check_node_version(node_path)  # Raises on failure
-                return tool, option
-        raise RuntimeError(f"JS_RUNTIME={js_runtime_env} not recognized")
 
     # Auto-detect
-    for option in options:
+    return _auto_detect_runtime(options)
-        if tool := shutil.which(option):
-            # Check Node.js version if using npm
-            if option == "npm":
-                node_path = shutil.which("node", path=str(Path(tool).parent))
-                if node_path is None:
-                    continue
-                try:
-                    _check_node_version(node_path)
-                except RuntimeError as e:
-                    node_version_error = e
-                    continue  # Try next runtime
-            return tool, option
-
-    # No runtime found - provide helpful error
-    if node_version_error:
-        raise node_version_error
-    raise RuntimeError("Node.js (v20+), Deno or Bun is required but none was found")
 
 
-def find_build_tool():
+def find_build_tool() -> tuple[list[str], list[str]]:
     """Find JavaScript runtime and construct install/build commands.
 
     Returns (install_cmd, build_cmd) tuples of command lists.
@@ -143,7 +183,7 @@ def find_dev_tool() -> list[str]:
 
     if name == "bun":
         logger.warning(
-            "Bun has a bug in WS proxying (https://github.com/oven-sh/bun/issues/9882). Consider using npm instead."
+            "Bun has a WS proxy bug (github.com/oven-sh/bun/issues/9882). Consider npm.",
         )
 
     return [tool, *dev_args[name]]
@@ -176,16 +216,16 @@ def build(folder: str = "frontend") -> None:
         install_cmd, build_cmd = find_build_tool()
     except RuntimeError as e:
         logger.warning(e)
-        raise SystemExit(1)
+        raise SystemExit(1) from None
 
-    def run(cmd):
+    def run(cmd: list[str]) -> None:
         display_cmd = [Path(cmd[0]).stem, *cmd[1:]]
         logger.info("### %s", " ".join(display_cmd))
-        subprocess.run(cmd, check=True, cwd=folder)
+        subprocess.run(cmd, check=True, cwd=folder)  # noqa: S603
 
     try:
         run(install_cmd)
         logger.info("")
         run(build_cmd)
     except subprocess.CalledProcessError:
-        raise SystemExit(1)
+        raise SystemExit(1) from None

template/scripts/fastapi-vue/devutil.py

@@ -1,27 +1,33 @@
+# ruff: noqa: INP001
 """Utilities meant for devserver script, used only in source repository with dev deps."""
 
 import asyncio
 import subprocess
 import sys
-from collections.abc import Coroutine
 from contextlib import suppress
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any, Self
+from urllib.parse import urlsplit
 
-import httpx
 from buildutil import find_dev_tool, find_install_tool, logger
 from fastapi_vue.hostutil import parse_endpoint
 
+if TYPE_CHECKING:
+    from collections.abc import Coroutine
+
 
 class ProcessGroup:
     """Manage async subprocesses with automatic cleanup, like TaskGroup for processes."""
 
-    def __init__(self):
+    def __init__(self) -> None:
+        """Initialize empty process tracking."""
         self._procs: list[asyncio.subprocess.Process] = []
         self._cmds: dict[int, str] = {}  # pid -> command name
 
     async def spawn(
-        self, *cmd: str, cwd: str | None = None
+        self,
+        *cmd: str,
+        cwd: str | None = None,
     ) -> asyncio.subprocess.Process:
         """Spawn a subprocess and track it."""
         cmd_name = Path(cmd[0]).stem
@@ -32,7 +38,8 @@ class ProcessGroup:
         return proc
 
     async def wait(
-        self, *waitables: "asyncio.subprocess.Process | Coroutine[Any, Any, Any]"
+        self,
+        *waitables: "asyncio.subprocess.Process | Coroutine[Any, Any, Any]",
     ) -> None:
         """Wait for processes/coroutines to complete, raise SystemExit on failure."""
 
@@ -43,8 +50,7 @@ class ProcessGroup:
                 raise subprocess.CalledProcessError(returncode, cmd_name)
 
         tasks = [
-            wait_proc(w) if isinstance(w, asyncio.subprocess.Process) else w
+            wait_proc(w) if isinstance(w, asyncio.subprocess.Process) else w for w in waitables
-            for w in waitables
         ]
         try:
             await asyncio.gather(*tasks)
@@ -52,14 +58,15 @@ class ProcessGroup:
             logger.warning("%s failed with exit status %d", e.cmd, e.returncode)
             raise SystemExit(1) from None
 
-    async def __aenter__(self):
+    async def __aenter__(self) -> Self:
+        """Enter the async context manager."""
         return self
 
-    async def __aexit__(self, exc_type, *_):
+    async def __aexit__(self, exc_type: type[BaseException] | None, *_: object) -> None:
         """Wait for one process to exit, terminate others, then wait for all."""
         await self._cleanup(immediate=exc_type is not None)
 
-    async def _cleanup(self, immediate: bool = False):
+    async def _cleanup(self, *, immediate: bool = False) -> None:
         running = [p for p in self._procs if p.returncode is None]
         if not running:
             return
@@ -87,7 +94,7 @@ class ProcessGroup:
                         asyncio.wait_for(
                             asyncio.gather(*[p.wait() for p in still_running]),
                             timeout=10,
-                        )
+                        ),
                     )
                 except TimeoutError:
                     for p in self._procs:
@@ -97,43 +104,69 @@ class ProcessGroup:
                             await p.wait()
 
 
+async def http_get_server(url: str, timeout: float) -> str | None:  # noqa: ASYNC109
+    """GET url with plain asyncio streams, return the response Server header.
+
+    Returns an empty string when the server responds without a Server header,
+    and None when the server is unreachable or doesn't answer in time.
+    """
+    parts = urlsplit(url)
+    host = parts.hostname or "localhost"
+    port = parts.port or (443 if parts.scheme == "https" else 80)
+    path = parts.path or "/"
+    if parts.query:
+        path += f"?{parts.query}"
+    try:
+        async with asyncio.timeout(timeout):
+            reader, writer = await asyncio.open_connection(host, port)
+            try:
+                writer.write(f"GET {path} HTTP/1.0\r\nHost: {host}\r\n\r\n".encode())
+                await writer.drain()
+                data = await reader.readuntil(b"\r\n\r\n")
+            finally:
+                writer.close()
+    except (OSError, EOFError, ValueError, TimeoutError):
+        return None
+    for line in data.decode("latin-1").split("\r\n"):
+        if line.lower().startswith("server:"):
+            return line.split(":", 1)[1].strip()
+    return ""
+
+
 async def check_ports_free(*urls: str) -> None:
     """Verify URLs are not responding (ports are free). Raise SystemExit if any respond."""
 
-    async def check(client: httpx.AsyncClient, url: str) -> None:
+    async def check(url: str) -> None:
-        with suppress(httpx.RequestError):
+        server = await http_get_server(url, timeout=0.1)
-            res = await client.get(url, timeout=0.1)
+        if server is not None:
-            server = res.headers.get("server", "server")
+            logger.warning("Conflicting %s already running at %s", server or "server", url)
-            logger.warning("Conflicting %s already running at %s", server, url)
             raise SystemExit(1)
 
-    async with httpx.AsyncClient() as client:
+    await asyncio.gather(*[check(url) for url in urls])
-        await asyncio.gather(*[check(client, url) for url in urls])
 
 
-async def ready(url: str, path: str = "") -> None:
+async def ready(url: str, path: str = "", max_attempts: int = 50) -> None:
     """Wait for the server to be ready by polling an endpoint.
 
+    Use empty path to disable the check and make this return immediately.
     Raises SystemExit(1) if server doesn't start in time.
     """
-    max_attempts = 50
+    if not path:
-    full_url = f"{url}{path}"
+        return
 
-    async with httpx.AsyncClient() as client:
+    for attempt in range(max_attempts):
-        for attempt in range(max_attempts):
+        if await http_get_server(f"{url}{path}", timeout=1.0) is not None:
-            try:
+            logger.info("✓ Backend ready!")
-                await client.get(full_url, timeout=1.0)
+            return
-                logger.info("✓ Backend ready!")
+        if attempt == max_attempts - 1:
-                return
+            logger.warning("Backend didn't start in time")
-            except httpx.RequestError:
+            raise SystemExit(1)
-                if attempt == max_attempts - 1:
+        await asyncio.sleep(0.1)
-                    logger.warning("Backend didn't start in time")
-                    raise SystemExit(1)
-                await asyncio.sleep(0.1)
 
 
 def setup_vite(
-    endpoint: str, default_port: int = 5173
+    endpoint: str,
+    default_port: int = 5173,
 ) -> tuple[str, list[str], list[str]]:
     """Parse frontend endpoint and build commands.
 
@@ -159,7 +192,9 @@ def setup_vite(
 
 
 def setup_fastapi(
-    endpoint: str, module: str, default_port: int = 8000
+    endpoint: str,
+    module: str,
+    default_port: int = 8000,
 ) -> tuple[str, list[str]]:
     """Parse backend endpoint and build uvicorn command.
 
@@ -174,7 +209,7 @@ def setup_fastapi(
 
     host = endpoints[0]["host"]
     port = endpoints[0]["port"]
-    reload_dir = module.split(".")[0]  # Don't reload on frontend changes
+    reload_dir = module.split(".", maxsplit=1)[0]  # Don't reload on frontend changes
 
     cmd = [
         sys.executable,
@@ -191,7 +226,9 @@ def setup_fastapi(
 
 
 def setup_cli(
-    cli: str, endpoint: str, default_port: int = 8000
+    cli: str,
+    endpoint: str,
+    default_port: int = 8000,
 ) -> tuple[str, list[str]]:
     """Parse backend endpoint and build CLI command.
 
@@ -207,5 +244,7 @@ def setup_cli(
     host = endpoints[0]["host"]
     port = endpoints[0]["port"]
 
-    cmd = [cli, f"--listen={host}:{port}"]
+    # Run the package as a module with the current interpreter, instead of
+    # relying on a PATH-installed CLI entry point.
+    cmd = [sys.executable, "-m", cli, f"--listen={host}:{port}"]
     return f"http://{host}:{port}", cmd