refactor(server): extract MetadataDB into lib/metadata_db.py (R3)

Move-only. The library metadata cache -- the `MetadataDB` class (4,018 lines) plus the query helpers it owns (keyset paging cursors, the tuning grouping key, smart-arrangement naming, tag normalisation, the startup DB-restore swap) -- moves out of server.py into a flat `lib/` module. Every moved block is byte-identical to its server.py original; server.py is exactly origin/main minus the six cut ranges, minus the now-dead `import contextlib`, plus the import-back block and the constructor call site. server.py: 14,037 -> 9,705 lines. The one non-verbatim change is the seam that lets the class leave server.py: `MetadataDB.__init__` now takes `config_dir` explicitly instead of reading the module-level CONFIG_DIR, so `lib/metadata_db.py` does no IO at import (Principle V). The `meta_db` singleton stays in server.py, so `server.meta_db` (282 refs) and `server.app` (67 refs) resolve unchanged and no route moves. None of the 114 `monkeypatch.setattr(server, ...)` targets moved. Logging still goes through the `feedBack.server` logger, so log filters and caplog assertions resolve to the same logger object. `tests/test_settings_export_library_db.py` imports `_apply_pending_db_restore` from metadata_db (the test moves with its subject); no other test changed. Verified: pyflakes clean on the new module (zero undefined names, zero unused imports) and no new undefined name in server.py; pytest 2341 passed; node --test 1030 passed; eslint 0 errors; uvicorn boot smoke serves /api/version, /api/library, and all three migrated plugins' src/ module graphs (stems, studio, editor -> 200). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-09-12 10:08:32 +00:00 · 2026-07-10 14:08:11 +02:00
23 changed files with 1850 additions and 2871 deletions
@@ -7,65 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 ### Fixed
 - **The packaged desktop app could not start (`ModuleNotFoundError: No module named
  'appstate'`).** feedback-desktop's `scripts/bundle-slopsmith.sh` copies a *hardcoded
  list* of core files into the app bundle — `server.py`, `VERSION`, `lib/`, `data/`,
  `static/`, `plugins/__init__.py`. The root-level `appstate.py` and `routers/` added in
  R3 shipped correctly in Docker and passed every test, and were then silently dropped
  from the packaged app, which died at startup. Both now live under **`lib/`** — the one
  core directory the Dockerfile (`COPY lib/`), `docker-compose.yml`, and the desktop
  bundler (`cp -r lib`) all copy wholesale, and that all three put on `sys.path` (on
  Windows via the embeddable-Python `._pth`, where `PYTHONPATH` is ignored). This needs no
  change in feedback-desktop and no new release to take effect. Placing them there is also
  correct under Principle V: with the injection seam, `appstate.py` constructs nothing and
  does no import-time IO, and a route module only builds an `APIRouter`. The
  `Dockerfile` / `.dockerignore` / `docker-compose.yml` entries added for the root layout
  are reverted. New `tests/test_packaging.py` walks `server.py`'s module-level imports and
  fails if any first-party module resolves outside a directory the packagers copy, so the
  next root-level module can't ship broken.
 ### Added
 - **`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/` — 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
  where they used to be defined** — FastAPI matches routes in registration order, so the
  mount site preserves it. Verified: the full 143-route table (paths, methods, *and*
  order) is byte-for-byte identical to `main`. Bodies are verbatim; the only edits are
  the decorator receiver (`@app.get` → `@router.get`) and the singleton read
  (`audio_effect_mappings` → `appstate.audio_effect_mappings`, a module attribute
  resolved at call time). This proves the seam from #833 under a real consumer, including
  the second slot. The `_demo_mode_guard` middleware still blocks all four moved write
  routes with 403, and `Query(...)` validation still 422s — both checked against a running
  server. `server.py`: **9,445 → 9,386 lines**.
 - **`appstate.py` — the router seam (R3).** Route modules moving out of `server.py`
  need `meta_db` and friends but must not `import server`, or the import graph goes
  circular the moment `server` imports them back. So `server.py` keeps *constructing*
  its singletons and now **injects** them once — `appstate.configure(meta_db=…,
  audio_effect_mappings=…)` — and a router reads them back as module attributes at call
  time (`import appstate; appstate.meta_db.…`). This is the Python analogue of the
  frontend refactor's injected `configureX({…})` seams and of the plugin
  `setup(app, context)` contract: dependencies flow one way, `server → routers →
  appstate`. Two properties are load-bearing and pinned by `tests/test_appstate.py`:
  (1) `import appstate` constructs nothing and touches no disk, so the ~49 test fixtures
  that `sys.modules.pop("server")` + re-import (to rebuild `meta_db` under a patched
  `CONFIG_DIR`) keep working untouched — a singleton *owned* by `appstate` would survive
  that pop and go stale; (2) reads must be late-bound (`appstate.meta_db`, never
  `from appstate import meta_db`), since a `from` import freezes the binding and defeats
  both a later `configure()` and `monkeypatch.setattr` — the same read-only-binding trap
  as ES `import`. `configure()` rejects an unknown slot rather than silently creating a
  global nothing reads, and the suite asserts `server` actually calls it (a seam whose
  wiring can no-op undetected is worse than no seam). Lives at `lib/appstate.py`.
 ### Changed
 - **`AudioEffectsMappingDB` moved out of `server.py` into `lib/audio_effects_db.py`
  (R3, move-only).** The core-owned song/tone → provider routing index follows
  `MetadataDB` out of the host file, byte-identical apart from the same constructor
  seam (`__init__` takes `config_dir`; `audio_effect_mappings = AudioEffectsMappingDB(CONFIG_DIR)`),
  so the module does no IO at import. The singleton stays in `server.py`; no route,
  no test and no `monkeypatch.setattr(server, …)` target moves. `server.py`:
  **9,705 → 9,433 lines**.
 - **`MetadataDB` moved out of `server.py` into `lib/metadata_db.py` (R3, move-only).**
  The library metadata cache — the `MetadataDB` class (4,018 lines) plus the query
  helpers it owns (keyset paging cursors, the tuning grouping key, smart-arrangement
@@ -55,8 +55,7 @@ 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,909 — was 14,037; ratcheted by the R3 `MetadataDB` + `AudioEffectsMappingDB`
+(9,705 — was 14,037; ratcheted by the R3 `MetadataDB` extraction) ·
 extractions and six `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`
@@ -1,102 +0,0 @@
 """Shared application state — the seam that lets route modules reach core
 singletons without importing ``server``.
 ``server.py`` is the host: it owns the FastAPI ``app``, constructs the DB
 singletons, and runs the lifecycle. As routes move out into ``routers/`` (R3),
 those modules need ``meta_db`` and friends — but they must not ``import
 server``, or the import graph goes circular the moment ``server`` imports them
 back.
 So ``server`` **injects** its singletons here once, at the point it builds them::
    # server.py
    meta_db = MetadataDB(CONFIG_DIR)
    appstate.configure(meta_db=meta_db, ...)
 and a router reads them back as **module attributes, at call time**::
    # routers/artists.py
    import appstate
    @router.get("/api/artist/{name}/page")
    def artist_page(name):
        return appstate.meta_db.artist_page(name)
 This is the Python analogue of the injected `configureX({...})` seams the
 frontend refactor uses (stems' ``configureStreaming``, studio's
 ``configureAudioGraph``, the editor's ``src/host.js``), and of the plugin
 ``setup(app, context)`` contract in Principle III: dependencies flow one way,
 ``server -> routers -> appstate``, and nothing imports back up.
 Two properties this shape buys, both load-bearing:
 * **``import appstate`` performs no IO and constructs nothing.** ``server``
  still owns construction, so the ~49 test fixtures that do
  ``sys.modules.pop("server")`` + re-import (to rebuild ``meta_db`` under a
  patched ``CONFIG_DIR``) keep working untouched — a singleton *owned* here
  would survive that pop and go stale.
 * **Reads are late-bound.** Routers must use ``appstate.meta_db``, never
  ``from appstate import meta_db`` — a ``from`` import freezes the binding at
  its current value, so a later ``configure()`` (or a
  ``monkeypatch.setattr(appstate, "meta_db", fake)``) would not reach the
  router. This is the same read-only-binding trap as ES ``import``.
 Defaults are ``None`` on purpose: they are inert but *type-honest*, so a router
 that runs before ``configure()`` fails loudly on ``NoneType`` instead of
 quietly operating on a stand-in.
 Slots are added here only when a router actually needs one — this is a seam,
 not a grab-bag for everything in ``server.py``.
 **Why this lives in ``lib/`` and not the repo root.** Because it constructs
 nothing and does no import-time IO, it satisfies Principle V's rule for ``lib/``
 modules — and ``lib/`` is the only core directory every packaging path already
 copies: the Dockerfile (``COPY lib/``), ``docker-compose.yml``, and
 feedback-desktop's ``bundle-slopsmith.sh`` (``cp -r lib``). All three also put
 both the bundle root and ``lib/`` on ``sys.path``. A root-level module ships in
 Docker but is silently dropped from the packaged desktop app, whose bundler
 copies a hardcoded file list — that regression is what moved this file here.
 """
 # The singletons routers may read. Every name here must also be a `_SLOTS` key.
 meta_db = None
 audio_effect_mappings = None
 # Config paths. server.py derives these from the environment (fresh on every
 # import, so the ~49 pop-and-reimport fixtures keep working) and injects them
 # here. Routers read them as `appstate.config_dir` etc. — a module attribute at
 # call time. NOTE: config_dir/dlc_dir are env-derived, so a `setenv`+reimport
 # test reconfigures them for free; STATIC_DIR/SLOPPAK_CACHE_DIR are patched via
 # `setattr(server, …)` in a few tests, so those slots (when added) need their
 # tests retargeted to appstate in the same PR.
 config_dir = None
 dlc_dir = None          # the DLC_DIR env value as a Path (Path("") if unset)
 dlc_dir_env = None      # the raw DLC_DIR env string, "" if unset — distinguishes
                        # "unset" from Path("")→"." (see dlc_paths._get_dlc_dir)
 # Cache/asset dirs. static_dir + sloppak_cache_dir are patched via
 # `setattr(server, …)` in a few tests, so a router reading them here needs those
 # setattr sites retargeted to `setattr(appstate, …)` in the same PR (ws_highway
 # retargets the 3 test_highway_ws_* SLOPPAK sites). config_dir-derived dirs are
 # reconfigured for free on a setenv+reimport.
 static_dir = None
 sloppak_cache_dir = None
 audio_cache_dir = None
 _SLOTS = frozenset({
    "meta_db", "audio_effect_mappings",
    "config_dir", "dlc_dir", "dlc_dir_env",
    "static_dir", "sloppak_cache_dir", "audio_cache_dir",
 })
 def configure(**kwargs) -> None:
    """Publish `server`'s singletons into this module. Called once per
    `server` import (and again on re-import), so it must be idempotent."""
    unknown = set(kwargs) - _SLOTS
    if unknown:
        raise TypeError(
            f"appstate.configure() got unknown slot(s): {sorted(unknown)}. "
            f"Known slots: {sorted(_SLOTS)}. Add the name to _SLOTS if a router "
            f"genuinely needs it."
        )
    globals().update(kwargs)
@@ -1,287 +0,0 @@
 """Core-owned song/tone -> audio-effect-provider mapping index.
 Extracted verbatim from ``server.py`` (R3). ``server.py`` still owns the
 ``audio_effect_mappings`` singleton; this module only supplies the class, so
 nothing here touches config paths at import time — the caller passes
 ``config_dir`` in.
 """
 import json
 import sqlite3
 import threading
 from pathlib import Path
 class AudioEffectsMappingDB:
    """Core-owned public song/tone -> provider mapping index.
    Providers own the preset/chain rows addressed by provider_ref. Core owns
    the cross-provider routing index and the active mapping per song/tone.
    """
    def __init__(self, config_dir: Path):
        config_dir.mkdir(parents=True, exist_ok=True)
        self.db_path = str(config_dir / "audio_effects.db")
        self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
        self.conn.execute("PRAGMA journal_mode=WAL")
        self.conn.execute("PRAGMA foreign_keys=ON")
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS audio_effect_mappings (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                song_key TEXT NOT NULL,
                filename TEXT NOT NULL DEFAULT '',
                tone_key TEXT NOT NULL,
                provider_id TEXT NOT NULL,
                provider_ref TEXT NOT NULL,
                label TEXT NOT NULL DEFAULT '',
                source TEXT NOT NULL DEFAULT 'manual',
                metadata_json TEXT NOT NULL DEFAULT '{}',
                created_at TEXT NOT NULL DEFAULT (datetime('now')),
                updated_at TEXT NOT NULL DEFAULT (datetime('now')),
                UNIQUE(song_key, tone_key, provider_id)
            )
        """)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS audio_effect_active_mappings (
                song_key TEXT NOT NULL,
                tone_key TEXT NOT NULL,
                mapping_id INTEGER NOT NULL,
                updated_at TEXT NOT NULL DEFAULT (datetime('now')),
                PRIMARY KEY (song_key, tone_key),
                FOREIGN KEY (mapping_id) REFERENCES audio_effect_mappings(id) ON DELETE CASCADE
            )
        """)
        self.conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_audio_effect_mappings_provider "
            "ON audio_effect_mappings(provider_id)"
        )
        self.conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_audio_effect_mappings_filename "
            "ON audio_effect_mappings(filename)"
        )
        self.conn.commit()
        self._lock = threading.Lock()
    @staticmethod
    def _text(value, *, field: str, limit: int, allow_empty: bool = False) -> str:
        if value is None:
            text = ""
        elif not isinstance(value, str):
            raise ValueError(f"{field} must be a string")
        else:
            text = value.strip()
        if not text and not allow_empty:
            raise ValueError(f"{field} is required")
        if len(text) > limit:
            raise ValueError(f"{field} is too long")
        return text
    @staticmethod
    def _mapping_id(value) -> int | None:
        # Bind only values SQLite can store as an INTEGER; an out-of-range id is a
        # clean miss (404), not a 500 at bind time.
        if isinstance(value, int) and not isinstance(value, bool) and -(2 ** 63) <= value < 2 ** 63:
            return value
        return None
    @staticmethod
    def _field(data: dict, *keys):
        # Select the first present snake/camel alias by key, not by truthiness, so a
        # falsey non-string value (false/0) still reaches _text() and is rejected
        # instead of being silently swallowed by an `or` chain.
        for key in keys:
            if key in data:
                return data[key]
        return None
    @staticmethod
    def _metadata(value) -> str:
        if value is None:
            return "{}"
        if not isinstance(value, dict):
            raise ValueError("metadata must be an object")
        encoded = json.dumps(value, ensure_ascii=True, sort_keys=True)
        if len(encoded) > 8192:
            raise ValueError("metadata is too large")
        return encoded
    @staticmethod
    def _row(row) -> dict | None:
        if row is None:
            return None
        metadata = {}
        try:
            metadata = json.loads(row[8]) if row[8] else {}
        except Exception:
            metadata = {}
        return {
            "id": int(row[0]),
            "song_key": row[1],
            "filename": row[2] or "",
            "tone_key": row[3],
            "provider_id": row[4],
            "provider_ref": row[5],
            "label": row[6] or "",
            "source": row[7] or "manual",
            "metadata": metadata if isinstance(metadata, dict) else {},
            "created_at": row[9] or "",
            "updated_at": row[10] or "",
            "active": bool(row[11]),
        }
    def _select_sql(self) -> str:
        return """
            SELECT m.id, m.song_key, m.filename, m.tone_key, m.provider_id,
                   m.provider_ref, m.label, m.source, m.metadata_json,
                   m.created_at, m.updated_at,
                   CASE WHEN a.mapping_id IS NULL THEN 0 ELSE 1 END AS active
            FROM audio_effect_mappings m
            LEFT JOIN audio_effect_active_mappings a
              ON a.song_key = m.song_key AND a.tone_key = m.tone_key AND a.mapping_id = m.id
        """
    def list(self, *, song_key: str = "", filename: str = "", tone_key: str = "", provider_id: str = "") -> list[dict]:
        clauses: list[str] = []
        params: list[str] = []
        song_key = self._text(song_key, field="song_key", limit=240, allow_empty=True)
        filename = self._text(filename, field="filename", limit=500, allow_empty=True)
        tone_key = self._text(tone_key, field="tone_key", limit=160, allow_empty=True)
        provider_id = self._text(provider_id, field="provider_id", limit=96, allow_empty=True)
        if song_key and filename:
            clauses.append("(m.song_key = ? OR m.filename = ?)")
            params.extend([song_key, filename])
        elif song_key:
            clauses.append("m.song_key = ?")
            params.append(song_key)
        elif filename:
            clauses.append("(m.song_key = ? OR m.filename = ?)")
            params.extend([filename, filename])
        if tone_key:
            clauses.append("m.tone_key = ?")
            params.append(tone_key)
        if provider_id:
            clauses.append("m.provider_id = ?")
            params.append(provider_id)
        sql = self._select_sql()
        if clauses:
            sql += " WHERE " + " AND ".join(clauses)
        sql += " ORDER BY m.song_key COLLATE NOCASE, m.tone_key COLLATE NOCASE, m.provider_id COLLATE NOCASE"
        with self._lock:
            rows = self.conn.execute(sql, params).fetchall()
        return [self._row(row) for row in rows]
    def get(self, mapping_id: int) -> dict | None:
        mapping_id = self._mapping_id(mapping_id)
        if mapping_id is None:
            return None
        with self._lock:
            row = self.conn.execute(self._select_sql() + " WHERE m.id = ?", (mapping_id,)).fetchone()
        return self._row(row)
    def upsert(self, data: dict) -> dict:
        if not isinstance(data, dict):
            raise ValueError("mapping body must be an object")
        filename = self._text(data.get("filename", ""), field="filename", limit=500, allow_empty=True)
        song_key_raw = self._field(data, "song_key", "songKey")
        if song_key_raw is None or song_key_raw == "":
            song_key_raw = filename
        song_key = self._text(song_key_raw, field="song_key", limit=240)
        tone_key = self._text(self._field(data, "tone_key", "toneKey"), field="tone_key", limit=160, allow_empty=True)
        provider_id = self._text(self._field(data, "provider_id", "providerId"), field="provider_id", limit=96)
        provider_ref = self._text(self._field(data, "provider_ref", "providerRef"), field="provider_ref", limit=240)
        label = self._text(data.get("label", ""), field="label", limit=160, allow_empty=True)
        source = self._text(data.get("source", "manual"), field="source", limit=40, allow_empty=True) or "manual"
        metadata_json = self._metadata(data.get("metadata", {}))
        with self._lock:
            self.conn.execute(
                """
                INSERT INTO audio_effect_mappings
                    (song_key, filename, tone_key, provider_id, provider_ref, label, source, metadata_json, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))
                ON CONFLICT(song_key, tone_key, provider_id) DO UPDATE SET
                    -- Only overwrite filename when a non-empty one was supplied; an
                    -- omitted/empty filename must preserve the stored value (it's an
                    -- alternate lookup key for list(..., filename=...)).
                    filename=CASE WHEN excluded.filename <> '' THEN excluded.filename ELSE audio_effect_mappings.filename END,
                    provider_ref=excluded.provider_ref,
                    label=excluded.label,
                    source=excluded.source,
                    metadata_json=excluded.metadata_json,
                    updated_at=datetime('now')
                """,
                (song_key, filename, tone_key, provider_id, provider_ref, label, source, metadata_json),
            )
            row = self.conn.execute(
                "SELECT id FROM audio_effect_mappings WHERE song_key = ? AND tone_key = ? AND provider_id = ?",
                (song_key, tone_key, provider_id),
            ).fetchone()
            if row is None:
                raise ValueError("failed to create audio-effects mapping")
            mapping_id = int(row[0])
            if data.get("active") is True:
                self.conn.execute(
                    """
                    INSERT INTO audio_effect_active_mappings (song_key, tone_key, mapping_id, updated_at)
                    VALUES (?, ?, ?, datetime('now'))
                    ON CONFLICT(song_key, tone_key) DO UPDATE SET
                        mapping_id=excluded.mapping_id,
                        updated_at=datetime('now')
                    """,
                    (song_key, tone_key, mapping_id),
                )
            self.conn.commit()
        return self.get(mapping_id)
    def delete(self, mapping_id: int, *, provider_id: str = "") -> bool:
        mapping_id = self._mapping_id(mapping_id)
        if mapping_id is None:
            return False
        provider_id = self._text(provider_id, field="provider_id", limit=96, allow_empty=True)
        with self._lock:
            if provider_id:
                cur = self.conn.execute(
                    "DELETE FROM audio_effect_mappings WHERE id = ? AND provider_id = ?",
                    (mapping_id, provider_id),
                )
            else:
                cur = self.conn.execute("DELETE FROM audio_effect_mappings WHERE id = ?", (mapping_id,))
            self.conn.commit()
            return cur.rowcount > 0
    def activate(self, mapping_id: int, *, provider_id: str = "") -> dict | None:
        mapping_id = self._mapping_id(mapping_id)
        if mapping_id is None:
            return None
        provider_id = self._text(provider_id, field="provider_id", limit=96, allow_empty=True)
        with self._lock:
            row = self.conn.execute(
                self._select_sql() + " WHERE m.id = ?",
                (mapping_id,),
            ).fetchone()
            mapping = self._row(row)
            if not mapping or (provider_id and mapping["provider_id"] != provider_id):
                return None
            self.conn.execute(
                """
                INSERT INTO audio_effect_active_mappings (song_key, tone_key, mapping_id, updated_at)
                VALUES (?, ?, ?, datetime('now'))
                ON CONFLICT(song_key, tone_key) DO UPDATE SET
                    mapping_id=excluded.mapping_id,
                    updated_at=datetime('now')
                """,
                (mapping["song_key"], mapping["tone_key"], mapping_id),
            )
            self.conn.commit()
            selected = self.conn.execute(self._select_sql() + " WHERE m.id = ?", (mapping_id,)).fetchone()
        return self._row(selected)
    def clear_active(self, *, song_key: str, tone_key: str) -> bool:
        song_key = self._text(song_key, field="song_key", limit=240)
        tone_key = self._text(tone_key, field="tone_key", limit=160, allow_empty=True)
        with self._lock:
            cur = self.conn.execute(
                "DELETE FROM audio_effect_active_mappings WHERE song_key = ? AND tone_key = ?",
                (song_key, tone_key),
            )
            self.conn.commit()
            return cur.rowcount > 0
@@ -1,99 +0,0 @@
 """DLC library path resolution — where the song files live, plus safe containment.
 Extracted from ``server.py`` (R3). ``_resolve_dlc_path`` is pure and moved
 verbatim. ``_get_dlc_dir`` reads the env-derived paths through the ``appstate``
 seam (``server.py`` configures ``dlc_dir``/``dlc_dir_env``/``config_dir`` at
 import, fresh on every re-import), so this module does no import-time IO and the
 pop-and-reimport fixtures keep working. ``server.py`` re-exports both names, so
 existing ``server._get_dlc_dir`` / ``server._resolve_dlc_path`` references
 (tests, other handlers) resolve unchanged.
 """
 import json
 import os
 from pathlib import Path
 import appstate
 def _get_dlc_dir(cfg: dict | None = None) -> Path | None:
    # Only consider DLC_DIR if the env var was non-empty. `Path("")` collapses
    # to `.` and reports `.is_dir() == True`, which would silently shadow the
    # config.json fallback. Checking the raw env string preserves
    # `DLC_DIR=.` as a valid opt-in for cwd while keeping unset/empty out.
    if appstate.dlc_dir_env and appstate.dlc_dir.is_dir():
        return appstate.dlc_dir
    if cfg is None:
        config_file = appstate.config_dir / "config.json"
        if config_file.exists():
            try:
                cfg = json.loads(config_file.read_text(encoding="utf-8"))
            except Exception:
                pass
    if isinstance(cfg, dict):
        raw = str(cfg.get("dlc_dir", "")).strip()
        if raw:
            p = Path(raw)
            if p.is_dir():
                return p
    return None
 def _resolve_dlc_path(dlc: Path, filename: str) -> Path | None:
    """Resolve `filename` under DLC_DIR and refuse anything that escapes.
    `filename` arrives from `:path` route params and can contain `..`
    segments. The Sloppak and archive paths happen to fail safely later
    because their loaders raise on missing/invalid files, but loose-
    folder format detection (`is_loose_song`) globs and parses XML on
    disk first, which lets a crafted path trigger filesystem reads
    outside DLC_DIR before any guard fires. Centralise the containment
    check so every filename-bound handler validates before touching the
    filesystem.
    Containment here is LEXICAL (normalize `.`/`..` WITHOUT following
    symlinks), not `safe_join`'s `.resolve()`-based check — because users
    commonly mount their song library through a directory JUNCTION/symlink
    (a library shared across app installs; the desktop app's own mounts).
    `.resolve()` follows that junction to its real target, sees it sits
    outside DLC_DIR, and wrongly rejects every song reached through it — the
    scanner's `rglob` indexes those songs, but art/load then 403/404s (broken
    covers, unplayable songs). Lexical normalization still rejects the only
    escapes a `:path` filename can express — `..` traversal and absolute
    paths — which the traversal tests pin. `safe_join` stays strict (it is
    the zip-slip / plugin-asset guard, where following a symlink out IS the
    defense); the loose-folder art handler keeps its own per-file symlink
    re-check for defence-in-depth.
    Returns the validated Path (not necessarily link-resolved), or None if
    the filename is empty, contains a NUL, or escapes the DLC root.
    """
    if not filename:
        return None
    # Backslashes → forward slashes so a Windows-style `..\\x` traversal is
    # rejected identically on POSIX (mirrors safe_join's normalisation).
    safe = filename.replace("\\", "/")
    if "\x00" in safe:
        return None
    # Reject drive-letter / absolute paths in BOTH conventions. A POSIX "/x" is
    # caught by the containment check below (the `/` operator discards `root`),
    # but a Windows drive-absolute "C:/x" is treated as a relative "C:" dir on
    # POSIX and would otherwise slip in as `<root>/C:/x` — so the contract must
    # hold cross-platform (a shared library is reached from either OS).
    from pathlib import PurePosixPath, PureWindowsPath
    if (PurePosixPath(safe).is_absolute()
            or PureWindowsPath(safe).is_absolute()
            or PureWindowsPath(safe).drive):
        return None
    try:
        root = dlc.resolve()
        # normpath collapses `.`/`..`/duplicate separators purely lexically —
        # it never touches the filesystem, so an in-library junction component
        # is preserved (allowed) while `..`/absolute segments still escape and
        # get caught by the containment check below.
        candidate = Path(os.path.normpath(root / safe))
        if not candidate.is_relative_to(root):
            return None
    except (ValueError, OSError):
        return None
    return candidate
@@ -1,13 +0,0 @@
 """Request-field coercion helpers shared by the raw-`dict` POST handlers.
 Extracted verbatim from ``server.py`` (R3). Pure — no IO, no globals — so it
 imports cleanly from both ``server`` and any ``routers/`` module.
 """
 def _clean_str(value) -> str:
    """Trim a request field to a string; non-strings (or missing) → ''.
    Lets the raw-`dict` POST handlers treat wrong-typed JSON (an int/list/etc.
    where a string was expected) as "empty" and answer 400, instead of raising
    AttributeError/TypeError → 500 on a later .strip()/`in`."""
    return value.strip() if isinstance(value, str) else ""
@@ -1,31 +0,0 @@
 """FastAPI route modules extracted from ``server.py`` (R3).
 Each module here exposes a module-level ``router`` (a ``fastapi.APIRouter``)
 that ``server.py`` mounts with ``app.include_router(...)`` at the point in the
 file where those routes used to be defined — FastAPI matches routes in
 registration order, so keeping the mount site preserves it.
 **Routers must never ``import server``.** They reach core singletons through
 the injected seam instead::
    import appstate
    @router.get("/api/thing")
    def get_thing():
        return appstate.meta_db.thing()
 and always as a **module attribute, at call time** — never
 ``from appstate import meta_db``, which freezes the binding and defeats both a
 later ``appstate.configure()`` and ``monkeypatch.setattr``. See ``appstate.py``.
 Dependencies flow one way: ``server -> routers -> appstate``.
 **Why this lives under ``lib/``.** ``lib/`` is the only core directory every
 packaging path already copies wholesale — the Dockerfile (``COPY lib/``),
 ``docker-compose.yml``, and feedback-desktop's ``bundle-slopsmith.sh``
 (``cp -r lib``) — and all three put it on ``sys.path``. A root-level package
 ships in Docker but is silently dropped from the packaged desktop app, whose
 bundler copies a hardcoded file list. Route modules import nothing at module
 scope beyond FastAPI and ``appstate``, so they do no import-time IO and satisfy
 Principle V's rule for ``lib/``.
 """
@@ -1,68 +0,0 @@
 """Artist aliases / Tidy-up (P4) — canonicalize messy artist tags at DISPLAY
 ("ACDC" -> "AC/DC") without touching feedpak files or the scanner-derived
 songs.artist. All DB-only.
 Extracted verbatim from ``server.py`` (R3); only the decorator receiver
 (``@app`` -> ``@router``) and the singleton read (``meta_db`` ->
 ``appstate.meta_db``) changed. The read stays a module attribute so a re-imported
 ``server`` re-publishes a fresh DB into the seam — see ``appstate.py``.
 """
 from fastapi import APIRouter
 from fastapi.responses import JSONResponse
 import appstate
 router = APIRouter()
@router.get("/api/artist-aliases")
 def list_artist_aliases():
    """Existing raw→canonical overrides (the Tidy-up 'current merges' list)."""
    return {"aliases": appstate.meta_db.list_artist_aliases()}
@router.get("/api/artists/raw")
 def list_raw_artists(limit: int = 2000):
    """Distinct RAW artist names + song counts + current canonical — the Tidy-up
    picker (you merge raw variants into one canonical)."""
    return {"artists": appstate.meta_db.raw_artists(limit)}
@router.post("/api/artist-aliases")
 def set_artist_alias(data: dict):
    """Upsert one override: {raw_name, canonical_name, mb_artist_id?}. A self-alias
    (raw == canonical) clears the row instead (un-merge)."""
    raw = (data.get("raw_name") or "").strip()
    canon = (data.get("canonical_name") or "").strip()
    if not raw or not canon:
        return JSONResponse({"error": "raw_name and canonical_name are required"}, 400)
    result = appstate.meta_db.set_artist_alias(raw, canon, (data.get("mb_artist_id") or None))
    if not result.get("ok"):
        # Would form a cycle (raw → … → raw) — refuse rather than corrupt the chain.
        return JSONResponse(
            {"error": "alias would create a cycle", "raw_name": raw, "canonical_name": canon},
            409)
    return {"ok": True, "raw_name": raw, "canonical_name": result.get("canonical_name", canon)}
@router.post("/api/artist-aliases/merge")
 def merge_artist_aliases(data: dict):
    """Merge several raw artist variants into one canonical:
    {raw_names: [...], canonical_name}. The canonical's own self-alias is skipped.
    Returns {merged: N}."""
    canon = (data.get("canonical_name") or "").strip()
    raws = data.get("raw_names")
    if not canon:
        return JSONResponse({"error": "canonical_name is required"}, 400)
    if not isinstance(raws, list) or not raws:
        return JSONResponse({"error": "raw_names must be a non-empty array"}, 400)
    n = appstate.meta_db.merge_artists(raws, canon)
    return {"merged": n, "canonical_name": canon}
@router.delete("/api/artist-aliases/{raw_name:path}")
 def delete_artist_alias(raw_name: str):
    """Remove one override so that raw artist stands on its own again."""
    appstate.meta_db.remove_artist_alias(raw_name)
    return {"ok": True}
@@ -1,80 +0,0 @@
 """Audio-effects mapping API — the core-owned song/tone -> provider routing index.
 Extracted verbatim from ``server.py`` (R3); only the decorator receiver
 (``@app`` -> ``@router``) and the singleton read (``audio_effect_mappings`` ->
 ``appstate.audio_effect_mappings``) changed. The read must stay a module
 attribute so a re-imported ``server`` re-publishes a fresh DB into the seam and
 `monkeypatch.setattr` reaches this module — see ``appstate.py``.
 """
 from fastapi import APIRouter, Body, Query
 from fastapi.responses import JSONResponse
 import appstate
 router = APIRouter()
 def _audio_effects_error(exc: Exception):
    return JSONResponse({"error": str(exc)}, status_code=400)
@router.get("/api/audio-effects/mappings")
 def list_audio_effect_mappings(
    song_key: str = Query(""),
    filename: str = Query(""),
    tone_key: str = Query(""),
    provider_id: str = Query(""),
 ):
    try:
        return {
            "mappings": appstate.audio_effect_mappings.list(
                song_key=song_key,
                filename=filename,
                tone_key=tone_key,
                provider_id=provider_id,
            )
        }
    except ValueError as exc:
        return _audio_effects_error(exc)
@router.post("/api/audio-effects/mappings")
 def upsert_audio_effect_mapping(data: dict = Body(...)):
    try:
        mapping = appstate.audio_effect_mappings.upsert(data)
    except ValueError as exc:
        return _audio_effects_error(exc)
    return {"ok": True, "mapping": mapping}
@router.delete("/api/audio-effects/mappings/{mapping_id}")
 def delete_audio_effect_mapping(mapping_id: int, provider_id: str = Query("")):
    try:
        deleted = appstate.audio_effect_mappings.delete(mapping_id, provider_id=provider_id)
    except ValueError as exc:
        return _audio_effects_error(exc)
    if not deleted:
        return JSONResponse({"error": "mapping not found"}, status_code=404)
    return {"ok": True}
@router.post("/api/audio-effects/mappings/{mapping_id}/activate")
 def activate_audio_effect_mapping(mapping_id: int, data: dict = Body(default_factory=dict)):
    try:
        provider_id = data.get("provider_id") if "provider_id" in data else data.get("providerId")
        mapping = appstate.audio_effect_mappings.activate(mapping_id, provider_id="" if provider_id is None else provider_id)
    except ValueError as exc:
        return _audio_effects_error(exc)
    if not mapping:
        return JSONResponse({"error": "mapping not found"}, status_code=404)
    return {"ok": True, "mapping": mapping}
@router.delete("/api/audio-effects/active-mapping")
 def clear_audio_effect_active_mapping(song_key: str = Query(...), tone_key: str = Query("")):
    try:
        cleared = appstate.audio_effect_mappings.clear_active(song_key=song_key, tone_key=tone_key)
    except ValueError as exc:
        return _audio_effects_error(exc)
    return {"ok": True, "cleared": cleared}
@@ -1,115 +0,0 @@
 """Chart-level endpoints — split/unsplit a chart from its work, resolve work
 membership, and the context-menu "Get info" file inspector.
 Extracted verbatim from ``server.py`` (R3); edits: ``@app`` -> ``@router``,
 ``meta_db`` -> ``appstate.meta_db``. DLC path resolution comes from
 ``dlc_paths``; sloppak/loose detection from the shared lib modules.
 """
 from fastapi import APIRouter, HTTPException
 import appstate
 from dlc_paths import _get_dlc_dir, _resolve_dlc_path
 import sloppak as sloppak_mod
 import loosefolder as loosefolder_mod
 router = APIRouter()
@router.post("/api/chart/{filename:path}/split")
 def api_split_chart(filename: str):
    """'These aren't the same song' — split this chart out as its own singleton
    work. Under /api/chart (NOT /api/song) so the DELETE /api/song/{path}
    catch-all can't shadow it."""
    key = appstate.meta_db._canonical_song_filename(filename)
    appstate.meta_db.split_chart(key)
    return {"ok": True, "filename": key}
@router.post("/api/chart/{filename:path}/unsplit")
 def api_unsplit_chart(filename: str):
    """Undo a split — rejoin the chart to its work."""
    key = appstate.meta_db._canonical_song_filename(filename)
    appstate.meta_db.unsplit_chart(key)
    return {"ok": True, "filename": key}
@router.get("/api/chart/{filename:path}/work")
 def api_get_chart_work(filename: str):
    """Resolve a chart's work membership: {work_key, chart_count}. For openers
    on rows that came from an ungrouped query (the tree view) — grouped grid
    rows already carry both fields inline."""
    return appstate.meta_db.chart_work(filename)
@router.get("/api/chart/{filename:path}/fileinfo")
 def api_chart_fileinfo(filename: str):
    """The context menu's "Get info": where the file lives + what the pack
    contains. Under /api/chart — the GET /api/song/{path} catch-all would
    swallow a /api/song/…/fileinfo suffix. Read-only; demo-mode blocks it
    because it exposes filesystem paths."""
    dlc = _get_dlc_dir()
    if not dlc:
        raise HTTPException(status_code=404, detail="not configured")
    p = _resolve_dlc_path(dlc, filename)
    if p is None:
        raise HTTPException(status_code=403, detail="forbidden")
    if not p.exists():
        raise HTTPException(status_code=404, detail="not found")
    # Restrict to actual charts — sloppak or loose song. Without this the route
    # would stat ANY file the user happens to keep under DLC_DIR (e.g. notes),
    # leaking its path/size; the app only recognises these two song formats.
    is_pak = sloppak_mod.is_sloppak(p)
    is_loose = loosefolder_mod.is_loose_song(p)
    if not (is_pak or is_loose):
        raise HTTPException(status_code=404, detail="not a chart")
    st = p.stat()
    info = {
        "filename": filename,
        "path": str(p),
        "folder": str(p.parent),
        "format": "sloppak" if is_pak else "loose",
        # Directory-form songs report the tree's total (covers loose folders
        # and dir-form paks); zip-form paks report the archive size. Symlinked
        # entries are skipped so a link inside the folder can't pull in — or
        # leak the size of — a file outside it.
        "size": (st.st_size if p.is_file()
                 else sum(f.stat().st_size for f in p.rglob("*")
                          if f.is_file() and not f.is_symlink())),
        "mtime": st.st_mtime,
    }
    if is_pak:
        try:
            m = sloppak_mod.load_manifest(p) or {}
        except Exception:
            m = {}
        arrs = [str(a.get("name", a.get("id", ""))) for a in (m.get("arrangements") or [])
                if isinstance(a, dict)]
        stems = [str(s.get("id", "")) for s in (m.get("stems") or []) if isinstance(s, dict)]
        try:
            has_cover = sloppak_mod.read_cover_bytes(p, m) is not None
        except Exception:
            has_cover = False
        # The optional identity/catalog keys, listed only when present — the
        # Get-info panel's "what this pack carries vs what's missing" readout.
        identity = {k: m.get(k) for k in
                    ("mbid", "isrc", "genres", "track", "disc", "album_artist",
                     "feedpak_version", "language")
                    if m.get(k) not in (None, "", [])}
        info["manifest"] = {
            "title": str(m.get("title", "")), "artist": str(m.get("artist", "")),
            "album": str(m.get("album", "")), "year": str(m.get("year", "") or ""),
            "arrangements": arrs, "stems": stems,
            "has_cover": has_cover, "has_lyrics": bool(m.get("lyrics")),
            "authors": [a.get("name", "") if isinstance(a, dict) else str(a)
                        for a in (m.get("authors") or [])],
            "identity": identity,
        }
    # The enrichment verdict, so Get info can say "Matched (auto, 96%)" /
    # "Pinned by you" / "Not matched" alongside the file facts.
    row = appstate.meta_db.get_enrichment(filename)
    if row:
        info["match"] = {k: row.get(k) for k in
                         ("match_state", "match_source", "match_score",
                          "canon_artist", "canon_title", "canon_album", "canon_year")}
    return info
@@ -1,60 +0,0 @@
 """Practice loops — saved A/B regions per song.
 Extracted verbatim from ``server.py`` (R3); only the decorator receiver
 (``@app`` -> ``@router``) and the singleton reads (``meta_db`` ->
 ``appstate.meta_db``) changed. See ``appstate.py`` for why the reads stay
 module attributes.
 """
 from fastapi import APIRouter
 import appstate
 router = APIRouter()
@router.get("/api/loops")
 def list_loops(filename: str):
    # Hold the DB lock for the read: the shared single connection
    # (check_same_thread=False) is serialized through meta_db._lock by every
    # writer, so an unlocked SELECT here can overlap a POST/DELETE commit.
    db = appstate.meta_db
    with db._lock:
        rows = db.conn.execute(
            "SELECT id, name, start_time, end_time FROM loops WHERE filename = ? ORDER BY start_time",
            (filename,)
        ).fetchall()
    return [{"id": r[0], "name": r[1], "start": r[2], "end": r[3]} for r in rows]
@router.post("/api/loops")
 def save_loop(data: dict):
    filename = data.get("filename", "")
    name = data.get("name", "").strip()
    start = data.get("start")
    end = data.get("end")
    if not filename or start is None or end is None:
        return {"error": "Missing fields"}
    db = appstate.meta_db
    with db._lock:
        # COUNT + INSERT under one lock so two unnamed POSTs can't read the same
        # count and both mint "Loop N" (the count is only used to name the row).
        if not name:
            count = db.conn.execute(
                "SELECT COUNT(*) FROM loops WHERE filename = ?", (filename,)
            ).fetchone()[0]
            name = f"Loop {count + 1}"
        db.conn.execute(
            "INSERT INTO loops (filename, name, start_time, end_time) VALUES (?, ?, ?, ?)",
            (filename, name, float(start), float(end))
        )
        db.conn.commit()
    return {"ok": True, "name": name}
@router.delete("/api/loops/{loop_id}")
 def delete_loop(loop_id: int):
    with appstate.meta_db._lock:
        appstate.meta_db.conn.execute("DELETE FROM loops WHERE id = ?", (loop_id,))
        appstate.meta_db.conn.commit()
    return {"ok": True}
@@ -1,267 +0,0 @@
 """Playlists + custom playlist covers (fee[dB]ack v0.3.0).
 Extracted verbatim from ``server.py`` (R3). Edits: ``@app`` -> ``@router``,
 ``meta_db`` -> ``appstate.meta_db``, ``CONFIG_DIR`` -> ``appstate.config_dir``
 (both read at call time through the seam), and ``_clean_str`` now imports from
 ``reqfields``. See ``appstate.py``.
 """
 import logging
 import os
 import tempfile
 from pathlib import Path
 from fastapi import APIRouter
 from fastapi.responses import FileResponse, JSONResponse
 import appstate
 from reqfields import _clean_str
 log = logging.getLogger("feedBack.server")
 router = APIRouter()
 # Cache policy for the custom-cover file response: revalidate every time so a
 # replaced cover is never served stale (pairs with the mtime-ns URL token).
 _ART_CACHE_HEADERS = {"Cache-Control": "no-cache"}
 def _playlist_cover_path(pid) -> Path | None:
    """Filesystem path of a playlist's optional custom cover image (PNG),
    stored under CONFIG_DIR. Returns None for a non-integer id."""
    try:
        pid = int(pid)
    except (TypeError, ValueError):
        return None
    return appstate.config_dir / "playlist_covers" / f"{pid}.png"
 def _playlist_cover_url(pid) -> str | None:
    cover = _playlist_cover_path(pid)
    if not cover or not cover.exists():
        return None
    try:
        # Nanosecond mtime so a same-second replace/remove/re-upload still
        # changes the cache-bust token (int seconds could collide → stale image).
        mt = cover.stat().st_mtime_ns
    except OSError:
        mt = 0
    return f"/api/playlists/{pid}/cover?v={mt}"
@router.get("/api/playlists")
 def api_list_playlists():
    lists = appstate.meta_db.list_playlists()
    for pl in lists:
        pl["cover_url"] = _playlist_cover_url(pl["id"])
    return lists
@router.post("/api/playlists")
 def api_create_playlist(data: dict):
    name = _clean_str(data.get("name"))
    if not (1 <= len(name) <= 100):
        return JSONResponse({"error": "Playlist name must be 1–100 characters."}, status_code=400)
    # kind='album' = a curated album (§7.2): hand-picked works, a chosen chart
    # per slot, played front-to-back on the queue. Absent/None = a regular mix.
    kind = _clean_str(data.get("kind")) or None
    if kind not in (None, "album"):
        return JSONResponse({"error": "kind must be 'album' or omitted"}, status_code=400)
    return appstate.meta_db.create_playlist(name, kind=kind)
@router.get("/api/playlists/{pid}")
 def api_get_playlist(pid: int):
    pl = appstate.meta_db.get_playlist(pid)
    if pl is None:
        return JSONResponse({"error": "not found"}, status_code=404)
    pl["cover_url"] = _playlist_cover_url(pid)
    return pl
@router.patch("/api/playlists/{pid}")
 def api_rename_playlist(pid: int, data: dict):
    pl = appstate.meta_db.get_playlist(pid)
    if pl is None:
        return JSONResponse({"error": "not found"}, status_code=404)
    if pl["system_key"]:
        return JSONResponse({"error": "System playlists cannot be renamed."}, status_code=400)
    name = _clean_str(data.get("name"))
    if not (1 <= len(name) <= 100):
        return JSONResponse({"error": "Playlist name must be 1–100 characters."}, status_code=400)
    appstate.meta_db.rename_playlist(pid, name)
    return appstate.meta_db.get_playlist(pid)
@router.delete("/api/playlists/{pid}")
 def api_delete_playlist(pid: int):
    pl = appstate.meta_db.get_playlist(pid)
    if pl is None:
        return JSONResponse({"error": "not found"}, status_code=404)
    if pl["system_key"]:
        return JSONResponse({"error": "System playlists cannot be deleted."}, status_code=400)
    if not appstate.meta_db.delete_playlist(pid):   # vanished under us (concurrent delete)
        return JSONResponse({"error": "not found"}, status_code=404)
    cover = _playlist_cover_path(pid)       # drop any custom cover with the playlist
    if cover and cover.exists():
        try:
            cover.unlink()
        except OSError:
            pass
    return {"ok": True}
@router.post("/api/playlists/{pid}/songs")
 def api_add_playlist_song(pid: int, data: dict):
    if appstate.meta_db.get_playlist(pid) is None:
        return JSONResponse({"error": "not found"}, status_code=404)
    filename = _clean_str(data.get("filename"))
    if not filename:
        return JSONResponse({"error": "filename required"}, status_code=400)
    if appstate.meta_db.add_playlist_song(pid, filename) is None:   # playlist vanished under us
        return JSONResponse({"error": "not found"}, status_code=404)
    pl = appstate.meta_db.get_playlist(pid)
    return pl if pl is not None else JSONResponse({"error": "not found"}, status_code=404)
@router.patch("/api/playlists/{pid}/songs/{filename:path}")
 def api_update_playlist_slot(pid: int, filename: str, data: dict):
    """Edit one curated-album slot: {"arrangement": name|null} pins/clears the
    slot's arrangement; {"chart_filename": fn} swaps the slot to another chart
    of the same work (position + pin kept). Albums only — a mix has no slots."""
    pl = appstate.meta_db.get_playlist(pid)
    if pl is None:
        return JSONResponse({"error": "not found"}, status_code=404)
    if pl.get("kind") != "album":
        return JSONResponse({"error": "Slot editing is for albums."}, status_code=400)
    kwargs = {}
    if "chart_filename" in data:
        new_fn = _clean_str(data.get("chart_filename"))
        if not new_fn:
            return JSONResponse({"error": "chart_filename must be a filename"}, status_code=400)
        kwargs["new_filename"] = new_fn
    if "arrangement" in data:
        arr = data.get("arrangement")
        if arr is not None and not (isinstance(arr, str) and 1 <= len(arr.strip()) <= 100):
            return JSONResponse({"error": "arrangement must be a name or null"}, status_code=400)
        kwargs["arrangement"] = arr.strip() if isinstance(arr, str) else None
    if not kwargs:
        return JSONResponse({"error": "nothing to update"}, status_code=400)
    if appstate.meta_db.update_playlist_slot(pid, filename, **kwargs) is None:
        return JSONResponse(
            {"error": "no such slot, or the chart isn't a version of this song"},
            status_code=400)
    return appstate.meta_db.get_playlist(pid)
@router.delete("/api/playlists/{pid}/songs/{filename:path}")
 def api_remove_playlist_song(pid: int, filename: str):
    if appstate.meta_db.get_playlist(pid) is None:
        return JSONResponse({"error": "not found"}, status_code=404)
    appstate.meta_db.remove_playlist_song(pid, filename)
    pl = appstate.meta_db.get_playlist(pid)
    return pl if pl is not None else JSONResponse({"error": "not found"}, status_code=404)
@router.post("/api/playlists/{pid}/reorder")
 def api_reorder_playlist(pid: int, data: dict):
    pl = appstate.meta_db.get_playlist(pid)
    if pl is None:
        return JSONResponse({"error": "not found"}, status_code=404)
    order = data.get("order")
    if not isinstance(order, list) or not all(isinstance(f, str) for f in order):
        return JSONResponse({"error": "order must be a list of filenames"}, status_code=400)
    # Require an exact permutation of the playlist's current songs: a list with
    # duplicates, omissions, or extras would otherwise produce duplicate
    # positions / a partial reorder while still returning 200.
    current = [s["filename"] for s in pl["songs"]]
    if len(order) != len(current) or sorted(order) != sorted(current):
        return JSONResponse(
            {"error": "order must be a permutation of the playlist's current songs"},
            status_code=400,
        )
    appstate.meta_db.reorder_playlist(pid, order)
    return appstate.meta_db.get_playlist(pid)
@router.post("/api/playlists/{pid}/cover")
 async def api_set_playlist_cover(pid: int, data: dict):
    """Set a playlist's custom cover from a base64 / data-URL image (PNG/JPG).
    Overrides the content-dependent (song-art) cover. Stored as a small PNG
    thumbnail under CONFIG_DIR/playlist_covers/."""
    if appstate.meta_db.get_playlist(pid) is None:
        return JSONResponse({"error": "not found"}, status_code=404)
    import base64
    import io
    b64 = data.get("image", "")
    # Guard the type before the `","` membership test — a non-string image
    # (e.g. {"image": 123} / null) would otherwise raise TypeError → 500.
    # Mirrors the avatar/song-art upload guard.
    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]
    if not b64:
        return JSONResponse({"error": "No image data"}, status_code=400)
    try:
        img_data = base64.b64decode(b64)
    except Exception:
        return JSONResponse({"error": "Invalid base64"}, status_code=400)
    cover = _playlist_cover_path(pid)
    cover.parent.mkdir(parents=True, exist_ok=True)
    # Decode/validate the image — a bad payload is a CLIENT error (400), and the
    # message stays generic so it can't echo internals.
    try:
        from PIL import Image
        img = Image.open(io.BytesIO(img_data)).convert("RGB")
        img.thumbnail((640, 640))                  # covers stay small
    except Exception:
        return JSONResponse({"error": "Invalid image"}, status_code=400)
    # Persist. A save/replace failure is a SERVER error (500, logged, no
    # filesystem detail leaked) — the pre-split handler mislabeled these as 400
    # and echoed the exception. A unique temp name in the cover dir (not a shared
    # `{pid}.png.tmp`) means two concurrent uploads can't clobber each other's
    # temp file; the atomic replace publishes. Re-check the playlist still exists
    # just before publishing so a delete that raced the decode above can't leave
    # an orphan cover — cheap belt-and-braces; FeedBack is single-user
    # (Principle I), so a full per-playlist lock would be for a race the
    # deployment model precludes.
    tmp = None
    try:
        # mkstemp is inside the try too: an unwritable dir / full disk raises
        # here, and that's the same class of persistence failure as save/replace.
        fd, tmp_name = tempfile.mkstemp(prefix=f".{pid}.", suffix=".png.tmp", dir=str(cover.parent))
        tmp = Path(tmp_name)
        with os.fdopen(fd, "wb") as f:
            img.save(f, "PNG")
        if appstate.meta_db.get_playlist(pid) is None:
            tmp.unlink(missing_ok=True)
            return JSONResponse({"error": "not found"}, status_code=404)
        tmp.replace(cover)
    except Exception:
        if tmp is not None:
            tmp.unlink(missing_ok=True)
        log.exception("playlist cover save failed (pid=%s)", pid)
        return JSONResponse({"error": "could not save cover"}, status_code=500)
    return {"ok": True, "cover_url": _playlist_cover_url(pid)}
@router.get("/api/playlists/{pid}/cover")
 def api_get_playlist_cover(pid: int):
    cover = _playlist_cover_path(pid)
    if not cover or not cover.exists():
        return JSONResponse({"error": "not found"}, status_code=404)
    # no-cache (revalidate) like song art, so a replaced cover is never served
    # stale — pairs with the mtime-ns cache-bust token on the URL.
    return FileResponse(str(cover), media_type="image/png", headers=_ART_CACHE_HEADERS)
@router.delete("/api/playlists/{pid}/cover")
 def api_delete_playlist_cover(pid: int):
    cover = _playlist_cover_path(pid)
    if cover and cover.exists():
        try:
            cover.unlink()
        except OSError:
            pass
    return {"ok": True}
@@ -1,153 +0,0 @@
 """The router seam (`appstate.py`).
 The load-bearing assertion here is `test_server_wires_the_seam`: that `server`
 actually calls `appstate.configure(...)`. Every other test in this file would
 pass just fine against a seam nothing ever wires up — the same class of silent
 no-op that bit the frontend refactor twice when a scripted `setHostHooks` edit
 stopped matching its anchor. Unit tests cannot see wiring unless you make them
 look at it.
 """
 import importlib
 import sys
 import pytest
 import appstate
 def _close_server_dbs(mod):
    conn = getattr(getattr(mod, "meta_db", None), "conn", None)
    if conn is not None:
        getattr(mod, "_join_background_db_threads", lambda: None)()
        conn.close()
    ae_conn = getattr(getattr(mod, "audio_effect_mappings", None), "conn", None)
    if ae_conn is not None:
        ae_conn.close()
@pytest.fixture()
 def isolated_server(tmp_path, monkeypatch):
    """A freshly imported `server` bound to a throwaway CONFIG_DIR.
    Importing `server` constructs `MetadataDB` + `AudioEffectsMappingDB` at
    module level, so it MUST be re-imported under a patched CONFIG_DIR — an
    unguarded `import server` would create/mutate the developer's real
    `~/.local/share/feedback` databases. Same idiom as the other ~49
    server-importing suites.
    Teardown restores the appstate slots as well as closing the connections:
    leaving `appstate.meta_db` published but pointing at a closed sqlite handle
    would hand a later test (or router) a live-looking, dead singleton.
    """
    previous = (appstate.meta_db, appstate.audio_effect_mappings)
    monkeypatch.setenv("CONFIG_DIR", str(tmp_path))
    sys.modules.pop("server", None)
    mod = importlib.import_module("server")
    yield mod
    _close_server_dbs(mod)
    # Leave no half-torn-down `server` behind: the next fixture re-imports it.
    sys.modules.pop("server", None)
    appstate.configure(meta_db=previous[0], audio_effect_mappings=previous[1])
 def test_import_is_side_effect_free():
    """`import appstate` must construct nothing and touch no disk.
    This is why the ~49 fixtures that `sys.modules.pop("server")` and re-import
    (to rebuild `meta_db` under a patched CONFIG_DIR) keep working untouched:
    server owns construction, appstate only mirrors it. A singleton *owned*
    here would survive that pop and go stale.
    """
    sys.modules.pop("appstate", None)
    fresh = importlib.import_module("appstate")
    try:
        assert fresh.meta_db is None
        assert fresh.audio_effect_mappings is None
    finally:
        sys.modules["appstate"] = appstate
 def test_configure_publishes_known_slots():
    sentinel = object()
    original = appstate.meta_db
    try:
        appstate.configure(meta_db=sentinel)
        assert appstate.meta_db is sentinel
    finally:
        appstate.configure(meta_db=original)
 def test_configure_is_idempotent():
    """server re-imports call configure() again; the last write must win."""
    original = appstate.meta_db
    try:
        appstate.configure(meta_db="first")
        appstate.configure(meta_db="second")
        assert appstate.meta_db == "second"
    finally:
        appstate.configure(meta_db=original)
 def test_configure_rejects_an_unknown_slot():
    """A typo'd or stale keyword must raise, not silently create a global that
    nothing reads. A seam whose wiring can no-op undetected is worse than none."""
    with pytest.raises(TypeError, match="unknown slot"):
        appstate.configure(met_db="typo")
    assert not hasattr(appstate, "met_db")
 def test_late_bound_read_sees_a_later_configure():
    """Routers must read `appstate.meta_db`, never `from appstate import meta_db`.
    This pins the property that makes that rule work."""
    def router_style_read():
        return appstate.meta_db          # module attribute, resolved at call time
    original = appstate.meta_db
    try:
        appstate.configure(meta_db="before")
        assert router_style_read() == "before"
        appstate.configure(meta_db="after")
        assert router_style_read() == "after"
    finally:
        appstate.configure(meta_db=original)
 def test_server_wires_the_seam(isolated_server):
    """The one that catches a dropped `appstate.configure(...)` call.
    Identity, not truthiness, so a stray re-assignment or a half-applied edit
    fails here rather than in some router months later.
    """
    assert appstate.meta_db is isolated_server.meta_db
    assert appstate.audio_effect_mappings is isolated_server.audio_effect_mappings
    assert appstate.meta_db is not None
 def test_reimporting_server_republishes_the_fresh_singletons(
    isolated_server, tmp_path, monkeypatch
 ):
    """The 49-fixture contract, exercised end to end.
    Those fixtures `sys.modules.pop("server")` + re-import to rebuild `meta_db`
    under a new CONFIG_DIR, and know nothing about appstate. So the seam must
    re-publish on that second import. This is the test that would fail if
    `appstate` ever *owned* the singletons: a module-level `meta_db` there
    survives the pop and the assertions below would still see the FIRST DB.
    """
    first_db = isolated_server.meta_db
    assert appstate.meta_db is first_db
    assert str(tmp_path) in first_db.db_path
    second_config = tmp_path / "second"
    monkeypatch.setenv("CONFIG_DIR", str(second_config))
    sys.modules.pop("server", None)
    second_server = importlib.import_module("server")
    try:
        assert second_server.meta_db is not first_db          # genuinely rebuilt
        assert str(second_config) in second_server.meta_db.db_path
        assert appstate.meta_db is second_server.meta_db      # ...and re-published
        assert appstate.audio_effect_mappings is second_server.audio_effect_mappings
    finally:
        _close_server_dbs(second_server)
        sys.modules.pop("server", None)
@@ -16,8 +16,6 @@ import pytest
 import yaml
 from fastapi.testclient import TestClient
 from routers.ws_highway import _sanitize_authors
 # ── _sanitize_authors unit tests ────────────────────────────────────────────
@@ -37,7 +35,7 @@ def server_mod(monkeypatch, tmp_path):
 def test_sanitize_authors_valid(server_mod):
-    out = _sanitize_authors(
+    out = server_mod._sanitize_authors(
        {
            "authors": [
                {"name": "Azure", "role": "charter", "email": "a@b.c", "url": "x"},
@@ -55,7 +53,7 @@ def test_sanitize_authors_valid(server_mod):
 def test_sanitize_authors_skips_malformed(server_mod):
-    out = _sanitize_authors(
+    out = server_mod._sanitize_authors(
        {
            "authors": [
                {"name": ""},          # blank name → skipped
@@ -71,7 +69,7 @@ def test_sanitize_authors_skips_malformed(server_mod):
@pytest.mark.parametrize("manifest", [None, {}, {"authors": None}, {"authors": "x"}, "nope"])
 def test_sanitize_authors_absent_or_nonlist(server_mod, manifest):
-    assert _sanitize_authors(manifest) == []
+    assert server_mod._sanitize_authors(manifest) == []
 # ── song_info WS integration ────────────────────────────────────────────────
@@ -120,9 +118,6 @@ def make_client(tmp_path, monkeypatch):
        monkeypatch.setattr(server, "load_plugins", lambda *a, **kw: None)
        monkeypatch.setattr(server, "startup_scan", lambda: None)
        monkeypatch.setattr(server, "SLOPPAK_CACHE_DIR", tmp_path / "cache")
        # ws_highway reads the cache dir through the appstate seam now.
        import appstate as _appstate
        monkeypatch.setattr(_appstate, "sloppak_cache_dir", tmp_path / "cache")
        return server
    (tmp_path / "dlc").mkdir()
@@ -96,9 +96,6 @@ def make_client(tmp_path, monkeypatch):
        monkeypatch.setattr(server, "load_plugins", lambda *a, **kw: None)
        monkeypatch.setattr(server, "startup_scan", lambda: None)
        monkeypatch.setattr(server, "SLOPPAK_CACHE_DIR", tmp_path / "cache")
        # ws_highway reads the cache dir through the appstate seam now.
        import appstate as _appstate
        monkeypatch.setattr(_appstate, "sloppak_cache_dir", tmp_path / "cache")
        return server
    (tmp_path / "dlc").mkdir()
@@ -123,9 +123,6 @@ def make_client(tmp_path, monkeypatch):
        monkeypatch.setattr(server, "load_plugins", lambda *a, **kw: None)
        monkeypatch.setattr(server, "startup_scan", lambda: None)
        monkeypatch.setattr(server, "SLOPPAK_CACHE_DIR", tmp_path / "cache")
        # ws_highway reads the cache dir through the appstate seam now.
        import appstate as _appstate
        monkeypatch.setattr(_appstate, "sloppak_cache_dir", tmp_path / "cache")
        return server
    (tmp_path / "dlc").mkdir()
@@ -1,84 +0,0 @@
 """Concurrent unnamed loop saves must get unique names.
 `save_loop` auto-names an unnamed loop `Loop {count+1}`. If the `COUNT(*)` runs
 outside the DB lock (as it did before the fix), two simultaneous unnamed POSTs
 read the same count and both mint the same name. A `threading.Barrier` releases
 all workers into `save_loop` at once to force that interleave.
 Single-user app, so this race is unlikely in practice — but the fix is one lock
 scope, and the test pins it.
 """
 import threading
 import pytest
 import appstate
 from metadata_db import MetadataDB
 from routers import loops
@pytest.fixture()
 def meta_db(tmp_path):
    prev = appstate.meta_db
    db = MetadataDB(tmp_path)
    appstate.configure(meta_db=db)
    yield db
    db.conn.close()
    appstate.configure(meta_db=prev)
 def test_concurrent_unnamed_saves_get_unique_names(meta_db):
    workers = 16
    barrier = threading.Barrier(workers)
    names, errors = [], []
    lock = threading.Lock()
    def save():
        try:
            barrier.wait()                       # release all at once
            r = loops.save_loop({"filename": "song.feedpak", "start": 0.0, "end": 1.0})
            with lock:
                names.append(r["name"])
        except Exception as e:                   # noqa: BLE001 — surface any thread error
            with lock:
                errors.append(e)
    threads = [threading.Thread(target=save) for _ in range(workers)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    assert not errors, errors
    assert len(names) == workers
    # The load-bearing assertion: no two auto-named loops collide.
    assert len(set(names)) == workers, f"duplicate loop names: {sorted(names)}"
    # And the DB agrees — every insert landed.
    stored = meta_db.conn.execute(
        "SELECT COUNT(*) FROM loops WHERE filename = ?", ("song.feedpak",)
    ).fetchone()[0]
    assert stored == workers
 def test_list_and_save_do_not_error_under_interleave(meta_db):
    """A read overlapping writes must not raise (shared connection, one lock)."""
    stop = threading.Event()
    errors = []
    def reader():
        while not stop.is_set():
            try:
                loops.list_loops("song.feedpak")
            except Exception as e:               # noqa: BLE001
                errors.append(e)
    r = threading.Thread(target=reader)
    r.start()
    try:
        for i in range(40):
            loops.save_loop({"filename": "song.feedpak", "name": f"n{i}", "start": 0.0, "end": 1.0})
    finally:
        stop.set()
        r.join()
    assert not errors, errors
@@ -1,112 +0,0 @@
 """Guard: every first-party module `server.py` imports must be one the packagers copy.
 feedback-desktop's `scripts/bundle-slopsmith.sh` copies a **hardcoded list** from
 core into the app bundle — `server.py`, `VERSION`, `lib/`, `data/`, `static/`,
 `plugins/__init__.py`. A new root-level module (say `appstate.py`) ships fine in
 Docker, imports fine under pytest, and is then *silently dropped* from the
 packaged desktop app, which dies at startup with:
    File ".../Resources/slopsmith/server.py", line 71, in <module>
        import appstate
    ModuleNotFoundError: No module named 'appstate'
 That shipped once. This test is why it can't ship twice: it walks `server.py`'s
 module-level imports, keeps the ones that resolve inside this repo, and asserts
 each lives under a directory every packaging path already copies wholesale.
 If you add a first-party module for `server.py`, put it in `lib/` — the one core
 directory the Dockerfile (`COPY lib/`), `docker-compose.yml`, and the desktop
 bundler (`cp -r lib`) all copy, and that all three put on `sys.path`. If you
 genuinely need it at the repo root, you must also teach `bundle-slopsmith.sh`,
 the `Dockerfile`, `.dockerignore`, and `docker-compose.yml` about it — and then
 update `BUNDLED_ROOTS` below.
 """
 import ast
 import importlib.util
 import pathlib
 import pytest
 REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent
 # Directories every packaging path copies wholesale, plus the files copied by name.
 BUNDLED_ROOTS = ("lib", "plugins", "data", "static")
 BUNDLED_FILES = ("server.py", "main.py")
 def _server_toplevel_imports():
    """Module names imported at `server.py`'s top level (not inside a function)."""
    tree = ast.parse((REPO_ROOT / "server.py").read_text())
    names = set()
    for node in tree.body:                      # top level only — lazy imports are fine
        if isinstance(node, ast.Import):
            names.update(a.name.split(".")[0] for a in node.names)
        elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module:
            names.add(node.module.split(".")[0])
    return sorted(names)
 # `spec.origin` is not always a path: built-in and frozen stdlib modules use
 # these sentinels. `Path("frozen").resolve()` would land inside the repo and
 # report `os` as first-party — so filter them before touching the filesystem.
 _NON_PATH_ORIGINS = {"built-in", "frozen", "namespace"}
 def _first_party_origin(name):
    """Path of `name` if it resolves inside this repo, else None (stdlib/site-package)."""
    try:
        spec = importlib.util.find_spec(name)
    except (ImportError, ValueError):
        return None
    if spec is None:
        return None
    if spec.origin and spec.origin not in _NON_PATH_ORIGINS:
        origin = pathlib.Path(spec.origin)
    else:
        # Namespace/frozen: fall back to the first search location, if any.
        locations = list(getattr(spec, "submodule_search_locations", None) or [])
        if not locations:
            return None
        origin = pathlib.Path(locations[0])
    if not origin.is_absolute():
        return None                              # a sentinel, not a real path
    origin = origin.resolve()
    try:
        origin.relative_to(REPO_ROOT)
    except ValueError:
        return None                              # outside the repo → a dependency
    return origin
@pytest.mark.parametrize("name", _server_toplevel_imports())
 def test_server_import_is_bundled(name):
    origin = _first_party_origin(name)
    if origin is None:
        return                                   # stdlib or an installed dependency
    rel = origin.relative_to(REPO_ROOT)
    if rel.as_posix() in BUNDLED_FILES or rel.parts[0] in BUNDLED_ROOTS:
        return
    raise AssertionError(
        f"server.py imports `{name}` from {rel}, which no packager copies.\n"
        f"The desktop bundler (scripts/bundle-slopsmith.sh) copies only "
        f"{BUNDLED_FILES} and {BUNDLED_ROOTS}/, so the packaged app would die "
        f"at startup with ModuleNotFoundError: No module named '{name}'.\n"
        f"Move it under lib/, or teach bundle-slopsmith.sh + Dockerfile + "
        f".dockerignore + docker-compose.yml about it and update BUNDLED_ROOTS."
    )
 def test_the_seam_and_routers_live_under_lib():
    """Pin the two that already caused a shipped break."""
    for name in ("appstate", "routers"):
        origin = _first_party_origin(name)
        assert origin is not None, f"{name} does not resolve inside the repo"
        assert origin.relative_to(REPO_ROOT).parts[0] == "lib", (
            f"{name} resolved to {origin.relative_to(REPO_ROOT)}; it must live "
            f"under lib/ or the packaged desktop app will not ship it"
        )
@@ -1,107 +0,0 @@
 """Playlist cover upload: client vs server errors, no temp litter, no leak.
 The pre-split handler caught decode AND persistence failures in one `except`,
 returned 400 for both, and echoed the exception (`Invalid image: {e}`) — so a
 disk/permission failure was mislabeled as a client error and could leak a
 filesystem path. These pin the split.
 """
 import base64
 import io
 import pytest
 from fastapi.testclient import TestClient
 from PIL import Image
 import appstate
 from metadata_db import MetadataDB
 from routers import playlists
@pytest.fixture()
 def client(tmp_path):
    prev = (appstate.meta_db, appstate.config_dir)
    db = MetadataDB(tmp_path)
    appstate.configure(meta_db=db, config_dir=tmp_path)
    app_ = __import__("fastapi").FastAPI()
    app_.include_router(playlists.router)
    try:
        yield TestClient(app_), tmp_path
    finally:
        db.conn.close()
        appstate.configure(meta_db=prev[0], config_dir=prev[1])
 def _png_b64():
    buf = io.BytesIO()
    Image.new("RGB", (8, 8), (10, 20, 30)).save(buf, "PNG")
    return base64.b64encode(buf.getvalue()).decode()
 def _make_playlist(client):
    return client.post("/api/playlists", json={"name": "P"}).json()["id"]
 def test_valid_cover_saves_and_leaves_no_temp(client):
    c, tmp_path = client
    pid = _make_playlist(c)
    r = c.post(f"/api/playlists/{pid}/cover", json={"image": _png_b64()})
    assert r.status_code == 200
    cover_dir = tmp_path / "playlist_covers"
    assert (cover_dir / f"{pid}.png").exists()
    # The atomic-publish temp file must not linger.
    assert not list(cover_dir.glob("*.tmp"))
 def test_undecodable_image_is_a_400_without_leaking(client):
    c, _ = client
    pid = _make_playlist(c)
    # valid base64, not a valid image
    r = c.post(f"/api/playlists/{pid}/cover", json={"image": base64.b64encode(b"not an image").decode()})
    assert r.status_code == 400
    body = r.json()["error"]
    assert body == "Invalid image"          # generic — no exception detail echoed
    assert "playlist_covers" not in body    # no filesystem path leak
 def test_save_failure_is_a_500_not_a_400(client, monkeypatch):
    """A persistence failure (here: Image.save raising) must be a logged 500,
    not a 400 — the whole point of the decode/persist split. Negative-checks
    against the pre-fix behavior, which returned 400 for exactly this."""
    c, tmp_path = client
    pid = _make_playlist(c)
    payload = _png_b64()                     # build BEFORE patching save
    def boom(self, fp, *a, **k):
        raise OSError("disk full")
    monkeypatch.setattr(Image.Image, "save", boom)
    r = c.post(f"/api/playlists/{pid}/cover", json={"image": payload})
    assert r.status_code == 500
    assert "disk full" not in r.json()["error"]           # no internal detail
    assert not list((tmp_path / "playlist_covers").glob("*.tmp"))  # temp cleaned up
 def test_temp_creation_failure_is_a_500(client, monkeypatch):
    """mkstemp raising (unwritable dir / full disk) must hit the same logged 500
    path as a save failure, not escape as an unhandled server error."""
    import tempfile as _tempfile
    c, _ = client
    pid = _make_playlist(c)
    payload = _png_b64()
    def boom(*a, **k):
        raise OSError("read-only file system")
    monkeypatch.setattr(_tempfile, "mkstemp", boom)
    r = c.post(f"/api/playlists/{pid}/cover", json={"image": payload})
    assert r.status_code == 500
    assert "read-only" not in r.json()["error"]
 def test_upload_to_missing_playlist_is_404(client):
    c, _ = client
    r = c.post("/api/playlists/9999/cover", json={"image": _png_b64()})
    assert r.status_code == 404
@@ -6,10 +6,6 @@ import sys
 import pytest
 from fastapi.testclient import TestClient
 # Moved to routers/playlists in R3; reads appstate.config_dir, which the
 # `server` fixture configures via CONFIG_DIR before this is called.
 from routers.playlists import _playlist_cover_path
@pytest.fixture()
 def server(tmp_path, monkeypatch, isolate_logging):
@@ -172,6 +168,6 @@ def test_cover_rejects_non_string_image_with_400_not_500(client):
 def test_deleting_playlist_removes_custom_cover(client, server):
    pid = client.post("/api/playlists", json={"name": "Doomed"}).json()["id"]
    client.post(f"/api/playlists/{pid}/cover", json={"image": _png_b64()})
-    assert _playlist_cover_path(pid).exists()
+    assert server._playlist_cover_path(pid).exists()
    client.delete(f"/api/playlists/{pid}")
-    assert not _playlist_cover_path(pid).exists()
+    assert not server._playlist_cover_path(pid).exists()
@@ -1,110 +0,0 @@
 """A client that disconnects mid-stream is routine, not a logged error.
 The highway WS streams ~15 message batches before its keep-alive loop. A
 disconnect during that streaming used to fall through to the blanket
 `except Exception` and log `highway_ws unhandled error`. The dedicated
 `except WebSocketDisconnect: return` makes it quiet.
 """
 import asyncio
 import importlib
 import json
 import logging
 import sys
 import pytest
 import yaml
 from fastapi import WebSocketDisconnect
 def _write_sloppak(dlc_root):
    pak = dlc_root / "disc.sloppak"
    pak.mkdir(parents=True)
    (pak / "arrangements").mkdir()
    (pak / "arrangements" / "lead.json").write_text(json.dumps({
        "notes": [], "chords": [], "anchors": [], "handshapes": [],
        "templates": [], "beats": [{"time": 0.0, "measure": 1}],
        "sections": [{"name": "intro", "number": 1, "time": 0.0}],
    }))
    (pak / "manifest.yaml").write_text(yaml.safe_dump({
        "title": "Disc", "artist": "T", "album": "", "year": 2026, "duration": 10.0,
        "arrangements": [{"id": "lead", "name": "Lead", "file": "arrangements/lead.json"}],
        "stems": [],
    }, sort_keys=False))
    return pak
 class _DisconnectingWS:
    """Accepts, then raises WebSocketDisconnect on the first streamed send —
    i.e. a client that drops mid-stream."""
    def __init__(self):
        self.sends = 0
    async def accept(self):
        pass
    async def send_json(self, data):
        self.sends += 1
        # Raise only on the FIRST send. If the handler wrongly kept streaming
        # after catching the disconnect, later sends would succeed and `sends`
        # would climb past 1 — the exactly-one assertion catches that.
        if self.sends == 1:
            raise WebSocketDisconnect(code=1001)
    async def receive_text(self):
        raise WebSocketDisconnect(code=1001)
    async def close(self):
        pass
@pytest.fixture()
 def server(tmp_path, monkeypatch):
    (tmp_path / "dlc").mkdir()
    monkeypatch.setenv("CONFIG_DIR", str(tmp_path / "config"))
    monkeypatch.setenv("DLC_DIR", str(tmp_path / "dlc"))
    monkeypatch.setenv("FEEDBACK_SKIP_STARTUP_TASKS", "1")
    sys.modules.pop("server", None)
    mod = importlib.import_module("server")
    yield mod, tmp_path / "dlc"
    conn = getattr(getattr(mod, "meta_db", None), "conn", None)
    if conn is not None:
        getattr(mod, "_join_background_db_threads", lambda: None)()
        conn.close()
    sys.modules.pop("server", None)
 class _Capture(logging.Handler):
    def __init__(self):
        super().__init__()
        self.messages = []
    def emit(self, record):
        self.messages.append(record.getMessage())
 def test_midstream_disconnect_is_not_logged_as_error(server):
    """The first streamed send (`loading`) raises WebSocketDisconnect; the
    handler must return quietly, not log `highway_ws unhandled error`.
    A raw handler on `feedBack.server` is used rather than pytest's `caplog`
    because `configure_logging()` (run at server import) reroutes that logger
    through the structlog pipeline, which caplog's fixture doesn't observe.
    """
    _server, dlc = server
    _write_sloppak(dlc)
    from routers.ws_highway import highway_ws
    cap = _Capture()
    cap.setLevel(logging.ERROR)
    lg = logging.getLogger("feedBack.server")
    lg.addHandler(cap)
    try:
        ws = _DisconnectingWS()
        asyncio.run(highway_ws(ws, "disc.sloppak", arrangement=0))  # must not raise
    finally:
        lg.removeHandler(cap)
    assert ws.sends == 1, f"handler kept streaming after the disconnect ({ws.sends} sends)"
    unhandled = [m for m in cap.messages if "highway_ws unhandled error" in m]
    assert not unhandled, f"disconnect logged as unhandled error: {unhandled}"