refactor(server): extract the diagnostics routes into routers/diagnostics.py (R3)

The three /api/diagnostics/* routes (export, preview, hardware) plus their
exclusive payload-cap helpers and the `_diag_*` normalisers. Bodies verbatim
except @app->@router, CONFIG_DIR->appstate.config_dir, _running_version()->
appstate.running_version() (a new seam slot; the impl stays in server.py where
the settings region also calls it), and the builtin-plugins lookup in
_diag_plugins_roots: Path(__file__).parent -> Path(__file__).resolve().parents[2]
(routers -> lib -> app root; plugins/ ships at the app root in every packaging
path).

The pure caps/normalisers (_diag_cap_console/_dict/_contributions,
_diag_coerce_bool, _diag_normalize_include, _DIAG_MAX_*) are re-exported from
server.py so the existing `server._diag_*` / `server._DIAG_*` tests keep
resolving — none of them monkeypatch these, so no test retargets.

server.py: 7,216 -> 6,960.

Verified: pyflakes clean (bar the intentional re-export lines); route table
IDENTICAL (143); full pytest 2399 passed (77 diag/packaging + 122 diagnostic-
matched cases incl the cap/coerce/normalize suites); eslint 0; Codex pending.
Boot smoke: /hardware, /preview, and POST /export (200 application/zip) all
serve from the new router location.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

CHANGELOG.md

@@ -27,7 +27,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 - **Perf harness now measures 2D-highway frame time (R3c gate).** `scripts/perf-baseline.mjs` gains a `--song` mode that reports per-frame draw-cost p50/p95/p99 (draw-tagged via `highway.addDrawHook`), the metric that gates the `highway.js` split. Maintainer/CI-only; baseline recorded in `docs/perf-baseline.md`.
-- **`routers/` — extracting `server.py`'s route layer, cheapest-first (R3).** Each PR moves a cohesive route group into a `fastapi.APIRouter` under `lib/routers/`, mounted with `app.include_router(...)` at its original site (FastAPI matches in registration order; the full route table stays byte-identical). Bodies are verbatim — only the decorator receiver (`@app` → `@router`) and singleton reads (`meta_db` → `appstate.meta_db`, resolved at call time) change. So far: `audio_effects` (5), `artist_aliases` (5), `loops` (3), `playlists` (12 + covers), `ws_highway` (the 902-line highway chart WebSocket), `chart` (split/unsplit/work/fileinfo — unblocked by the DLC-path substrate). The DLC library-path resolution (`_get_dlc_dir`, pure `_resolve_dlc_path`) moved to `lib/dlc_paths.py`, reading paths through the seam; `config_dir`/`dlc_dir`/`dlc_dir_env` now ride the `appstate` seam (env-derived, so the pop-and-reimport fixtures reconfigure it for free), and the shared request-field sanitizer `_clean_str` moved to `lib/reqfields.py`. The next cut is picked by a dependency-closure scan that ranks groups by how many `monkeypatch.setattr(server, …)` targets they'd drag along.
+- **`routers/` — extracting `server.py`'s route layer, cheapest-first (R3).** Each PR moves a cohesive route group into a `fastapi.APIRouter` under `lib/routers/`, mounted with `app.include_router(...)` at its original site (FastAPI matches in registration order; the full route table stays byte-identical). Bodies are verbatim — only the decorator receiver (`@app` → `@router`) and singleton reads (`meta_db` → `appstate.meta_db`, resolved at call time) change. So far: `audio_effects` (5), `artist_aliases` (5), `loops` (3), `playlists` (12 + covers), `ws_highway` (the 902-line highway chart WebSocket), `chart` (split/unsplit/work/fileinfo — unblocked by the DLC-path substrate), `library_extras`, `wanted`, `shop`, `progression`, `profile`, `stats` (the `/api/stats/{path}` catch-all stays registered last so it can't shadow `/recent` `/best` `/top`), `version` (`/api/version`; VERSION-file lookup adjusted for the router subdir depth), and `diagnostics` (`/api/diagnostics/export|preview|hardware`; the plugins-root lookup adjusted for the router subdir depth, `_running_version` reached through the `appstate` seam, pure payload-cap helpers re-exported for the `server._diag_*` tests). The DLC library-path resolution (`_get_dlc_dir`, pure `_resolve_dlc_path`) moved to `lib/dlc_paths.py`, reading paths through the seam; `config_dir`/`dlc_dir`/`dlc_dir_env` now ride the `appstate` seam (env-derived, so the pop-and-reimport fixtures reconfigure it for free), and the shared request-field sanitizer `_clean_str` moved to `lib/reqfields.py`. The next cut is picked by a dependency-closure scan that ranks groups by how many `monkeypatch.setattr(server, …)` targets they'd drag along.
 - **`routers/` — the first extracted route module (R3).** The five audio-effects mapping
   endpoints move out of `server.py` into `lib/routers/audio_effects.py` as a
   `fastapi.APIRouter`, mounted with `app.include_router(...)` **at the point in the file

docs/size-exemptions.md

@@ -55,8 +55,8 @@ without a *signed* exemption" is unenforceable.
 ## Planned, NOT exempt (owned by split plans — listed so nothing falls between states)
 
 core `static/app.js` (11,852) · `static/highway.js` (4,168, whole file) · `server.py`
-(7,798 — was 14,037; ratcheted by the R3 `MetadataDB` + `AudioEffectsMappingDB`
-extractions and nine `routers/` modules) ·
+(6,960 — was 14,037; ratcheted by the R3 `MetadataDB` + `AudioEffectsMappingDB`
+extractions and fourteen `routers/` modules) ·
 `lib/metadata_db.py` (4,373 — new in R3; the `MetadataDB` class alone is 4,018 lines
 and is a monolith in its own right, to be split per-table once the router train
 lands) · `static/v3/songs.js` (4,134) · `static/capabilities/audio-session.js`

lib/appstate.py

@@ -86,12 +86,15 @@ audio_cache_dir = None
 # through the seam. get_progression_content wraps a lazy content cache that stays
 # in server.py (its `setattr(server, "_progression_content")` test is untouched).
 get_progression_content = None
+builtin_diagnostic_filename = None
+running_version = None
 
 _SLOTS = frozenset({
     "meta_db", "audio_effect_mappings",
     "config_dir", "dlc_dir", "dlc_dir_env",
     "static_dir", "sloppak_cache_dir", "audio_cache_dir",
-    "get_progression_content",
+    "get_progression_content", "builtin_diagnostic_filename",
+    "running_version",
 })
 
 

lib/routers/diagnostics.py

@@ -0,0 +1,295 @@
+"""Diagnostic bundle export + hardware probe (/api/diagnostics/*).
+
+One-click "Export Diagnostics" in Settings produces a redacted zip combining
+server logs, system info, hardware (CPU/GPU/RAM), plugin inventory, and the
+browser-side console transcript + hardware probe. Bundle format is specified in
+docs/diagnostics-bundle-spec.md.
+
+Extracted verbatim from server.py (R3) except:
+  - the decorators (@app -> @router),
+  - CONFIG_DIR -> appstate.config_dir and _running_version() ->
+    appstate.running_version() (both read through the appstate seam),
+  - the builtin-plugins lookup in _diag_plugins_roots: Path(__file__).parent
+    (the app root when this lived at the top level) ->
+    Path(__file__).resolve().parents[2] (routers -> lib -> app root). The
+    plugins/ dir ships at the app root in every packaging path.
+
+The pure helpers + caps here are re-exported from server.py so the existing
+`server._diag_*` / `server._DIAG_*` tests keep resolving (none monkeypatch them).
+"""
+
+import json
+import logging
+import os
+from pathlib import Path
+
+from fastapi import APIRouter, Body, Response
+
+import appstate
+from dlc_paths import _get_dlc_dir
+from diagnostics_bundle import build_bundle as _diag_build, preview_bundle as _diag_preview
+from diagnostics_hardware import collect as _diag_hardware
+from env_compat import getenv_compat
+
+log = logging.getLogger("feedBack.server")
+router = APIRouter()
+
+
+def _diag_log_file() -> Path | None:
+    raw = os.environ.get("LOG_FILE", "").strip()
+    if not raw:
+        return None
+    return Path(raw)
+
+
+def _diag_plugins_roots() -> list[Path]:
+    """Return all plugin root directories for orphan scanning.
+
+    Includes both the built-in ``plugins/`` directory and
+    ``FEEDBACK_PLUGINS_DIR`` when set, so user-installed plugins and
+    orphans in the external dir are reflected in the bundle.
+    """
+    roots: list[Path] = []
+    user_dir = getenv_compat("FEEDBACK_PLUGINS_DIR", "").strip()
+    if user_dir:
+        p = Path(user_dir)
+        if p.is_dir():
+            roots.append(p)
+    builtin = Path(__file__).resolve().parents[2] / "plugins"  # R3: app root from lib/routers/
+    if builtin not in roots:
+        roots.append(builtin)
+    return roots
+
+
+def _diag_coerce_bool(v, *, default: bool = True) -> bool:
+    """Coerce a request-side value to bool, accepting both JSON booleans and
+    string representations.
+
+    - Falsy strings: ``"false"``, ``"0"``, ``"no"``, ``""`` → ``False``
+    - ``None`` → *default*
+    - Everything else (including ``"true"``, ``"1"``) → ``True``
+    """
+    if v is None:
+        return default
+    if isinstance(v, bool):
+        return v
+    if isinstance(v, str):
+        return v.strip().lower() not in ("false", "0", "no", "")
+    return bool(v)
+
+
+def _diag_normalize_include(include: dict | None) -> dict:
+    """Coerce request-side flags to the booleans build_bundle expects.
+    Missing keys default to True so a bare {} request still produces
+    the full bundle.
+
+    Accepts both JSON booleans (``true``/``false``) and string
+    representations so callers that serialize flags as strings behave
+    consistently with the preview endpoint:
+    - Falsy strings: ``"false"``, ``"0"``, ``"no"``, ``""`` → ``False``
+    - Everything else (including ``"true"``, ``"1"``, ``"yes"``) → ``True``
+    """
+    keys = ("system", "hardware", "logs", "console", "plugins")
+    if not isinstance(include, dict):
+        return {k: True for k in keys}
+
+    return {k: _diag_coerce_bool(include.get(k), default=True) for k in keys}
+
+
+# Server-side caps on client-supplied payload sections.  diagnostics.js
+# enforces a 500-entry / ~250 KB ring buffer on the browser side; these
+# bounds give generous headroom while still preventing a crafted POST from
+# forcing the server to allocate arbitrarily large in-memory bundles.
+_DIAG_MAX_CONSOLE_ENTRIES = 1000          # hard cap: truncate silently
+_DIAG_MAX_CONSOLE_BYTES = 2 * 1024 * 1024  # 2 MB hard cap on total console list
+_DIAG_MAX_CLIENT_PAYLOAD_BYTES = 2 * 1024 * 1024   # 2 MB per dict section
+_DIAG_MAX_CONTRIBUTIONS_BYTES = 4 * 1024 * 1024    # 4 MB aggregate cap for contributions
+
+
+def _diag_cap_console(v) -> list | None:
+    """Return *v* if it is a list, truncated to _DIAG_MAX_CONSOLE_ENTRIES entries
+    and _DIAG_MAX_CONSOLE_BYTES total.  Entries are accumulated until either cap
+    is reached; no partial-entry splitting occurs."""
+    if not isinstance(v, list):
+        return None
+    result = v[:_DIAG_MAX_CONSOLE_ENTRIES]
+    # Also enforce a byte cap — the count cap alone does not bound memory when
+    # entries contain arbitrarily large strings.
+    try:
+        out = []
+        total = 0
+        for entry in result:
+            encoded = json.dumps(entry, separators=(",", ":")).encode("utf-8", errors="replace")
+            if total + len(encoded) > _DIAG_MAX_CONSOLE_BYTES:
+                break
+            out.append(entry)
+            total += len(encoded)
+        return out
+    except (TypeError, ValueError):
+        return None
+
+
+def _diag_cap_dict(v) -> dict | None:
+    """Return *v* if it is a dict whose JSON serialisation fits within
+    _DIAG_MAX_CLIENT_PAYLOAD_BYTES, otherwise return None."""
+    if not isinstance(v, dict):
+        return None
+    try:
+        encoded = json.dumps(v, separators=(",", ":")).encode("utf-8", errors="replace")
+    except (TypeError, ValueError) as e:
+        log.warning("diagnostics client payload is not JSON-serialisable, dropping: %s", e)
+        return None
+    if len(encoded) > _DIAG_MAX_CLIENT_PAYLOAD_BYTES:
+        return None
+    return v
+
+
+def _diag_cap_contributions(v, known_ids=None) -> dict | None:
+    """Apply per-plugin and aggregate size caps on client_contributions.
+
+    Unlike _diag_cap_dict(), which drops the whole dict when any plugin
+    exceeds the limit, this function caps each plugin independently so
+    one noisy plugin does not silence every other plugin's contribution.
+
+    Parameters
+    ----------
+    v:
+        The raw contributions dict from the POST payload.
+    known_ids:
+        When provided, contributions from plugins not in this set are
+        skipped *before* serialisation, preventing a malicious caller
+        from forcing the server to JSON-encode hundreds of near-limit
+        payloads that ``build_bundle()`` would later discard anyway.
+        ``None`` means "accept all plugin ids" (used in tests / preview).
+    """
+    if not isinstance(v, dict):
+        return None
+    result = {}
+    total_bytes = 0
+    for pid, contribution in v.items():
+        if not isinstance(pid, str):
+            continue
+        # Filter unknown plugin ids early — before serialising — so a
+        # crafted request cannot force large allocations for plugins that
+        # build_bundle() would drop.
+        if known_ids is not None and pid not in known_ids:
+            continue
+        try:
+            encoded = json.dumps(contribution, separators=(",", ":")).encode("utf-8", errors="replace")
+        except (TypeError, ValueError) as e:
+            log.warning(
+                "client_contributions[%r] is not JSON-serialisable, dropping: %s", pid, e
+            )
+            continue
+        if len(encoded) > _DIAG_MAX_CLIENT_PAYLOAD_BYTES:
+            log.warning(
+                "client_contributions[%r] exceeds %d bytes, dropping",
+                pid, _DIAG_MAX_CLIENT_PAYLOAD_BYTES,
+            )
+            continue
+        if total_bytes + len(encoded) > _DIAG_MAX_CONTRIBUTIONS_BYTES:
+            log.warning(
+                "client_contributions aggregate size limit (%d bytes) reached, "
+                "dropping remaining entries",
+                _DIAG_MAX_CONTRIBUTIONS_BYTES,
+            )
+            break
+        result[pid] = contribution
+        total_bytes += len(encoded)
+    return result or None
+
+
+@router.post("/api/diagnostics/export")
+def export_diagnostics(payload: dict = Body(default_factory=dict)):
+    """Build a diagnostic bundle and stream it back as a zip download.
+
+    The browser layers in `client_console`, `client_hardware`,
+    `client_ua`, and `local_storage` before posting; the server adds
+    server logs, hardware, plugin inventory, and packages everything
+    into a single zip.
+
+    Errors during plugin diagnostics callables are caught and logged
+    to the bundle's manifest `notes` rather than failing the export.
+    """
+    from plugins import LOADED_PLUGINS, PLUGINS_LOCK
+
+    redact = _diag_coerce_bool(payload.get("redact", True), default=True)
+    include = _diag_normalize_include(payload.get("include"))
+    client_console = _diag_cap_console(payload.get("client_console"))
+    client_hardware = _diag_cap_dict(payload.get("client_hardware"))
+    client_ua = _diag_cap_dict(payload.get("client_ua"))
+    local_storage = _diag_cap_dict(payload.get("local_storage"))
+    # Fetch the plugin list first so we can filter contributions to known
+    # plugin ids before serialising — prevents a crafted request from
+    # forcing large allocations for plugins build_bundle() would drop.
+    with PLUGINS_LOCK:
+        plugins_snapshot = list(LOADED_PLUGINS)
+    known_ids = {p.get("id") for p in plugins_snapshot if isinstance(p.get("id"), str)}
+    client_contributions = _diag_cap_contributions(
+        payload.get("client_contributions"), known_ids=known_ids
+    )
+
+    zip_bytes, filename, _manifest = _diag_build(
+        feedBack_version=appstate.running_version(),
+        config_dir=appstate.config_dir,
+        dlc_dir=_get_dlc_dir(),
+        log_file=_diag_log_file(),
+        loaded_plugins=plugins_snapshot,
+        include=include,
+        redact=redact,
+        client_console=client_console,
+        client_hardware=client_hardware,
+        client_ua=client_ua,
+        local_storage=local_storage,
+        client_contributions=client_contributions,
+        log=log,
+        plugins_root=_diag_plugins_roots(),
+    )
+    return Response(
+        content=zip_bytes,
+        media_type="application/zip",
+        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
+    )
+
+
+@router.get("/api/diagnostics/preview")
+def preview_diagnostics(
+    redact: bool = True,
+    system: bool = True,
+    hardware: bool = True,
+    logs: bool = True,
+    console: bool = True,
+    plugins: bool = True,
+):
+    """Return what `/api/diagnostics/export` would produce, minus the
+    actual file contents — file tree, sizes, schemas, redaction counts.
+    Lets the Settings UI show the user what's about to be sent."""
+    from plugins import LOADED_PLUGINS, PLUGINS_LOCK
+
+    include = {
+        "system": system,
+        "hardware": hardware,
+        "logs": logs,
+        "console": console,
+        "plugins": plugins,
+    }
+    with PLUGINS_LOCK:
+        plugins_snapshot = list(LOADED_PLUGINS)
+    return _diag_preview(
+        feedBack_version=appstate.running_version(),
+        config_dir=appstate.config_dir,
+        dlc_dir=_get_dlc_dir(),
+        log_file=_diag_log_file(),
+        loaded_plugins=plugins_snapshot,
+        include=include,
+        redact=redact,
+        log=log,
+        plugins_root=_diag_plugins_roots(),
+    )
+
+
+@router.get("/api/diagnostics/hardware")
+def diagnostics_hardware():
+    """Backend hardware probe (cross-platform). Reusable independently
+    of the bundle export — handy for "what's my GPU" plugin queries."""
+    return _diag_hardware()

lib/routers/profile.py

@@ -0,0 +1,138 @@
+"""Player profile — identity, avatars (bundled + custom uploads), and progress.
+
+Extracted verbatim from ``server.py`` (R3); edits: ``@app`` -> ``@router``,
+``meta_db`` -> ``appstate.meta_db``, ``CONFIG_DIR``/``STATIC_DIR`` ->
+``appstate.config_dir``/``appstate.static_dir`` (seam), ``_clean_str`` from
+``reqfields``, ``_get_progression_content()`` ->
+``appstate.get_progression_content()``. The bundled-avatar lister moves with it.
+"""
+
+import logging
+import secrets
+
+from fastapi import APIRouter
+from fastapi.responses import FileResponse, JSONResponse
+
+import appstate
+from reqfields import _clean_str
+
+log = logging.getLogger("feedBack.server")
+
+router = APIRouter()
+
+
+def _list_bundled_avatars() -> list[str]:
+    """Bundled default avatar filenames under static/v3/avatars/."""
+    d = appstate.static_dir / "v3" / "avatars"
+    if not d.is_dir():
+        return []
+    exts = {".svg", ".png", ".webp"}
+    return sorted(
+        p.name for p in d.iterdir()
+        if p.is_file() and p.suffix.lower() in exts and not p.name.startswith(".")
+    )
+
+
+@router.get("/api/profile")
+def api_get_profile():
+    profile = appstate.meta_db.get_profile()
+    # Equipped cosmetics ride along (resolved to their payloads) so the theme
+    # and avatar frame apply at boot without an extra request. Never let a
+    # cosmetics/content problem break the profile read.
+    cosmetics = {}
+    try:
+        shop = appstate.get_progression_content()["shop"]
+        for slot, item_id in appstate.meta_db.get_equipped().items():
+            item = shop.get(item_id)
+            if item:
+                cosmetics[slot] = {"item_id": item_id, "payload": item["payload"]}
+    except Exception:
+        log.warning("profile cosmetics enrich failed", exc_info=True)
+    profile["cosmetics"] = cosmetics
+    return profile
+
+
+
+@router.post("/api/profile")
+def api_set_profile(data: dict):
+    """Set/update the player profile. Body: {display_name, avatar:{type,value}}.
+    avatar.type is 'default' (value = bundled filename) or 'upload' (value =
+    the /api/profile/avatar/<name> URL returned by the upload endpoint); omit
+    avatar to keep the existing one (name-only edit)."""
+    name = _clean_str(data.get("display_name"))
+    if not (1 <= len(name) <= 32):
+        return JSONResponse({"error": "Display name must be 1–32 characters."}, status_code=400)
+    avatar = data.get("avatar")
+    if avatar is None:
+        avatar = {}            # omitted → keep the current avatar (name-only edit)
+    elif not isinstance(avatar, dict):
+        return JSONResponse({"error": "avatar must be an object."}, status_code=400)
+    atype = avatar.get("type")
+    aval = _clean_str(avatar.get("value"))
+    avatar_url = None
+    if atype == "default":
+        if aval not in _list_bundled_avatars():
+            return JSONResponse({"error": "Unknown default avatar."}, status_code=400)
+        avatar_url = f"/static/v3/avatars/{aval}"
+    elif atype == "upload":
+        from safepath import safe_join
+        fname = aval.rsplit("/", 1)[-1] if aval.startswith("/api/profile/avatar/") else ""
+        target = safe_join(appstate.config_dir / "avatars", fname) if fname else None
+        if target is None or not target.is_file():
+            return JSONResponse({"error": "Uploaded avatar not found."}, status_code=400)
+        avatar_url = f"/api/profile/avatar/{fname}"
+    elif atype:
+        return JSONResponse({"error": "Unknown avatar type."}, status_code=400)
+    # atype None/missing → keep the current avatar (name-only edit).
+    return appstate.meta_db.set_profile(name, avatar_url)
+
+
+@router.get("/api/profile/avatars")
+def api_list_avatars():
+    return [{"name": n, "url": f"/static/v3/avatars/{n}"} for n in _list_bundled_avatars()]
+
+
+@router.post("/api/profile/avatar")
+def api_upload_avatar(data: dict):
+    """Upload a custom avatar as base64 (mirrors the album-art upload pattern).
+    Re-encodes to a ≤512px PNG under appstate.config_dir/avatars/."""
+    import base64
+    import io
+    b64 = data.get("image", "")
+    if not isinstance(b64, str) or not b64:
+        return JSONResponse({"error": "No image data"}, status_code=400)
+    if "," in b64:
+        b64 = b64.split(",", 1)[1]
+    try:
+        raw = base64.b64decode(b64)
+    except Exception:
+        return JSONResponse({"error": "Invalid base64"}, status_code=400)
+    if len(raw) > 6 * 1024 * 1024:
+        return JSONResponse({"error": "Image too large (max 6 MB)."}, status_code=400)
+    avatars_dir = appstate.config_dir / "avatars"
+    avatars_dir.mkdir(parents=True, exist_ok=True)
+    try:
+        from PIL import Image
+        img = Image.open(io.BytesIO(raw)).convert("RGB")
+        img.thumbnail((512, 512))
+        fname = f"upload-{secrets.token_hex(4)}.png"  # token busts caches on change
+        img.save(str(avatars_dir / fname), "PNG")
+    except Exception as e:
+        return JSONResponse({"error": f"Invalid image: {e}"}, status_code=400)
+    return {"url": f"/api/profile/avatar/{fname}"}
+
+
+@router.get("/api/profile/avatar/{name}")
+def api_get_avatar(name: str):
+    from safepath import safe_join
+    target = safe_join(appstate.config_dir / "avatars", name)
+    if target is None or not target.is_file():
+        return JSONResponse({"error": "not found"}, status_code=404)
+    return FileResponse(str(target), media_type="image/png")
+
+
+@router.get("/api/profile/progress")
+def api_profile_progress():
+    """One call for the whole profile badge: {level, xp, xp_in_level,
+    xp_to_next, current_streak, best_streak, last_active_date}."""
+    return appstate.meta_db.get_progress()

lib/routers/progression.py

@@ -0,0 +1,230 @@
+"""Progression (spec 010) — mastery rank, challenges, quests, onboarding paths.
+
+Extracted verbatim from ``server.py`` (R3); edits: ``@app`` -> ``@router``,
+``meta_db`` -> ``appstate.meta_db``, ``_clean_str`` from ``reqfields``, and the
+two shared server accessors read through the seam:
+``_get_progression_content()`` -> ``appstate.get_progression_content()`` and
+``_builtin_diagnostic_filename()`` -> ``appstate.builtin_diagnostic_filename()``.
+The exclusive helpers (_goal_ui_progress, _progression_overview) + the
+_PROGRESSION_EVENT_TYPES whitelist move with it.
+"""
+
+import math
+
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+
+import appstate
+from reqfields import _clean_str
+
+router = APIRouter()
+
+
+def _goal_ui_progress(goal: dict, state: dict, streak: int, xp_total: int) -> tuple:
+    """(count, target) for a challenge/quest progress bar. Count goals show
+    n/target; threshold goals show how far the live stat is along the line."""
+    import progression as progression_mod
+    gtype = goal.get("type")
+    if gtype in progression_mod.COUNT_GOAL_TYPES:
+        target = int(goal.get("target") or 1)
+        count = target if state.get("completed") else min(int(state.get("count") or 0), target)
+        return count, target
+    if gtype == "streak_reached":
+        target = int(goal.get("days") or 1)
+        return (target if state.get("completed") else min(streak, target)), target
+    if gtype == "db_earned":
+        target = int(goal.get("amount") or 1)
+        return (target if state.get("completed") else min(xp_total, target)), target
+    return 0, 1
+
+
+def _progression_overview() -> dict:
+    """The full GET /api/progression payload (also the capability `inspect`
+    result): rank, onboarding, per-path challenge checklists, quests, wallet."""
+    import progression as progression_mod
+    from datetime import datetime as _dt
+    content = appstate.get_progression_content()
+    now = _dt.now()
+    appstate.meta_db.ensure_quest_period(content, now)
+
+    state = appstate.meta_db.get_progression_state()
+    player_paths = appstate.meta_db.get_player_paths()
+    challenge_state = appstate.meta_db.get_challenge_state()
+    wallet = appstate.meta_db.get_wallet()
+    streak_progress = appstate.meta_db.get_progress()
+    streak = int(streak_progress.get("current_streak") or 0)
+    xp_total = wallet["lifetime_db"]
+    keys = progression_mod.period_keys(now)
+
+    def _path_order(pid):
+        pdef = content["paths"].get(pid) or {}
+        return (pdef.get("order") or 0, pid)
+
+    paths_payload = []
+    for pid in sorted(player_paths, key=_path_order):
+        pdef = content["paths"].get(pid)
+        level = player_paths[pid]
+        if not pdef:
+            # Path selected under older content that no longer ships: keep its
+            # rank contribution visible rather than silently dropping it.
+            paths_payload.append({"id": pid, "name": pid, "icon": "", "level": level,
+                                  "max_level": level, "next": None})
+            continue
+        next_block = None
+        active = progression_mod.active_challenges(content, pid, level)
+        if active:
+            level_def = next(e for e in pdef["levels"] if e["level"] == level + 1)
+            challenges = []
+            completed_count = 0
+            for ch in active:
+                st = challenge_state.get(ch["id"]) or {}
+                count, target = _goal_ui_progress(ch["goal"], st, streak, xp_total)
+                if st.get("completed"):
+                    completed_count += 1
+                challenges.append({
+                    "id": ch["id"],
+                    "title": ch["title"],
+                    "description": ch["description"],
+                    "count": count,
+                    "target": target,
+                    "completed": bool(st.get("completed")),
+                    "completed_at": st.get("completed_at"),
+                })
+            next_block = {
+                "level": level + 1,
+                "required": level_def["required"],
+                "completed": completed_count,
+                "challenges": challenges,
+            }
+        paths_payload.append({
+            "id": pid,
+            "name": pdef["name"],
+            "icon": pdef["icon"],
+            "level": level,
+            "max_level": progression_mod.path_max_level(content, pid),
+            "next": next_block,
+        })
+
+    available = [
+        {"id": pid, "name": pdef["name"], "icon": pdef["icon"]}
+        for pid, pdef in sorted(content["paths"].items(), key=lambda kv: (kv[1].get("order") or 0, kv[0]))
+        if pid not in player_paths
+    ]
+
+    quest_rows = appstate.meta_db.get_quest_rows(keys)
+    quests_payload = {}
+    for period_type in ("daily", "weekly"):
+        pool = content["quests"][period_type]["pool"]
+        quests = []
+        for row in quest_rows:
+            if row["period_type"] != period_type:
+                continue
+            qdef = pool.get(row["quest_id"])
+            if not qdef:
+                continue  # removed from the pool mid-period: hide, keep the row
+            count, target = _goal_ui_progress(qdef["goal"], row, streak, xp_total)
+            quests.append({
+                "id": row["quest_id"],
+                "title": qdef["title"],
+                "description": qdef["description"],
+                "reward_db": row["reward_db"],
+                "count": count,
+                "target": target,
+                "completed": row["completed"],
+                "completed_at": row["completed_at"],
+            })
+        quests_payload[period_type] = {
+            "period_key": keys[period_type],
+            "resets_at": progression_mod.period_resets_at(period_type, now).isoformat(),
+            "quests": quests,
+        }
+
+    return {
+        "mastery_rank": progression_mod.mastery_rank(state["calibration_status"], player_paths),
+        "onboarding": {
+            "calibration_status": state["calibration_status"],
+            "calibration_completed_at": state["calibration_completed_at"],
+            "diagnostic_filename": appstate.builtin_diagnostic_filename(),
+        },
+        "paths": paths_payload,
+        "available_paths": available,
+        "quests": quests_payload,
+        "wallet": wallet,
+    }
+
+
+@router.get("/api/progression")
+def api_progression():
+    return _progression_overview()
+
+
+@router.post("/api/progression/paths")
+def api_progression_add_paths(data: dict):
+    """Select instrument paths. Body: {add: [path_id, ...]}. Idempotent;
+    removal is unsupported (Mastery Rank never decreases)."""
+    add = data.get("add")
+    if not isinstance(add, list) or not add:
+        return JSONResponse({"error": "add must be a non-empty list of path ids"}, status_code=400)
+    content = appstate.get_progression_content()
+    for pid in add:
+        if not isinstance(pid, str) or pid not in content["paths"]:
+            return JSONResponse({"error": f"unknown path: {pid!r}"}, status_code=400)
+    appstate.meta_db.add_player_paths(add)
+    return _progression_overview()
+
+
+@router.post("/api/progression/onboarding")
+def api_progression_onboarding(data: dict):
+    """Onboarding calibration choice. Body: {action: "skip"} — completing the
+    calibration needs no endpoint, it flows through the normal /api/stats path."""
+    if _clean_str(data.get("action")) != "skip":
+        return JSONResponse({"error": "action must be 'skip'"}, status_code=400)
+    # Spec invariant: onboarding requires picking at least one instrument path
+    # before finishing, so skipping straight to rank 1 with no paths would
+    # leave a rank that can never grow. Only enforced when the content bundle
+    # actually defines paths — broken/empty content must never brick onboarding.
+    if appstate.get_progression_content()["paths"] and not appstate.meta_db.get_player_paths():
+        return JSONResponse(
+            {"error": "select at least one instrument path before skipping calibration"},
+            status_code=400,
+        )
+    appstate.meta_db.skip_calibration()
+    return _progression_overview()
+
+
+# Externally postable progression events. song_completed is deliberately NOT
+# here: it is server-derived inside /api/stats so the scored-session authority
+# stays in one place.
+_PROGRESSION_EVENT_TYPES = {"minigame_run"}
+
+
+@router.post("/api/progression/events")
+def api_progression_events(data: dict):
+    """Generic progression-event intake for plugins (capability `record-event`).
+    Body: {type, payload}. Whitelisted types, scalar payload values only."""
+    etype = _clean_str(data.get("type"))
+    if etype not in _PROGRESSION_EVENT_TYPES:
+        return JSONResponse(
+            {"error": f"event type must be one of {sorted(_PROGRESSION_EVENT_TYPES)}"},
+            status_code=400,
+        )
+    payload = data.get("payload")
+    if payload is None:
+        payload = {}
+    if not isinstance(payload, dict) or len(payload) > 16:
+        return JSONResponse({"error": "payload must be a small object"}, status_code=400)
+    clean = {}
+    for key, value in payload.items():
+        if not isinstance(key, str) or len(key) > 64:
+            return JSONResponse({"error": "payload keys must be short strings"}, status_code=400)
+        if value is None:
+            continue
+        if isinstance(value, bool) or (
+            not isinstance(value, (int, float, str))
+        ) or (isinstance(value, float) and not math.isfinite(value)) or (
+            isinstance(value, str) and len(value) > 256
+        ):
+            return JSONResponse({"error": "payload values must be short strings or finite numbers"}, status_code=400)
+        clean[key] = value
+    summary = appstate.meta_db.record_progression_event(etype, clean, appstate.get_progression_content())
+    return {"ok": True, "progression": summary}

lib/routers/stats.py

@@ -0,0 +1,233 @@
+"""Gameplay scoring — XP award + per-song practice stats (record / recent / best /
+top / per-song). The `/api/stats/{filename:path}` route is registered LAST so its
+catch-all doesn't shadow the fixed /recent /best /top paths.
+
+Extracted verbatim from ``server.py`` (R3); edits: ``@app`` -> ``@router``,
+``meta_db`` -> ``appstate.meta_db``, ``_get_progression_content()`` /
+``_builtin_diagnostic_filename()`` read through the seam.
+"""
+
+import logging
+import math
+
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+
+import appstate
+from metadata_db import _as_int
+from reqfields import _clean_str
+
+log = logging.getLogger("feedBack.server")
+
+router = APIRouter()
+
+
+@router.post("/api/xp/award")
+def api_award_xp(data: dict):
+    """Award XP into the unified store. Body: {source, amount}. Returns the
+    new progress payload. The single XP authority — song-play, minigames, and
+    tutorials all feed this (no second curve)."""
+    try:
+        amount = _as_int(data.get("amount", 0))   # rejects bool / non-integral / inf
+    except (TypeError, ValueError, OverflowError):
+        return JSONResponse({"error": "amount must be an integer"}, status_code=400)
+    # Upper-bound it: an unbounded value overflows SQLite's 64-bit INTEGER on
+    # bind (→ 500) and no real run awards anywhere near this.
+    if not (0 <= amount <= 10_000_000):
+        return JSONResponse({"error": "amount must be between 0 and 10,000,000"}, status_code=400)
+    appstate.meta_db.award_xp(amount)
+    return appstate.meta_db.get_progress()
+
+
+@router.post("/api/stats")
+def api_record_stats(data: dict):
+    """Record a play. With `score`+`accuracy` → a scored session (plays += 1,
+    best_* = max, last_* = new) plus unified-XP + streak side-effects. With
+    only `lastPlayPosition`/`last_position` → a lightweight resume-position
+    touch (no plays change) so Continue-Playing works for non-scored plays."""
+    filename = _clean_str(data.get("filename"))
+    if not filename:
+        return JSONResponse({"error": "filename required"}, status_code=400)
+    # The recorder hands us URL-encoded filenames; canonicalize to the library
+    # key so stored rows line up with `songs` (and so the arrangement-count bound
+    # below resolves the real song). See MetadataDB._canonical_song_filename.
+    filename = appstate.meta_db._canonical_song_filename(filename)
+    arr_raw = data.get("arrangement", 0)
+    if arr_raw is None:
+        arrangement = 0
+    else:
+        try:
+            arrangement = _as_int(arr_raw)   # rejects bool / non-integral (1.9) / inf
+        except (TypeError, ValueError, OverflowError):
+            return JSONResponse({"error": "arrangement must be a non-negative integer"}, status_code=400)
+    # Reject (don't silently coerce to 0) so a malformed/out-of-range index
+    # can't corrupt arrangement 0's stats; also keeps it bindable to INTEGER.
+    if not (0 <= arrangement < 2**63):
+        return JSONResponse({"error": "arrangement must be a non-negative integer"}, status_code=400)
+    # Bound against the song's real arrangement count when it's a known library
+    # song, so a bad index can't create fake arrangement buckets that poison the
+    # per-song aggregate / Continue. Skipped when the song isn't in the library
+    # yet (count unknown — dead-song reads are filtered anyway).
+    _acount = appstate.meta_db.arrangement_count(filename)
+    if _acount and arrangement >= _acount:
+        return JSONResponse({"error": "arrangement out of range for this song"}, status_code=400)
+    score = data.get("score")
+    accuracy = data.get("accuracy")
+    last_pos = data.get("lastPlayPosition", data.get("last_position"))
+    if isinstance(last_pos, bool):   # float(False)=0.0 would otherwise store a bogus position
+        return JSONResponse({"error": "lastPlayPosition must be a finite number"}, status_code=400)
+
+    # A scored session needs BOTH score and accuracy. Exactly one provided is
+    # ambiguous — don't silently fall through to the position-only branch.
+    if (score is None) != (accuracy is None):
+        return JSONResponse({"error": "score and accuracy must be provided together"}, status_code=400)
+
+    if score is not None and accuracy is not None:
+        # Reject booleans explicitly — float(True) would otherwise record a play.
+        if isinstance(score, bool) or isinstance(accuracy, bool):
+            return JSONResponse({"error": "score/accuracy must be finite numbers"}, status_code=400)
+        # Reject NaN/Inf too: round(inf) raises OverflowError (→ 500), and a
+        # stored Inf/NaN later breaks JSON serialization of /api/stats reads.
+        try:
+            score = float(score)
+            accuracy = float(accuracy)
+            if not (math.isfinite(score) and math.isfinite(accuracy)):
+                raise ValueError("non-finite")
+            score = int(round(score))
+        except (TypeError, ValueError, OverflowError):
+            return JSONResponse({"error": "score/accuracy must be finite numbers"}, status_code=400)
+        # A huge-but-finite score passes isfinite() yet overflows SQLite's
+        # 64-bit INTEGER on bind (→ 500). Bound it to the int64 range.
+        if not (0 <= score < 2**63):
+            return JSONResponse({"error": "score out of range"}, status_code=400)
+        # accuracy is a 0..1 fraction (the recorder's contract); reject
+        # out-of-range values so they don't surface as >100% / negative in
+        # /api/stats/best and the badge UI.
+        if not (0 <= accuracy <= 1):
+            return JSONResponse({"error": "accuracy must be between 0 and 1"}, status_code=400)
+        # Validate the optional resume position in this branch too (the
+        # position-only branch below already rejects non-finite).
+        if last_pos is not None:
+            try:
+                last_pos = float(last_pos)
+                if not math.isfinite(last_pos):
+                    raise ValueError("non-finite")
+            except (TypeError, ValueError, OverflowError):
+                return JSONResponse({"error": "lastPlayPosition must be a finite number"}, status_code=400)
+        row = appstate.meta_db.record_session(filename, arrangement, score=score,
+                                     accuracy=accuracy, last_position=last_pos)
+        # Unified XP + streak side-effects — never let these drop the stat write.
+        progress = None
+        try:
+            from xp import xp_for_run
+            from datetime import date
+            appstate.meta_db.award_xp(xp_for_run(score))
+            appstate.meta_db.record_active_day(date.today().isoformat())
+            progress = appstate.meta_db.get_progress()
+        except Exception:
+            log.warning("stats side-effects (xp/streak) failed", exc_info=True)
+        # Progression engine (spec 010) — same never-drop-the-stat-write
+        # contract. Scored sessions are the server-derived `song_completed`
+        # authority (scored == note detection by construction); instrument is
+        # resolved from library arrangement metadata, after the XP award so
+        # db_earned goals see this run's Decibels.
+        progression_summary = None
+        try:
+            import progression as progression_mod
+            instrument = progression_mod.instrument_for_arrangement(
+                appstate.meta_db.arrangement_entry(filename, arrangement)
+            )
+            progression_summary = appstate.meta_db.record_progression_event(
+                "song_completed",
+                {
+                    "filename": filename,
+                    "instrument": instrument,
+                    "accuracy": accuracy,
+                    "score": score,
+                    "is_diagnostic": filename == appstate.builtin_diagnostic_filename(),
+                },
+                appstate.get_progression_content(),
+            )
+        except Exception:
+            log.warning("stats side-effects (progression) failed", exc_info=True)
+        return {"stats": row, "progress": progress, "progression": progression_summary}
+
+    # Position-only touch.
+    if last_pos is None:
+        return JSONResponse(
+            {"error": "provide score+accuracy (scored) or lastPlayPosition (resume)"},
+            status_code=400,
+        )
+    try:
+        pos = float(last_pos)
+        if not math.isfinite(pos):
+            raise ValueError("non-finite")
+        row = appstate.meta_db.touch_position(filename, arrangement, pos)
+    except (TypeError, ValueError, OverflowError):
+        return JSONResponse({"error": "lastPlayPosition must be a finite number"}, status_code=400)
+    # A resume session still counts as playing today: advance the streak (no XP —
+    # that's scoring-only) so a non-scored practice day keeps the streak alive,
+    # consistent with these sessions also surfacing in recent / continue.
+    progress = None
+    try:
+        from datetime import date
+        appstate.meta_db.record_active_day(date.today().isoformat())
+        progress = appstate.meta_db.get_progress()
+    except Exception:
+        log.warning("stats side-effects (streak) failed", exc_info=True)
+    return {"stats": row, "progress": progress}
+
+
+@router.get("/api/stats/recent")
+def api_recent_stats(limit: int = 12):
+    """Recently-played rows joined to song metadata for 'Jump back in'."""
+    from urllib.parse import quote
+    out = []
+    for r in appstate.meta_db.recent_stats(limit):
+        meta = appstate.meta_db.conn.execute(
+            "SELECT title, artist, tuning_name FROM songs WHERE filename = ?",
+            (r["filename"],),
+        ).fetchone()
+        title, artist, tuning_name = meta if meta else (None, None, None)
+        out.append({
+            **r,
+            "title": title or r["filename"],
+            "artist": artist or "",
+            "tuning_name": tuning_name or "",
+            "art_url": f"/api/song/{quote(r['filename'])}/art",
+        })
+    return out
+
+
+@router.get("/api/stats/best")
+def api_stats_best():
+    """{filename: best_accuracy} for all songs with a recorded best — one call
+    to badge the library grid (defined before the {filename} catch-all)."""
+    return appstate.meta_db.best_accuracy_map()
+
+
+@router.get("/api/stats/top")
+def api_top_stats(limit: int = 5):
+    """Top scored songs (best first), joined to song metadata, for the profile
+    'Your best scores' panel (defined before the {filename} catch-all)."""
+    from urllib.parse import quote
+    out = []
+    for r in appstate.meta_db.top_stats(limit):
+        meta = appstate.meta_db.conn.execute(
+            "SELECT title, artist, tuning_name FROM songs WHERE filename = ?",
+            (r["filename"],),
+        ).fetchone()
+        title, artist, tuning_name = meta if meta else (None, None, None)
+        out.append({
+            **r,
+            "title": title or r["filename"],
+            "artist": artist or "",
+            "tuning_name": tuning_name or "",
+            "art_url": f"/api/song/{quote(r['filename'])}/art",
+        })
+    return out
+
+
+@router.get("/api/stats/{filename:path}")
+def api_song_stats(filename: str):
+    return appstate.meta_db.get_song_stats(filename)

lib/routers/version.py

@@ -0,0 +1,81 @@
+"""App version + source/license URLs (/api/version).
+
+Extracted verbatim from ``server.py`` (R3) except the decorator (``@app`` ->
+``@router``) and the VERSION-file lookup: ``Path(__file__).parent`` (app root
+when this lived at the top level) -> ``Path(__file__).resolve().parents[2]``
+(routers -> lib -> app root). VERSION ships at the app root in every packaging
+path (Dockerfile COPY, desktop bundle).
+"""
+
+import os
+from pathlib import Path
+
+from fastapi import APIRouter
+
+router = APIRouter()
+
+
+def _safe_http_url(raw):
+    """Return `raw` stripped + trailing-slash-stripped if it parses as an
+    http(s) URL with a non-empty host; else None.
+
+    Used to validate operator-supplied `APP_SOURCE_URL` / `APP_LICENSE_URL`
+    env vars before they reach `<a href>` in the UI. A bare prefix check
+    like `startswith(("http://","https://"))` accepts malformed inputs
+    such as `"https://"` (no host) or `"https:///foo"` (empty host) that
+    still produce broken hrefs — and, when used as a base for the default
+    `license_url`, garbage like `"https:///blob/main/LICENSE"`.
+    """
+    from urllib.parse import urlsplit
+    if not raw:
+        return None
+    s = raw.strip().rstrip("/")
+    if not s:
+        return None
+    try:
+        parsed = urlsplit(s)
+    except ValueError:
+        return None
+    if parsed.scheme.lower() not in ("http", "https"):
+        return None
+    # `netloc` includes any `user:pass@` and `:port` — strings like
+    # "http://:80/path" have non-empty netloc (":80") but no real
+    # hostname. Validate `hostname` so only URLs with an actual host
+    # are accepted.
+    if not parsed.hostname:
+        return None
+    return s
+
+
+@router.get("/api/version")
+def get_version():
+    env_version = os.environ.get("APP_VERSION", "").strip()
+    if env_version:
+        version = env_version
+    else:
+        version_file = Path(__file__).resolve().parents[2] / "VERSION"  # R3: app root from lib/routers/
+        version = "unknown"
+        if version_file.exists():
+            try:
+                version = version_file.read_text().strip()
+            except (OSError, UnicodeDecodeError):
+                pass
+    default_source_url = "https://github.com/got-feedback/feedBack"
+    # APP_SOURCE_URL / APP_LICENSE_URL flow straight into <a href> in the UI,
+    # so validate with urllib.parse rather than a bare prefix check — a prefix
+    # check accepts malformed values like "https://" (no host) which produce
+    # broken hrefs (and a constructed license_url like "https:///blob/main/LICENSE").
+    # _safe_http_url requires scheme in {http,https} AND a non-empty hostname
+    # (not just netloc — that would still accept port-only authorities like
+    # "http://:80/path"); fall back to the safe default otherwise.
+    source_url = _safe_http_url(os.environ.get("APP_SOURCE_URL")) or default_source_url
+    # APP_LICENSE_URL: explicit override for the LICENSE link. The default
+    # constructed value (source_url + "/blob/main/LICENSE") is GitHub-
+    # specific and assumes the repo's default branch is `main`; non-GitHub
+    # hosts (GitLab, Gitea, self-hosted) need an explicit value.
+    license_url = _safe_http_url(os.environ.get("APP_LICENSE_URL")) or (source_url + "/blob/main/LICENSE")
+    return {
+        "version": version,
+        "source_url": source_url,
+        "license_url": license_url,
+    }

server.py

@@ -55,7 +55,7 @@ from dlc_paths import _get_dlc_dir, _resolve_dlc_path
 # Lives in lib/ because that is the one core dir every packaging path copies.
 import appstate
 # Extracted route modules. They import `appstate`, never `server` — one-way graph.
-from routers import audio_effects, artist_aliases, loops, playlists, ws_highway, chart, wanted, library_extras, shop
+from routers import audio_effects, artist_aliases, loops, playlists, ws_highway, chart, wanted, library_extras, shop, progression, profile, stats, version, diagnostics
 import sloppak as sloppak_mod
 import loosefolder as loosefolder_mod
 # Pure text-matching engine for MusicBrainz enrichment (P8): denoise/score/
@@ -1168,7 +1168,10 @@ def _get_progression_content() -> dict:
 # (the main configure() at import-top runs before this def). The cache global +
 # lock stay in server.py, so the `setattr(server, "_progression_content")` test
 # path is unchanged; routers call `appstate.get_progression_content()`.
-appstate.configure(get_progression_content=_get_progression_content)
+appstate.configure(
+    get_progression_content=_get_progression_content,
+    builtin_diagnostic_filename=_builtin_diagnostic_filename,
+)
 
 
 def _copy_builtin_packs(
@@ -3190,70 +3193,11 @@ def _periodic_rescan():
         time.sleep(300)
 
 
-def _safe_http_url(raw):
-    """Return `raw` stripped + trailing-slash-stripped if it parses as an
-    http(s) URL with a non-empty host; else None.
-
-    Used to validate operator-supplied `APP_SOURCE_URL` / `APP_LICENSE_URL`
-    env vars before they reach `<a href>` in the UI. A bare prefix check
-    like `startswith(("http://","https://"))` accepts malformed inputs
-    such as `"https://"` (no host) or `"https:///foo"` (empty host) that
-    still produce broken hrefs — and, when used as a base for the default
-    `license_url`, garbage like `"https:///blob/main/LICENSE"`.
-    """
-    from urllib.parse import urlsplit
-    if not raw:
-        return None
-    s = raw.strip().rstrip("/")
-    if not s:
-        return None
-    try:
-        parsed = urlsplit(s)
-    except ValueError:
-        return None
-    if parsed.scheme.lower() not in ("http", "https"):
-        return None
-    # `netloc` includes any `user:pass@` and `:port` — strings like
-    # "http://:80/path" have non-empty netloc (":80") but no real
-    # hostname. Validate `hostname` so only URLs with an actual host
-    # are accepted.
-    if not parsed.hostname:
-        return None
-    return s
 
 
-@app.get("/api/version")
-def get_version():
-    env_version = os.environ.get("APP_VERSION", "").strip()
-    if env_version:
-        version = env_version
-    else:
-        version_file = Path(__file__).parent / "VERSION"
-        version = "unknown"
-        if version_file.exists():
-            try:
-                version = version_file.read_text().strip()
-            except (OSError, UnicodeDecodeError):
-                pass
-    default_source_url = "https://github.com/got-feedback/feedBack"
-    # APP_SOURCE_URL / APP_LICENSE_URL flow straight into <a href> in the UI,
-    # so validate with urllib.parse rather than a bare prefix check — a prefix
-    # check accepts malformed values like "https://" (no host) which produce
-    # broken hrefs (and a constructed license_url like "https:///blob/main/LICENSE").
-    # _safe_http_url requires scheme in {http,https} AND a non-empty hostname
-    # (not just netloc — that would still accept port-only authorities like
-    # "http://:80/path"); fall back to the safe default otherwise.
-    source_url = _safe_http_url(os.environ.get("APP_SOURCE_URL")) or default_source_url
-    # APP_LICENSE_URL: explicit override for the LICENSE link. The default
-    # constructed value (source_url + "/blob/main/LICENSE") is GitHub-
-    # specific and assumes the repo's default branch is `main`; non-GitHub
-    # hosts (GitLab, Gitea, self-hosted) need an explicit value.
-    license_url = _safe_http_url(os.environ.get("APP_LICENSE_URL")) or (source_url + "/blob/main/LICENSE")
-    return {
-        "version": version,
-        "source_url": source_url,
-        "license_url": license_url,
-    }
+# ── App version / source URLs ────────────────────────────────────────────────
+# Mounted here (registration order). Implementation in lib/routers/version.py.
+app.include_router(version.router)
 
 
 @app.get("/api/scan-status")
@@ -4662,354 +4606,20 @@ def api_artist_links(name: str):
 def api_artist_links_refresh(name: str):
     """Explicit re-fetch of the cached links (the page's manual Refresh)."""
     return _artist_links_payload(name, force=True)
+# ── Player profile (identity / avatars / progress) ───────────────────────────
+# Mounted here (registration order). Implementation in lib/routers/profile.py.
+app.include_router(profile.router)
 
 
-# ── Player profile / unified XP / streak (fee[dB]ack v0.3.0) ──────────────────
-
-def _list_bundled_avatars() -> list[str]:
-    """Bundled default avatar filenames under static/v3/avatars/."""
-    d = STATIC_DIR / "v3" / "avatars"
-    if not d.is_dir():
-        return []
-    exts = {".svg", ".png", ".webp"}
-    return sorted(
-        p.name for p in d.iterdir()
-        if p.is_file() and p.suffix.lower() in exts and not p.name.startswith(".")
-    )
+# ── Gameplay scoring: XP award + per-song practice stats ─────────────────────
+# Mounted here (registration order; /api/stats/{path} is last inside the router).
+# Implementation in lib/routers/stats.py.
+app.include_router(stats.router)
 
 
-@app.get("/api/profile")
-def api_get_profile():
-    profile = meta_db.get_profile()
-    # Equipped cosmetics ride along (resolved to their payloads) so the theme
-    # and avatar frame apply at boot without an extra request. Never let a
-    # cosmetics/content problem break the profile read.
-    cosmetics = {}
-    try:
-        shop = _get_progression_content()["shop"]
-        for slot, item_id in meta_db.get_equipped().items():
-            item = shop.get(item_id)
-            if item:
-                cosmetics[slot] = {"item_id": item_id, "payload": item["payload"]}
-    except Exception:
-        log.warning("profile cosmetics enrich failed", exc_info=True)
-    profile["cosmetics"] = cosmetics
-    return profile
-
-
-
-@app.post("/api/profile")
-def api_set_profile(data: dict):
-    """Set/update the player profile. Body: {display_name, avatar:{type,value}}.
-    avatar.type is 'default' (value = bundled filename) or 'upload' (value =
-    the /api/profile/avatar/<name> URL returned by the upload endpoint); omit
-    avatar to keep the existing one (name-only edit)."""
-    name = _clean_str(data.get("display_name"))
-    if not (1 <= len(name) <= 32):
-        return JSONResponse({"error": "Display name must be 1–32 characters."}, status_code=400)
-    avatar = data.get("avatar")
-    if avatar is None:
-        avatar = {}            # omitted → keep the current avatar (name-only edit)
-    elif not isinstance(avatar, dict):
-        return JSONResponse({"error": "avatar must be an object."}, status_code=400)
-    atype = avatar.get("type")
-    aval = _clean_str(avatar.get("value"))
-    avatar_url = None
-    if atype == "default":
-        if aval not in _list_bundled_avatars():
-            return JSONResponse({"error": "Unknown default avatar."}, status_code=400)
-        avatar_url = f"/static/v3/avatars/{aval}"
-    elif atype == "upload":
-        from safepath import safe_join
-        fname = aval.rsplit("/", 1)[-1] if aval.startswith("/api/profile/avatar/") else ""
-        target = safe_join(CONFIG_DIR / "avatars", fname) if fname else None
-        if target is None or not target.is_file():
-            return JSONResponse({"error": "Uploaded avatar not found."}, status_code=400)
-        avatar_url = f"/api/profile/avatar/{fname}"
-    elif atype:
-        return JSONResponse({"error": "Unknown avatar type."}, status_code=400)
-    # atype None/missing → keep the current avatar (name-only edit).
-    return meta_db.set_profile(name, avatar_url)
-
-
-@app.get("/api/profile/avatars")
-def api_list_avatars():
-    return [{"name": n, "url": f"/static/v3/avatars/{n}"} for n in _list_bundled_avatars()]
-
-
-@app.post("/api/profile/avatar")
-def api_upload_avatar(data: dict):
-    """Upload a custom avatar as base64 (mirrors the album-art upload pattern).
-    Re-encodes to a ≤512px PNG under CONFIG_DIR/avatars/."""
-    import base64
-    import io
-    b64 = data.get("image", "")
-    if not isinstance(b64, str) or not b64:
-        return JSONResponse({"error": "No image data"}, status_code=400)
-    if "," in b64:
-        b64 = b64.split(",", 1)[1]
-    try:
-        raw = base64.b64decode(b64)
-    except Exception:
-        return JSONResponse({"error": "Invalid base64"}, status_code=400)
-    if len(raw) > 6 * 1024 * 1024:
-        return JSONResponse({"error": "Image too large (max 6 MB)."}, status_code=400)
-    avatars_dir = CONFIG_DIR / "avatars"
-    avatars_dir.mkdir(parents=True, exist_ok=True)
-    try:
-        from PIL import Image
-        img = Image.open(io.BytesIO(raw)).convert("RGB")
-        img.thumbnail((512, 512))
-        fname = f"upload-{secrets.token_hex(4)}.png"  # token busts caches on change
-        img.save(str(avatars_dir / fname), "PNG")
-    except Exception as e:
-        return JSONResponse({"error": f"Invalid image: {e}"}, status_code=400)
-    return {"url": f"/api/profile/avatar/{fname}"}
-
-
-@app.get("/api/profile/avatar/{name}")
-def api_get_avatar(name: str):
-    from safepath import safe_join
-    target = safe_join(CONFIG_DIR / "avatars", name)
-    if target is None or not target.is_file():
-        return JSONResponse({"error": "not found"}, status_code=404)
-    return FileResponse(str(target), media_type="image/png")
-
-
-@app.get("/api/profile/progress")
-def api_profile_progress():
-    """One call for the whole profile badge: {level, xp, xp_in_level,
-    xp_to_next, current_streak, best_streak, last_active_date}."""
-    return meta_db.get_progress()
-
-
-@app.post("/api/xp/award")
-def api_award_xp(data: dict):
-    """Award XP into the unified store. Body: {source, amount}. Returns the
-    new progress payload. The single XP authority — song-play, minigames, and
-    tutorials all feed this (no second curve)."""
-    try:
-        amount = _as_int(data.get("amount", 0))   # rejects bool / non-integral / inf
-    except (TypeError, ValueError, OverflowError):
-        return JSONResponse({"error": "amount must be an integer"}, status_code=400)
-    # Upper-bound it: an unbounded value overflows SQLite's 64-bit INTEGER on
-    # bind (→ 500) and no real run awards anywhere near this.
-    if not (0 <= amount <= 10_000_000):
-        return JSONResponse({"error": "amount must be between 0 and 10,000,000"}, status_code=400)
-    meta_db.award_xp(amount)
-    return meta_db.get_progress()
-
-
-# ── Progression (spec 010): mastery rank, challenges, quests, shop ───────────
-
-def _goal_ui_progress(goal: dict, state: dict, streak: int, xp_total: int) -> tuple:
-    """(count, target) for a challenge/quest progress bar. Count goals show
-    n/target; threshold goals show how far the live stat is along the line."""
-    import progression as progression_mod
-    gtype = goal.get("type")
-    if gtype in progression_mod.COUNT_GOAL_TYPES:
-        target = int(goal.get("target") or 1)
-        count = target if state.get("completed") else min(int(state.get("count") or 0), target)
-        return count, target
-    if gtype == "streak_reached":
-        target = int(goal.get("days") or 1)
-        return (target if state.get("completed") else min(streak, target)), target
-    if gtype == "db_earned":
-        target = int(goal.get("amount") or 1)
-        return (target if state.get("completed") else min(xp_total, target)), target
-    return 0, 1
-
-
-def _progression_overview() -> dict:
-    """The full GET /api/progression payload (also the capability `inspect`
-    result): rank, onboarding, per-path challenge checklists, quests, wallet."""
-    import progression as progression_mod
-    from datetime import datetime as _dt
-    content = _get_progression_content()
-    now = _dt.now()
-    meta_db.ensure_quest_period(content, now)
-
-    state = meta_db.get_progression_state()
-    player_paths = meta_db.get_player_paths()
-    challenge_state = meta_db.get_challenge_state()
-    wallet = meta_db.get_wallet()
-    streak_progress = meta_db.get_progress()
-    streak = int(streak_progress.get("current_streak") or 0)
-    xp_total = wallet["lifetime_db"]
-    keys = progression_mod.period_keys(now)
-
-    def _path_order(pid):
-        pdef = content["paths"].get(pid) or {}
-        return (pdef.get("order") or 0, pid)
-
-    paths_payload = []
-    for pid in sorted(player_paths, key=_path_order):
-        pdef = content["paths"].get(pid)
-        level = player_paths[pid]
-        if not pdef:
-            # Path selected under older content that no longer ships: keep its
-            # rank contribution visible rather than silently dropping it.
-            paths_payload.append({"id": pid, "name": pid, "icon": "", "level": level,
-                                  "max_level": level, "next": None})
-            continue
-        next_block = None
-        active = progression_mod.active_challenges(content, pid, level)
-        if active:
-            level_def = next(e for e in pdef["levels"] if e["level"] == level + 1)
-            challenges = []
-            completed_count = 0
-            for ch in active:
-                st = challenge_state.get(ch["id"]) or {}
-                count, target = _goal_ui_progress(ch["goal"], st, streak, xp_total)
-                if st.get("completed"):
-                    completed_count += 1
-                challenges.append({
-                    "id": ch["id"],
-                    "title": ch["title"],
-                    "description": ch["description"],
-                    "count": count,
-                    "target": target,
-                    "completed": bool(st.get("completed")),
-                    "completed_at": st.get("completed_at"),
-                })
-            next_block = {
-                "level": level + 1,
-                "required": level_def["required"],
-                "completed": completed_count,
-                "challenges": challenges,
-            }
-        paths_payload.append({
-            "id": pid,
-            "name": pdef["name"],
-            "icon": pdef["icon"],
-            "level": level,
-            "max_level": progression_mod.path_max_level(content, pid),
-            "next": next_block,
-        })
-
-    available = [
-        {"id": pid, "name": pdef["name"], "icon": pdef["icon"]}
-        for pid, pdef in sorted(content["paths"].items(), key=lambda kv: (kv[1].get("order") or 0, kv[0]))
-        if pid not in player_paths
-    ]
-
-    quest_rows = meta_db.get_quest_rows(keys)
-    quests_payload = {}
-    for period_type in ("daily", "weekly"):
-        pool = content["quests"][period_type]["pool"]
-        quests = []
-        for row in quest_rows:
-            if row["period_type"] != period_type:
-                continue
-            qdef = pool.get(row["quest_id"])
-            if not qdef:
-                continue  # removed from the pool mid-period: hide, keep the row
-            count, target = _goal_ui_progress(qdef["goal"], row, streak, xp_total)
-            quests.append({
-                "id": row["quest_id"],
-                "title": qdef["title"],
-                "description": qdef["description"],
-                "reward_db": row["reward_db"],
-                "count": count,
-                "target": target,
-                "completed": row["completed"],
-                "completed_at": row["completed_at"],
-            })
-        quests_payload[period_type] = {
-            "period_key": keys[period_type],
-            "resets_at": progression_mod.period_resets_at(period_type, now).isoformat(),
-            "quests": quests,
-        }
-
-    return {
-        "mastery_rank": progression_mod.mastery_rank(state["calibration_status"], player_paths),
-        "onboarding": {
-            "calibration_status": state["calibration_status"],
-            "calibration_completed_at": state["calibration_completed_at"],
-            "diagnostic_filename": _builtin_diagnostic_filename(),
-        },
-        "paths": paths_payload,
-        "available_paths": available,
-        "quests": quests_payload,
-        "wallet": wallet,
-    }
-
-
-@app.get("/api/progression")
-def api_progression():
-    return _progression_overview()
-
-
-@app.post("/api/progression/paths")
-def api_progression_add_paths(data: dict):
-    """Select instrument paths. Body: {add: [path_id, ...]}. Idempotent;
-    removal is unsupported (Mastery Rank never decreases)."""
-    add = data.get("add")
-    if not isinstance(add, list) or not add:
-        return JSONResponse({"error": "add must be a non-empty list of path ids"}, status_code=400)
-    content = _get_progression_content()
-    for pid in add:
-        if not isinstance(pid, str) or pid not in content["paths"]:
-            return JSONResponse({"error": f"unknown path: {pid!r}"}, status_code=400)
-    meta_db.add_player_paths(add)
-    return _progression_overview()
-
-
-@app.post("/api/progression/onboarding")
-def api_progression_onboarding(data: dict):
-    """Onboarding calibration choice. Body: {action: "skip"} — completing the
-    calibration needs no endpoint, it flows through the normal /api/stats path."""
-    if _clean_str(data.get("action")) != "skip":
-        return JSONResponse({"error": "action must be 'skip'"}, status_code=400)
-    # Spec invariant: onboarding requires picking at least one instrument path
-    # before finishing, so skipping straight to rank 1 with no paths would
-    # leave a rank that can never grow. Only enforced when the content bundle
-    # actually defines paths — broken/empty content must never brick onboarding.
-    if _get_progression_content()["paths"] and not meta_db.get_player_paths():
-        return JSONResponse(
-            {"error": "select at least one instrument path before skipping calibration"},
-            status_code=400,
-        )
-    meta_db.skip_calibration()
-    return _progression_overview()
-
-
-# Externally postable progression events. song_completed is deliberately NOT
-# here: it is server-derived inside /api/stats so the scored-session authority
-# stays in one place.
-_PROGRESSION_EVENT_TYPES = {"minigame_run"}
-
-
-@app.post("/api/progression/events")
-def api_progression_events(data: dict):
-    """Generic progression-event intake for plugins (capability `record-event`).
-    Body: {type, payload}. Whitelisted types, scalar payload values only."""
-    etype = _clean_str(data.get("type"))
-    if etype not in _PROGRESSION_EVENT_TYPES:
-        return JSONResponse(
-            {"error": f"event type must be one of {sorted(_PROGRESSION_EVENT_TYPES)}"},
-            status_code=400,
-        )
-    payload = data.get("payload")
-    if payload is None:
-        payload = {}
-    if not isinstance(payload, dict) or len(payload) > 16:
-        return JSONResponse({"error": "payload must be a small object"}, status_code=400)
-    clean = {}
-    for key, value in payload.items():
-        if not isinstance(key, str) or len(key) > 64:
-            return JSONResponse({"error": "payload keys must be short strings"}, status_code=400)
-        if value is None:
-            continue
-        if isinstance(value, bool) or (
-            not isinstance(value, (int, float, str))
-        ) or (isinstance(value, float) and not math.isfinite(value)) or (
-            isinstance(value, str) and len(value) > 256
-        ):
-            return JSONResponse({"error": "payload values must be short strings or finite numbers"}, status_code=400)
-        clean[key] = value
-    summary = meta_db.record_progression_event(etype, clean, _get_progression_content())
-    return {"ok": True, "progression": summary}
+# ── Progression (spec 010) ───────────────────────────────────────────────────
+# Mounted here (registration order). Implementation in lib/routers/progression.py.
+app.include_router(progression.router)
 
 
 # ── Cosmetics shop (spec 010) ────────────────────────────────────────────────
@@ -5017,195 +4627,6 @@ def api_progression_events(data: dict):
 app.include_router(shop.router)
 
 
-# ── Per-song practice stats (fee[dB]ack v0.3.0) ───────────────────────────────
-
-@app.post("/api/stats")
-def api_record_stats(data: dict):
-    """Record a play. With `score`+`accuracy` → a scored session (plays += 1,
-    best_* = max, last_* = new) plus unified-XP + streak side-effects. With
-    only `lastPlayPosition`/`last_position` → a lightweight resume-position
-    touch (no plays change) so Continue-Playing works for non-scored plays."""
-    filename = _clean_str(data.get("filename"))
-    if not filename:
-        return JSONResponse({"error": "filename required"}, status_code=400)
-    # The recorder hands us URL-encoded filenames; canonicalize to the library
-    # key so stored rows line up with `songs` (and so the arrangement-count bound
-    # below resolves the real song). See MetadataDB._canonical_song_filename.
-    filename = meta_db._canonical_song_filename(filename)
-    arr_raw = data.get("arrangement", 0)
-    if arr_raw is None:
-        arrangement = 0
-    else:
-        try:
-            arrangement = _as_int(arr_raw)   # rejects bool / non-integral (1.9) / inf
-        except (TypeError, ValueError, OverflowError):
-            return JSONResponse({"error": "arrangement must be a non-negative integer"}, status_code=400)
-    # Reject (don't silently coerce to 0) so a malformed/out-of-range index
-    # can't corrupt arrangement 0's stats; also keeps it bindable to INTEGER.
-    if not (0 <= arrangement < 2**63):
-        return JSONResponse({"error": "arrangement must be a non-negative integer"}, status_code=400)
-    # Bound against the song's real arrangement count when it's a known library
-    # song, so a bad index can't create fake arrangement buckets that poison the
-    # per-song aggregate / Continue. Skipped when the song isn't in the library
-    # yet (count unknown — dead-song reads are filtered anyway).
-    _acount = meta_db.arrangement_count(filename)
-    if _acount and arrangement >= _acount:
-        return JSONResponse({"error": "arrangement out of range for this song"}, status_code=400)
-    score = data.get("score")
-    accuracy = data.get("accuracy")
-    last_pos = data.get("lastPlayPosition", data.get("last_position"))
-    if isinstance(last_pos, bool):   # float(False)=0.0 would otherwise store a bogus position
-        return JSONResponse({"error": "lastPlayPosition must be a finite number"}, status_code=400)
-
-    # A scored session needs BOTH score and accuracy. Exactly one provided is
-    # ambiguous — don't silently fall through to the position-only branch.
-    if (score is None) != (accuracy is None):
-        return JSONResponse({"error": "score and accuracy must be provided together"}, status_code=400)
-
-    if score is not None and accuracy is not None:
-        # Reject booleans explicitly — float(True) would otherwise record a play.
-        if isinstance(score, bool) or isinstance(accuracy, bool):
-            return JSONResponse({"error": "score/accuracy must be finite numbers"}, status_code=400)
-        # Reject NaN/Inf too: round(inf) raises OverflowError (→ 500), and a
-        # stored Inf/NaN later breaks JSON serialization of /api/stats reads.
-        try:
-            score = float(score)
-            accuracy = float(accuracy)
-            if not (math.isfinite(score) and math.isfinite(accuracy)):
-                raise ValueError("non-finite")
-            score = int(round(score))
-        except (TypeError, ValueError, OverflowError):
-            return JSONResponse({"error": "score/accuracy must be finite numbers"}, status_code=400)
-        # A huge-but-finite score passes isfinite() yet overflows SQLite's
-        # 64-bit INTEGER on bind (→ 500). Bound it to the int64 range.
-        if not (0 <= score < 2**63):
-            return JSONResponse({"error": "score out of range"}, status_code=400)
-        # accuracy is a 0..1 fraction (the recorder's contract); reject
-        # out-of-range values so they don't surface as >100% / negative in
-        # /api/stats/best and the badge UI.
-        if not (0 <= accuracy <= 1):
-            return JSONResponse({"error": "accuracy must be between 0 and 1"}, status_code=400)
-        # Validate the optional resume position in this branch too (the
-        # position-only branch below already rejects non-finite).
-        if last_pos is not None:
-            try:
-                last_pos = float(last_pos)
-                if not math.isfinite(last_pos):
-                    raise ValueError("non-finite")
-            except (TypeError, ValueError, OverflowError):
-                return JSONResponse({"error": "lastPlayPosition must be a finite number"}, status_code=400)
-        row = meta_db.record_session(filename, arrangement, score=score,
-                                     accuracy=accuracy, last_position=last_pos)
-        # Unified XP + streak side-effects — never let these drop the stat write.
-        progress = None
-        try:
-            from xp import xp_for_run
-            from datetime import date
-            meta_db.award_xp(xp_for_run(score))
-            meta_db.record_active_day(date.today().isoformat())
-            progress = meta_db.get_progress()
-        except Exception:
-            log.warning("stats side-effects (xp/streak) failed", exc_info=True)
-        # Progression engine (spec 010) — same never-drop-the-stat-write
-        # contract. Scored sessions are the server-derived `song_completed`
-        # authority (scored == note detection by construction); instrument is
-        # resolved from library arrangement metadata, after the XP award so
-        # db_earned goals see this run's Decibels.
-        progression_summary = None
-        try:
-            import progression as progression_mod
-            instrument = progression_mod.instrument_for_arrangement(
-                meta_db.arrangement_entry(filename, arrangement)
-            )
-            progression_summary = meta_db.record_progression_event(
-                "song_completed",
-                {
-                    "filename": filename,
-                    "instrument": instrument,
-                    "accuracy": accuracy,
-                    "score": score,
-                    "is_diagnostic": filename == _builtin_diagnostic_filename(),
-                },
-                _get_progression_content(),
-            )
-        except Exception:
-            log.warning("stats side-effects (progression) failed", exc_info=True)
-        return {"stats": row, "progress": progress, "progression": progression_summary}
-
-    # Position-only touch.
-    if last_pos is None:
-        return JSONResponse(
-            {"error": "provide score+accuracy (scored) or lastPlayPosition (resume)"},
-            status_code=400,
-        )
-    try:
-        pos = float(last_pos)
-        if not math.isfinite(pos):
-            raise ValueError("non-finite")
-        row = meta_db.touch_position(filename, arrangement, pos)
-    except (TypeError, ValueError, OverflowError):
-        return JSONResponse({"error": "lastPlayPosition must be a finite number"}, status_code=400)
-    # A resume session still counts as playing today: advance the streak (no XP —
-    # that's scoring-only) so a non-scored practice day keeps the streak alive,
-    # consistent with these sessions also surfacing in recent / continue.
-    progress = None
-    try:
-        from datetime import date
-        meta_db.record_active_day(date.today().isoformat())
-        progress = meta_db.get_progress()
-    except Exception:
-        log.warning("stats side-effects (streak) failed", exc_info=True)
-    return {"stats": row, "progress": progress}
-
-
-@app.get("/api/stats/recent")
-def api_recent_stats(limit: int = 12):
-    """Recently-played rows joined to song metadata for 'Jump back in'."""
-    from urllib.parse import quote
-    out = []
-    for r in meta_db.recent_stats(limit):
-        meta = meta_db.conn.execute(
-            "SELECT title, artist, tuning_name FROM songs WHERE filename = ?",
-            (r["filename"],),
-        ).fetchone()
-        title, artist, tuning_name = meta if meta else (None, None, None)
-        out.append({
-            **r,
-            "title": title or r["filename"],
-            "artist": artist or "",
-            "tuning_name": tuning_name or "",
-            "art_url": f"/api/song/{quote(r['filename'])}/art",
-        })
-    return out
-
-
-@app.get("/api/stats/best")
-def api_stats_best():
-    """{filename: best_accuracy} for all songs with a recorded best — one call
-    to badge the library grid (defined before the {filename} catch-all)."""
-    return meta_db.best_accuracy_map()
-
-
-@app.get("/api/stats/top")
-def api_top_stats(limit: int = 5):
-    """Top scored songs (best first), joined to song metadata, for the profile
-    'Your best scores' panel (defined before the {filename} catch-all)."""
-    from urllib.parse import quote
-    out = []
-    for r in meta_db.top_stats(limit):
-        meta = meta_db.conn.execute(
-            "SELECT title, artist, tuning_name FROM songs WHERE filename = ?",
-            (r["filename"],),
-        ).fetchone()
-        title, artist, tuning_name = meta if meta else (None, None, None)
-        out.append({
-            **r,
-            "title": title or r["filename"],
-            "artist": artist or "",
-            "tuning_name": tuning_name or "",
-            "art_url": f"/api/song/{quote(r['filename'])}/art",
-        })
-    return out
 
 
 @app.get("/api/library/practice-suggestions")
@@ -5232,9 +4653,6 @@ def api_practice_suggestions(limit: int = 8):
     return out
 
 
-@app.get("/api/stats/{filename:path}")
-def api_song_stats(filename: str):
-    return meta_db.get_song_stats(filename)
 
 
 # ── Playlists / custom covers (fee[dB]ack v0.3.0) ─────────────────────────────
@@ -5839,6 +5257,11 @@ def _running_version() -> str:
     return "unknown"
 
 
+# _running_version is defined below the import-top configure() calls, so publish
+# it here (configure is idempotent/additive) for routers/diagnostics.py.
+appstate.configure(running_version=_running_version)
+
+
 def _validate_server_config_types(cfg: dict) -> str | None:
     """Type-and-range gate for the server_config block of an import
     bundle, mirroring the per-key checks in `POST /api/settings`. The
@@ -6565,277 +5988,16 @@ def import_settings(bundle: dict):
     }
 
 
-# ── Diagnostic bundle export (feedBack#166) ──────────────────────────
-#
-# One-click "Export Diagnostics" in Settings produces a redacted zip
-# combining server logs, system info, hardware (CPU/GPU/RAM), plugin
-# inventory, and the browser-side console transcript + hardware probe.
-# The bundle format is specified in docs/diagnostics-bundle-spec.md.
-
-from fastapi import Body
-
-from diagnostics_bundle import build_bundle as _diag_build, preview_bundle as _diag_preview
-from diagnostics_hardware import collect as _diag_hardware
-
-
-def _diag_log_file() -> Path | None:
-    raw = os.environ.get("LOG_FILE", "").strip()
-    if not raw:
-        return None
-    return Path(raw)
-
-
-def _diag_plugins_roots() -> list[Path]:
-    """Return all plugin root directories for orphan scanning.
-
-    Includes both the built-in ``plugins/`` directory and
-    ``FEEDBACK_PLUGINS_DIR`` when set, so user-installed plugins and
-    orphans in the external dir are reflected in the bundle.
-    """
-    roots: list[Path] = []
-    user_dir = getenv_compat("FEEDBACK_PLUGINS_DIR", "").strip()
-    if user_dir:
-        p = Path(user_dir)
-        if p.is_dir():
-            roots.append(p)
-    builtin = Path(__file__).parent / "plugins"
-    if builtin not in roots:
-        roots.append(builtin)
-    return roots
-
-
-def _diag_coerce_bool(v, *, default: bool = True) -> bool:
-    """Coerce a request-side value to bool, accepting both JSON booleans and
-    string representations.
-
-    - Falsy strings: ``"false"``, ``"0"``, ``"no"``, ``""`` → ``False``
-    - ``None`` → *default*
-    - Everything else (including ``"true"``, ``"1"``) → ``True``
-    """
-    if v is None:
-        return default
-    if isinstance(v, bool):
-        return v
-    if isinstance(v, str):
-        return v.strip().lower() not in ("false", "0", "no", "")
-    return bool(v)
-
-
-def _diag_normalize_include(include: dict | None) -> dict:
-    """Coerce request-side flags to the booleans build_bundle expects.
-    Missing keys default to True so a bare {} request still produces
-    the full bundle.
-
-    Accepts both JSON booleans (``true``/``false``) and string
-    representations so callers that serialize flags as strings behave
-    consistently with the preview endpoint:
-    - Falsy strings: ``"false"``, ``"0"``, ``"no"``, ``""`` → ``False``
-    - Everything else (including ``"true"``, ``"1"``, ``"yes"``) → ``True``
-    """
-    keys = ("system", "hardware", "logs", "console", "plugins")
-    if not isinstance(include, dict):
-        return {k: True for k in keys}
-
-    return {k: _diag_coerce_bool(include.get(k), default=True) for k in keys}
-
-
-# Server-side caps on client-supplied payload sections.  diagnostics.js
-# enforces a 500-entry / ~250 KB ring buffer on the browser side; these
-# bounds give generous headroom while still preventing a crafted POST from
-# forcing the server to allocate arbitrarily large in-memory bundles.
-_DIAG_MAX_CONSOLE_ENTRIES = 1000          # hard cap: truncate silently
-_DIAG_MAX_CONSOLE_BYTES = 2 * 1024 * 1024  # 2 MB hard cap on total console list
-_DIAG_MAX_CLIENT_PAYLOAD_BYTES = 2 * 1024 * 1024   # 2 MB per dict section
-_DIAG_MAX_CONTRIBUTIONS_BYTES = 4 * 1024 * 1024    # 4 MB aggregate cap for contributions
-
-
-def _diag_cap_console(v) -> list | None:
-    """Return *v* if it is a list, truncated to _DIAG_MAX_CONSOLE_ENTRIES entries
-    and _DIAG_MAX_CONSOLE_BYTES total.  Entries are accumulated until either cap
-    is reached; no partial-entry splitting occurs."""
-    if not isinstance(v, list):
-        return None
-    result = v[:_DIAG_MAX_CONSOLE_ENTRIES]
-    # Also enforce a byte cap — the count cap alone does not bound memory when
-    # entries contain arbitrarily large strings.
-    try:
-        out = []
-        total = 0
-        for entry in result:
-            encoded = json.dumps(entry, separators=(",", ":")).encode("utf-8", errors="replace")
-            if total + len(encoded) > _DIAG_MAX_CONSOLE_BYTES:
-                break
-            out.append(entry)
-            total += len(encoded)
-        return out
-    except (TypeError, ValueError):
-        return None
-
-
-def _diag_cap_dict(v) -> dict | None:
-    """Return *v* if it is a dict whose JSON serialisation fits within
-    _DIAG_MAX_CLIENT_PAYLOAD_BYTES, otherwise return None."""
-    if not isinstance(v, dict):
-        return None
-    try:
-        encoded = json.dumps(v, separators=(",", ":")).encode("utf-8", errors="replace")
-    except (TypeError, ValueError) as e:
-        log.warning("diagnostics client payload is not JSON-serialisable, dropping: %s", e)
-        return None
-    if len(encoded) > _DIAG_MAX_CLIENT_PAYLOAD_BYTES:
-        return None
-    return v
-
-
-def _diag_cap_contributions(v, known_ids=None) -> dict | None:
-    """Apply per-plugin and aggregate size caps on client_contributions.
-
-    Unlike _diag_cap_dict(), which drops the whole dict when any plugin
-    exceeds the limit, this function caps each plugin independently so
-    one noisy plugin does not silence every other plugin's contribution.
-
-    Parameters
-    ----------
-    v:
-        The raw contributions dict from the POST payload.
-    known_ids:
-        When provided, contributions from plugins not in this set are
-        skipped *before* serialisation, preventing a malicious caller
-        from forcing the server to JSON-encode hundreds of near-limit
-        payloads that ``build_bundle()`` would later discard anyway.
-        ``None`` means "accept all plugin ids" (used in tests / preview).
-    """
-    if not isinstance(v, dict):
-        return None
-    result = {}
-    total_bytes = 0
-    for pid, contribution in v.items():
-        if not isinstance(pid, str):
-            continue
-        # Filter unknown plugin ids early — before serialising — so a
-        # crafted request cannot force large allocations for plugins that
-        # build_bundle() would drop.
-        if known_ids is not None and pid not in known_ids:
-            continue
-        try:
-            encoded = json.dumps(contribution, separators=(",", ":")).encode("utf-8", errors="replace")
-        except (TypeError, ValueError) as e:
-            log.warning(
-                "client_contributions[%r] is not JSON-serialisable, dropping: %s", pid, e
-            )
-            continue
-        if len(encoded) > _DIAG_MAX_CLIENT_PAYLOAD_BYTES:
-            log.warning(
-                "client_contributions[%r] exceeds %d bytes, dropping",
-                pid, _DIAG_MAX_CLIENT_PAYLOAD_BYTES,
-            )
-            continue
-        if total_bytes + len(encoded) > _DIAG_MAX_CONTRIBUTIONS_BYTES:
-            log.warning(
-                "client_contributions aggregate size limit (%d bytes) reached, "
-                "dropping remaining entries",
-                _DIAG_MAX_CONTRIBUTIONS_BYTES,
-            )
-            break
-        result[pid] = contribution
-        total_bytes += len(encoded)
-    return result or None
-
-
-@app.post("/api/diagnostics/export")
-def export_diagnostics(payload: dict = Body(default_factory=dict)):
-    """Build a diagnostic bundle and stream it back as a zip download.
-
-    The browser layers in `client_console`, `client_hardware`,
-    `client_ua`, and `local_storage` before posting; the server adds
-    server logs, hardware, plugin inventory, and packages everything
-    into a single zip.
-
-    Errors during plugin diagnostics callables are caught and logged
-    to the bundle's manifest `notes` rather than failing the export.
-    """
-    from plugins import LOADED_PLUGINS, PLUGINS_LOCK
-
-    redact = _diag_coerce_bool(payload.get("redact", True), default=True)
-    include = _diag_normalize_include(payload.get("include"))
-    client_console = _diag_cap_console(payload.get("client_console"))
-    client_hardware = _diag_cap_dict(payload.get("client_hardware"))
-    client_ua = _diag_cap_dict(payload.get("client_ua"))
-    local_storage = _diag_cap_dict(payload.get("local_storage"))
-    # Fetch the plugin list first so we can filter contributions to known
-    # plugin ids before serialising — prevents a crafted request from
-    # forcing large allocations for plugins build_bundle() would drop.
-    with PLUGINS_LOCK:
-        plugins_snapshot = list(LOADED_PLUGINS)
-    known_ids = {p.get("id") for p in plugins_snapshot if isinstance(p.get("id"), str)}
-    client_contributions = _diag_cap_contributions(
-        payload.get("client_contributions"), known_ids=known_ids
-    )
-
-    zip_bytes, filename, _manifest = _diag_build(
-        feedBack_version=_running_version(),
-        config_dir=CONFIG_DIR,
-        dlc_dir=_get_dlc_dir(),
-        log_file=_diag_log_file(),
-        loaded_plugins=plugins_snapshot,
-        include=include,
-        redact=redact,
-        client_console=client_console,
-        client_hardware=client_hardware,
-        client_ua=client_ua,
-        local_storage=local_storage,
-        client_contributions=client_contributions,
-        log=log,
-        plugins_root=_diag_plugins_roots(),
-    )
-    return Response(
-        content=zip_bytes,
-        media_type="application/zip",
-        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
-    )
-
-
-@app.get("/api/diagnostics/preview")
-def preview_diagnostics(
-    redact: bool = True,
-    system: bool = True,
-    hardware: bool = True,
-    logs: bool = True,
-    console: bool = True,
-    plugins: bool = True,
-):
-    """Return what `/api/diagnostics/export` would produce, minus the
-    actual file contents — file tree, sizes, schemas, redaction counts.
-    Lets the Settings UI show the user what's about to be sent."""
-    from plugins import LOADED_PLUGINS, PLUGINS_LOCK
-
-    include = {
-        "system": system,
-        "hardware": hardware,
-        "logs": logs,
-        "console": console,
-        "plugins": plugins,
-    }
-    with PLUGINS_LOCK:
-        plugins_snapshot = list(LOADED_PLUGINS)
-    return _diag_preview(
-        feedBack_version=_running_version(),
-        config_dir=CONFIG_DIR,
-        dlc_dir=_get_dlc_dir(),
-        log_file=_diag_log_file(),
-        loaded_plugins=plugins_snapshot,
-        include=include,
-        redact=redact,
-        log=log,
-        plugins_root=_diag_plugins_roots(),
-    )
-
-
-@app.get("/api/diagnostics/hardware")
-def diagnostics_hardware():
-    """Backend hardware probe (cross-platform). Reusable independently
-    of the bundle export — handy for "what's my GPU" plugin queries."""
-    return _diag_hardware()
+# ── Diagnostic bundle export (feedBack#166) → routers/diagnostics.py (R3) ────
+# The pure caps/normalisers are re-exported so existing server._diag_* /
+# server._DIAG_* tests keep resolving (none of them monkeypatch these).
+from routers.diagnostics import (  # noqa: E402  (re-export for test compatibility)
+    _diag_cap_console, _diag_cap_contributions, _diag_cap_dict,
+    _diag_coerce_bool, _diag_normalize_include,
+    _DIAG_MAX_CLIENT_PAYLOAD_BYTES, _DIAG_MAX_CONSOLE_BYTES,
+    _DIAG_MAX_CONSOLE_ENTRIES, _DIAG_MAX_CONTRIBUTIONS_BYTES,
+)
+app.include_router(diagnostics.router)
 
 
 # ── Plugin-provided routes are registered at startup via plugins/__init__.py ─

static/app.js

@@ -4877,6 +4877,15 @@ window.jucePlayer = jucePlayer;
     // value change (this runs on a 350ms poll — logging every tick would
     // flood the diagnostics buffer).
     let _loggedOutputType;
+    // [asio-diag] verbose diagnostics, gated on --debug (preload exposes
+    // audio.debugEnabled). Resolved once at install; until it resolves the
+    // flag stays false and verbose lines are skipped. Shared with the
+    // renderer-bus feeder below via window._asioDiagEnabled.
+    let _asioDiag = false;
+    if (typeof juceApi.debugEnabled === 'function') {
+        juceApi.debugEnabled().then((v) => { _asioDiag = !!v; }).catch(() => {});
+    }
+    window._asioDiagEnabled = () => _asioDiag;
     async function _outputIsExclusive() {
         if (typeof juceApi.getCurrentDevice !== 'function') {
             if (_loggedOutputType !== '<no-getCurrentDevice>') {
@@ -4892,6 +4901,15 @@ window.jucePlayer = jucePlayer;
             if (t !== _loggedOutputType) {
                 _loggedOutputType = t;
                 console.log('[feedpak-route] outputType=', JSON.stringify(t), '→ exclusive=', excl);
+                // [asio-diag] full device object on every type change — shows
+                // the exact strings the predicate saw (inputType vs outputType,
+                // device names, duplex), so a driver reporting a non-'ASIO'
+                // type name is visible in tester logs.
+                if (_asioDiag) {
+                    try {
+                        console.log('[asio-diag] getCurrentDevice=', JSON.stringify(dev));
+                    } catch (_) { /* circular/hostile object — skip */ }
+                }
             }
             return excl;
         } catch (e) {
@@ -5376,6 +5394,13 @@ window.jucePlayer = jucePlayer;
         if (typeof ctx.setSinkId !== 'function') throw new Error('setSinkId unsupported');
         await ctx.setSinkId(exclusive ? { type: 'none' } : '');
         if (ctx.state !== 'running') await ctx.resume().catch(() => {});
+        // [asio-diag] a context left on the default sink while the bus is
+        // engaged is exactly the "song on the wrong device" symptom — record
+        // every successful sink flip (failures throw and are logged upstream).
+        if (window._asioDiagEnabled?.()) {
+            console.log('[asio-diag] setSink:', exclusive ? 'null-sink' : 'default',
+                'state=', ctx.state, 'rate=', ctx.sampleRate);
+        }
     }
 
     async function _disengage() {
@@ -5446,6 +5471,24 @@ window.jucePlayer = jucePlayer;
                 else if (elementSong) want = 'element';
             }
 
+            // [asio-diag] full decision vector, change-gated (500ms poll —
+            // steady state must not flood the buffer). This is the feeder-side
+            // counterpart of the watcher's [feedpak-route] decision line: it
+            // shows WHY the bus did or didn't engage (exclusive predicate,
+            // stems graph presence, native transport ownership, element song).
+            if (window._asioDiagEnabled?.()) {
+                const d = 'running=' + running + ' exclusive=' + exclusive
+                    + ' stems=' + !!stems + ' songAudio=' + !!songAudio
+                    + ' juceMode=' + !!window._juceMode
+                    + ' elementSong=' + elementSong
+                    + ' want=' + want + ' mode=' + _mode;
+                if (d !== window._lastRendererBusDecision) {
+                    window._lastRendererBusDecision = d;
+                    console.log('[asio-diag] renderer-bus:', d);
+                }
+            }
+
+
             const stemsGraphChanged = _mode === 'stems' && stems !== _stemsGraph;
             if (want !== _mode || stemsGraphChanged) {
                 await _disengage();