diff --git a/Dockerfile b/Dockerfile
index 6b5e1d204..c6138d15a 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -23,7 +23,7 @@ RUN pip install --no-cache-dir --prefix=/install -r /tmp/requirements.txt
 
 FROM python:3.13.5-slim
 
-ARG COMMIT_TAG
+ARG COMMIT_TAG=dev
 ARG BUILD_DATE
 ARG DROPPEDNEEDLE_SOURCE_REVISION=unknown
 
diff --git a/backend/api/v1/routes/auth.py b/backend/api/v1/routes/auth.py
index 877d81744..c1d1c02a5 100644
--- a/backend/api/v1/routes/auth.py
+++ b/backend/api/v1/routes/auth.py
@@ -10,6 +10,8 @@
     AuthProvidersResponse,
     AuthResponse,
     CreateUserRequest,
+    DeviceSessionRequest,
+    DeviceSessionResponse,
     ImportCandidateListResponse,
     ImportUsersRequest,
     ImportUsersResponse,
@@ -210,6 +212,32 @@ async def list_sessions(
     return SessionListResponse(sessions = [session_to_response(token) for token in tokens])
 
 
+@router.post("/device-sessions", response_model=DeviceSessionResponse)
+async def create_device_session(
+    current_user: CurrentUserDep,
+    body: DeviceSessionRequest = MsgSpecBody(DeviceSessionRequest),
+    auth: AuthService = Depends(get_auth_service),
+) -> DeviceSessionResponse:
+    """Mint a separate session for a trusted companion such as Apple Watch.
+
+    The caller's bearer is never copied. The returned bearer is shown once,
+    belongs to the same user, and appears independently in Sessions.
+    """
+    try:
+        auth.validate_device_session_name(body.device_name)
+    except AuthenticationError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc))
+    providers = await auth.get_provider_names_for_users([current_user.id])
+    try:
+        token = await auth.issue_device_session(current_user.id, body.device_name)
+    except AuthenticationError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc))
+    return DeviceSessionResponse(
+        token=token,
+        user=user_to_response(current_user, providers.get(current_user.id)),
+    )
+
+
 @router.delete("/sessions/{session_id}", status_code = status.HTTP_204_NO_CONTENT)
 async def revoke_session(
     session_id: str,
diff --git a/backend/api/v1/routes/downloads.py b/backend/api/v1/routes/downloads.py
index b727dc48c..b8b89096e 100644
--- a/backend/api/v1/routes/downloads.py
+++ b/backend/api/v1/routes/downloads.py
@@ -75,6 +75,7 @@ def _to_response(  # noqa: ANN001 - DownloadTask
         user_id=task.user_id,
         download_type=task.download_type,
         source=task.source,
+        content_variant=task.content_variant,
         release_group_mbid=task.release_group_mbid,
         release_mbid=task.release_mbid,
         release_track_mbid=task.release_track_mbid,
diff --git a/backend/api/v1/routes/tracks.py b/backend/api/v1/routes/tracks.py
index 3b372c6da..57dcb2b1b 100644
--- a/backend/api/v1/routes/tracks.py
+++ b/backend/api/v1/routes/tracks.py
@@ -10,10 +10,9 @@
 from fastapi import APIRouter, Depends
 
 from api.v1.schemas.download import TrackRequestBody, TrackRequestResponse
-from core.dependencies import get_acquisition_dispatcher, get_quota_service
+from core.dependencies import get_request_service
 from infrastructure.msgspec_fastapi import MsgSpecBody, MsgSpecRoute
 from middleware import CurrentUserDep
-from services.native.download_service import ALREADY_IN_LIBRARY
 
 logger = logging.getLogger(__name__)
 
@@ -25,16 +24,13 @@ async def request_track(
     recording_mbid: str,
     current_user: CurrentUserDep,
     body: TrackRequestBody = MsgSpecBody(TrackRequestBody),
-    service=Depends(get_acquisition_dispatcher),
-    quota=Depends(get_quota_service),
+    service=Depends(get_request_service),
 ):
-    # Track asks bypass the approval queue (existing behaviour) but still count
-    # toward the rolling request quota (Feature C layer 1, D20) - their download
-    # task IS the ask, so the gate runs at this submit point.
-    await quota.check_request_quota(current_user.id, current_user.role)
-    task_id = await service.request_track(
+    return await service.request_track(
+        recording_mbid,
         user_id=current_user.id,
-        recording_mbid=recording_mbid,
+        user_role=current_user.role,
+        requested_by_name=current_user.display_name,
         artist_name=body.artist_name,
         track_title=body.track_title,
         album_title=body.album_title,
@@ -42,7 +38,5 @@ async def request_track(
         release_group_mbid=body.release_group_mbid,
         artist_mbid=body.artist_mbid,
         release_mbid=body.release_id,
+        content_variant=body.content_variant,
     )
-    if task_id == ALREADY_IN_LIBRARY:
-        return TrackRequestResponse(status="already_in_library")
-    return TrackRequestResponse(status="queued", task_id=task_id)
diff --git a/backend/api/v1/schemas/auth.py b/backend/api/v1/schemas/auth.py
index 4bfd8f10b..9b747ae28 100644
--- a/backend/api/v1/schemas/auth.py
+++ b/backend/api/v1/schemas/auth.py
@@ -36,6 +36,12 @@ class LoginRequest(AppStruct):
     password: str
 
 
+class DeviceSessionRequest(AppStruct):
+    """A caller-authorized, separately revocable session for one companion."""
+
+    device_name: str
+
+
 class PasswordRecoveryResetRequest(AppStruct):
     username: str
     recovery_code: str
@@ -58,6 +64,11 @@ class UserResponse(AppStruct):
     providers: list[str] = []
 
 
+class DeviceSessionResponse(AppStruct):
+    token: str
+    user: UserResponse
+
+
 class AuthResponse(AppStruct):
     token: str
     user: UserResponse
diff --git a/backend/api/v1/schemas/download.py b/backend/api/v1/schemas/download.py
index 4d8e82953..026049113 100644
--- a/backend/api/v1/schemas/download.py
+++ b/backend/api/v1/schemas/download.py
@@ -1,6 +1,7 @@
 """Request/response DTOs for the download-client + search + quarantine routes (Phase 6)."""
 
 import msgspec
+from typing import Literal
 
 from infrastructure.msgspec_fastapi import AppStruct
 from models.common import ServiceStatus
@@ -131,6 +132,7 @@ class DownloadTaskResponse(AppStruct):
     # "soulseek" | "usenet" - drives the source badge + the "via album NZB" label
     # (derived as source=="usenet" && download_type=="track").
     source: str
+    content_variant: str
     release_group_mbid: str
     release_mbid: str | None
     release_track_mbid: str | None
@@ -317,10 +319,12 @@ class TrackRequestBody(AppStruct):
     # MB RELEASE mbid (an edition): a SOFT acquisition target (D14) threaded into
     # DownloadTask.release_mbid - same value, two names (release_id on the wire).
     release_id: str | None = None
+    # ``clean`` opts into the fail-closed exact-recording verification contract.
+    content_variant: Literal["original", "clean"] = "original"
 
 
 class TrackRequestResponse(AppStruct):
-    status: str  # "queued" | "already_in_library"
+    status: str  # "awaiting_approval" | "queued" | "already_in_library"
     task_id: str | None = None
 
 
diff --git a/backend/api/v1/schemas/request.py b/backend/api/v1/schemas/request.py
index 69350223a..f6be90162 100644
--- a/backend/api/v1/schemas/request.py
+++ b/backend/api/v1/schemas/request.py
@@ -40,6 +40,9 @@ class BatchRequestResponse(AppStruct):
     requested: int = 0
     skipped: int = 0
     overflow: int = 0
+    # Native clients must render the decision made at this mutation boundary,
+    # not infer it from a role value that may have changed moments earlier.
+    status: str = "pending"
 
 
 class BatchCancelRequest(AppStruct):
diff --git a/backend/api/v1/schemas/requests_page.py b/backend/api/v1/schemas/requests_page.py
index 2770e5e71..ba21cafb1 100644
--- a/backend/api/v1/schemas/requests_page.py
+++ b/backend/api/v1/schemas/requests_page.py
@@ -30,6 +30,10 @@ class ActiveRequestItem(AppStruct):
     download_client: str | None = None
     user_id: str | None = None
     requested_by_name: str | None = None
+    request_kind: str = "album"
+    track_title: str | None = None
+    duration_seconds: int | None = None
+    track_release_group_mbid: str | None = None
 
 
 class RequestHistoryItem(AppStruct):
@@ -49,6 +53,10 @@ class RequestHistoryItem(AppStruct):
     reviewed_at: datetime | None = None
     download_task_id: str | None = None
     can_reimport: bool = False
+    request_kind: str = "album"
+    track_title: str | None = None
+    duration_seconds: int | None = None
+    track_release_group_mbid: str | None = None
 
 
 class ActiveRequestsResponse(AppStruct):
diff --git a/backend/api/v1/schemas/settings.py b/backend/api/v1/schemas/settings.py
index 5a7c78685..dd7ee9503 100644
--- a/backend/api/v1/schemas/settings.py
+++ b/backend/api/v1/schemas/settings.py
@@ -706,6 +706,10 @@ class ConnectAppsSettings(AppStruct):
 
     subsonic_enabled: bool = False
     jellyfin_enabled: bool = False
+    # Capability negotiation for clients that must distinguish the historical
+    # exact-track endpoint (which bypassed approval) from the approval-safe
+    # implementation. Older servers omit this field, so clients fail closed.
+    exact_track_approval_supported: bool = True
     transcoding_enabled: bool = True
     transcode_default_format: Literal["mp3", "opus"] = "mp3"
     transcode_max_bitrate_kbps: int = 320
@@ -714,6 +718,9 @@ class ConnectAppsSettings(AppStruct):
     discover_mode: Literal["local-only", "lazy-mb", "use-scrobble-targets"] = (
         "local-only"
     )
+    # Protocol capability advertised to clients. Older servers omit the field,
+    # allowing clients to fail closed instead of silently requesting an explicit copy.
+    clean_content_requests_supported: bool = True
 
     def __post_init__(self) -> None:
         if (
diff --git a/backend/core/config.py b/backend/core/config.py
index 977c2f74a..d34efc09a 100644
--- a/backend/core/config.py
+++ b/backend/core/config.py
@@ -3,6 +3,7 @@
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from typing import Self
 import logging
+import os
 import msgspec
 from core.exceptions import ConfigurationError
 from infrastructure.file_utils import atomic_write_json, read_json
@@ -64,6 +65,12 @@ class Settings(BaseSettings):
         default="contact@droppedneedle.com",
         description="Contact email for MusicBrainz API User-Agent. Override with your own if desired."
     )
+    http_user_agent: str | None = Field(
+        default=None,
+        max_length=512,
+        pattern=r"^[\x20-\x7E]*$",
+        description="Optional truthful application/version and contact identification for maintained integrations. Does not change provider rate limits.",
+    )
     discover_warmer_enabled: bool = Field(
         default=True,
         description="Proactively warm per-user Discover/Home in the background through the day (kill switch)."
@@ -157,8 +164,12 @@ def validate_config(self) -> Self:
         return self
     
     def get_user_agent(self) -> str:
+        if self.http_user_agent and self.http_user_agent.strip():
+            return self.http_user_agent.strip()
+        version = os.environ.get("COMMIT_TAG", "").strip() or "dev"
         id_part = self.instance_id[:8] if self.instance_id else "unknown"
-        return f"DroppedNeedle/1.0 ({id_part}; {self.contact_email}; https://www.droppedneedle.com)"
+        email = (self.contact_email or "").strip() or "contact@droppedneedle.com"
+        return f"DroppedNeedleApp/{version} ({id_part}; {email}; https://www.droppedneedle.com)"
 
     def load_from_file(self) -> None:
         if not self.config_file_path.exists():
diff --git a/backend/infrastructure/persistence/auth_store.py b/backend/infrastructure/persistence/auth_store.py
index def6668b8..f2d7efcd6 100644
--- a/backend/infrastructure/persistence/auth_store.py
+++ b/backend/infrastructure/persistence/auth_store.py
@@ -125,7 +125,8 @@ def _ensure_tables(self) -> None:
                     expires_at TEXT NOT NULL,
                     last_seen_at TEXT NOT NULL,
                     revoked INTEGER NOT NULL DEFAULT 0,
-                    user_agent TEXT
+                    user_agent TEXT,
+                    session_kind TEXT NOT NULL DEFAULT 'standard'
                 );
                 CREATE INDEX IF NOT EXISTS idx_auth_tokens_user
                     ON auth_tokens(user_id);
@@ -161,6 +162,15 @@ def _ensure_tables(self) -> None:
                 conn.execute("ALTER TABLE auth_oidc_states ADD COLUMN code_verifier TEXT")
             except sqlite3.OperationalError:
                 pass  # duplicate column - already present
+            # Companion sessions must not be inferred from an untrusted HTTP User-Agent.
+            # Existing rows remain standard because their provenance is ambiguous.
+            try:
+                conn.execute(
+                    "ALTER TABLE auth_tokens ADD COLUMN session_kind "
+                    "TEXT NOT NULL DEFAULT 'standard'"
+                )
+            except sqlite3.OperationalError:
+                pass  # duplicate column - already present
             # Username login (D3): additive, idempotent. `username` is the lowercased
             # login identifier; `username_display` preserves preferred casing. The
             # partial unique index lets pre-backfill NULL rows coexist.
@@ -640,6 +650,46 @@ def operation(conn: sqlite3.Connection) -> None:
             user_agent = user_agent,
         )
 
+    async def replace_companion_token(
+        self,
+        *,
+        id: str,
+        user_id: str,
+        token_hash: str,
+        user_agent: str,
+    ) -> TokenRecord:
+        """Issue one companion and revoke active same-label companions atomically."""
+        now = _now_iso()
+        expiry = _expiry_iso()
+
+        def operation(conn: sqlite3.Connection) -> None:
+            conn.execute(
+                """INSERT INTO auth_tokens
+                   (id, user_id, token_hash, issued_at, expires_at, last_seen_at,
+                    revoked, user_agent, session_kind)
+                   VALUES (?, ?, ?, ?, ?, ?, 0, ?, 'companion')""",
+                (id, user_id, token_hash, now, expiry, now, user_agent),
+            )
+            conn.execute(
+                """UPDATE auth_tokens SET revoked = 1
+                   WHERE user_id = ? AND user_agent = ? AND id != ?
+                     AND session_kind = 'companion'
+                     AND revoked = 0 AND expires_at > ?""",
+                (user_id, user_agent, id, now),
+            )
+
+        await self._write(operation)
+        return TokenRecord(
+            id=id,
+            user_id=user_id,
+            token_hash=token_hash,
+            issued_at=now,
+            expires_at=expiry,
+            last_seen_at=now,
+            revoked=False,
+            user_agent=user_agent,
+        )
+
     async def verify_token(self, raw_token: str) -> TokenRecord | None:
         candidate_hash = _hash_token(raw_token)
         now = _now_iso()
diff --git a/backend/infrastructure/persistence/download_store.py b/backend/infrastructure/persistence/download_store.py
index 2c4d326ee..f5908dc20 100644
--- a/backend/infrastructure/persistence/download_store.py
+++ b/backend/infrastructure/persistence/download_store.py
@@ -366,6 +366,7 @@
     "download_client",
     "source",
     "origin",
+    "content_variant",
     "source_username",
     "source_directory",
     "search_query",
@@ -471,6 +472,7 @@ def _ensure_tables(self) -> None:
                     -- source. Drives the origin-aware album gate, replace-on-import and
                     -- cap/quota exemptions (CollectionManagement D18/D19).
                     origin TEXT NOT NULL DEFAULT 'user',
+                    content_variant TEXT NOT NULL DEFAULT 'original',
                     source_username TEXT,
                     source_directory TEXT,
                     search_query TEXT,
@@ -550,6 +552,7 @@ def _ensure_tables(self) -> None:
                 ("release_track_mbid", "TEXT"),
                 ("source", "TEXT NOT NULL DEFAULT 'soulseek'"),
                 ("origin", "TEXT NOT NULL DEFAULT 'user'"),
+                ("content_variant", "TEXT NOT NULL DEFAULT 'original'"),
                 ("advertised_queue_depth", "INTEGER"),
                 ("queue_position_start", "INTEGER"),
                 ("queue_position_end", "INTEGER"),
@@ -665,6 +668,7 @@ async def create_task(
         download_client: str = "slskd",
         source: str = "soulseek",
         origin: str = "user",
+        content_variant: str = "original",
         search_query: str | None = None,
         search_job_id: str | None = None,
         candidate_index: int | None = None,
@@ -695,6 +699,7 @@ async def create_task(
             download_client=download_client,
             source=source,
             origin=origin,
+            content_variant=content_variant,
             search_query=search_query,
             search_job_id=search_job_id,
             candidate_index=candidate_index,
diff --git a/backend/infrastructure/persistence/native_library_store.py b/backend/infrastructure/persistence/native_library_store.py
index 2132e3405..bd05934f9 100644
--- a/backend/infrastructure/persistence/native_library_store.py
+++ b/backend/infrastructure/persistence/native_library_store.py
@@ -3067,6 +3067,84 @@ def operation(connection: sqlite3.Connection) -> set[str]:
 
         return await self._read(operation)
 
+    async def target_enrichment_candidates(
+        self, *, after_mbid: str | None, limit: int
+    ) -> list[tuple[str, str, dict[str, Any]]]:
+        """Return one keyset page of live MusicBrainz artist and album identities."""
+
+        cursor_type = ""
+        cursor_mbid = ""
+        legacy_cursor = after_mbid or ""
+        if after_mbid and ":" in after_mbid:
+            candidate_type, candidate_mbid = after_mbid.split(":", 1)
+            if candidate_type in {"artist", "album"}:
+                cursor_type = candidate_type
+                cursor_mbid = candidate_mbid.casefold()
+                legacy_cursor = ""
+
+        def operation(
+            connection: sqlite3.Connection,
+        ) -> list[tuple[str, str, dict[str, Any]]]:
+            union = (
+                "SELECT entity_type, mbid_lower, name, title, artist_name FROM ("
+                "SELECT 'album' AS entity_type, lower(identity.release_group_mbid) "
+                "AS mbid_lower, NULL AS name, album.title AS title, "
+                "album.album_artist_name AS artist_name "
+                "FROM local_album_external_identities identity "
+                "JOIN local_albums album ON album.id = identity.local_album_id "
+                "WHERE identity.provider = 'musicbrainz' "
+                "AND album.retired_into_album_id IS NULL "
+                "AND EXISTS (SELECT 1 FROM local_tracks track "
+                "WHERE track.local_album_id = album.id "
+                "AND track.availability = 'indexed') "
+                "UNION ALL "
+                "SELECT 'artist' AS entity_type, lower(identity.provider_artist_id) "
+                "AS mbid_lower, artist.display_name AS name, NULL AS title, "
+                "NULL AS artist_name "
+                "FROM local_artist_external_identities identity "
+                "JOIN local_artists artist ON artist.id = identity.local_artist_id "
+                "WHERE identity.provider = 'musicbrainz' "
+                "AND artist.retired_into_artist_id IS NULL "
+                "AND EXISTS (SELECT 1 FROM local_album_artists credit "
+                "JOIN local_albums album ON album.id = credit.local_album_id "
+                "JOIN local_tracks track ON track.local_album_id = album.id "
+                "WHERE credit.local_artist_id = artist.id "
+                "AND album.retired_into_album_id IS NULL "
+                "AND track.availability = 'indexed')) "
+            )
+            if legacy_cursor:
+                rows = connection.execute(
+                    union
+                    + "WHERE mbid_lower > ? ORDER BY mbid_lower, entity_type LIMIT ?",
+                    (legacy_cursor.casefold(), max(1, limit)),
+                ).fetchall()
+            else:
+                rows = connection.execute(
+                    union
+                    + "WHERE entity_type > ? OR "
+                    "(entity_type = ? AND mbid_lower > ?) "
+                    "ORDER BY entity_type, mbid_lower LIMIT ?",
+                    (cursor_type, cursor_type, cursor_mbid, max(1, limit)),
+                ).fetchall()
+
+            candidates: list[tuple[str, str, dict[str, Any]]] = []
+            for row in rows:
+                entity_type = str(row["entity_type"])
+                payload = (
+                    {"name": str(row["name"])}
+                    if entity_type == "artist"
+                    else {
+                        "title": str(row["title"]),
+                        "artist_name": str(row["artist_name"]),
+                    }
+                )
+                candidates.append(
+                    (entity_type, str(row["mbid_lower"]), payload)
+                )
+            return candidates
+
+        return await self._read(operation)
+
     async def target_existing_provider_artist_ids(
         self, identifiers: list[str]
     ) -> set[str]:
diff --git a/backend/infrastructure/persistence/request_history.py b/backend/infrastructure/persistence/request_history.py
index 69da9a927..58743f4da 100644
--- a/backend/infrastructure/persistence/request_history.py
+++ b/backend/infrastructure/persistence/request_history.py
@@ -22,6 +22,19 @@
     ")"
 )
 
+_REIMPORTABLE_JOIN_CONDITION = (
+    "rh.status = 'failed'"
+    " AND rh.download_task_id IS NOT NULL"
+    " AND EXISTS ("
+    "SELECT 1 FROM download_tasks"
+    " WHERE download_tasks.id = rh.download_task_id"
+    " AND download_tasks.status IN ('failed', 'partial')"
+    " AND download_tasks.source_username IS NOT NULL"
+    " AND download_tasks.search_job_id IS NOT NULL"
+    " AND download_tasks.candidate_index IS NOT NULL"
+    ")"
+)
+
 
 class RequestHistoryRecord(msgspec.Struct):
     musicbrainz_id: str
@@ -42,6 +55,11 @@ class RequestHistoryRecord(msgspec.Struct):
     reviewed_by_id: str | None = None
     reviewed_by_name: str | None = None
     reviewed_at: str | None = None
+    request_kind: str = "album"
+    track_title: str | None = None
+    duration_seconds: int | None = None
+    track_release_group_mbid: str | None = None
+    content_variant: str = "original"
 
 
 class RequestHistoryStore:
@@ -97,6 +115,11 @@ def _ensure_tables(self) -> None:
                 ("reviewed_at", "TEXT"),
                 ("download_task_id", "TEXT"),
                 ("release_mbid", "TEXT"),
+                ("request_kind", "TEXT NOT NULL DEFAULT 'album'"),
+                ("track_title", "TEXT"),
+                ("duration_seconds", "INTEGER"),
+                ("track_release_group_mbid", "TEXT"),
+                ("content_variant", "TEXT NOT NULL DEFAULT 'original'"),
             ]:
                 try:
                     conn.execute(
@@ -115,6 +138,31 @@ def _ensure_tables(self) -> None:
                 )
                 """
             )
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS request_history_requesters (
+                    user_id TEXT NOT NULL,
+                    musicbrainz_id_lower TEXT NOT NULL,
+                    requested_at TEXT NOT NULL,
+                    requested_by_name TEXT,
+                    PRIMARY KEY (user_id, musicbrainz_id_lower)
+                )
+                """
+            )
+            conn.execute(
+                "CREATE INDEX IF NOT EXISTS idx_request_history_requesters_mbid "
+                "ON request_history_requesters(musicbrainz_id_lower)"
+            )
+            # Preserve ownership of rows created before multi-listener attribution.
+            conn.execute(
+                """
+                INSERT OR IGNORE INTO request_history_requesters (
+                    user_id, musicbrainz_id_lower, requested_at, requested_by_name
+                )
+                SELECT user_id, musicbrainz_id_lower, requested_at, requested_by_name
+                FROM request_history WHERE user_id IS NOT NULL
+                """
+            )
             conn.commit()
         finally:
             conn.close()
@@ -154,7 +202,11 @@ def _row_to_record(row: sqlite3.Row | None) -> RequestHistoryRecord | None:
             artist_mbid=row["artist_mbid"],
             year=row["year"],
             cover_url=row["cover_url"],
-            requested_at=row["requested_at"],
+            requested_at=(
+                row["requester_requested_at"]
+                if "requester_requested_at" in keys
+                else row["requested_at"]
+            ),
             completed_at=row["completed_at"],
             status=row["status"],
             download_task_id=row["download_task_id"]
@@ -166,16 +218,34 @@ def _row_to_record(row: sqlite3.Row | None) -> RequestHistoryRecord | None:
             auto_download_artist=bool(row["auto_download_artist"])
             if row["auto_download_artist"] is not None
             else False,
-            user_id=row["user_id"] if "user_id" in keys else None,
-            requested_by_name=row["requested_by_name"]
-            if "requested_by_name" in keys
-            else None,
+            user_id=(
+                row["requester_user_id"]
+                if "requester_user_id" in keys
+                else (row["user_id"] if "user_id" in keys else None)
+            ),
+            requested_by_name=(
+                row["requester_name"]
+                if "requester_name" in keys
+                else (row["requested_by_name"] if "requested_by_name" in keys else None)
+            ),
             release_mbid=row["release_mbid"] if "release_mbid" in keys else None,
             reviewed_by_id=row["reviewed_by_id"] if "reviewed_by_id" in keys else None,
             reviewed_by_name=row["reviewed_by_name"]
             if "reviewed_by_name" in keys
             else None,
             reviewed_at=row["reviewed_at"] if "reviewed_at" in keys else None,
+            request_kind=(row["request_kind"] if "request_kind" in keys else "album")
+            or "album",
+            content_variant=(row["content_variant"] if "content_variant" in keys else "original") or "original",
+            track_title=row["track_title"] if "track_title" in keys else None,
+            duration_seconds=(
+                row["duration_seconds"] if "duration_seconds" in keys else None
+            ),
+            track_release_group_mbid=(
+                row["track_release_group_mbid"]
+                if "track_release_group_mbid" in keys
+                else None
+            ),
         )
 
     async def async_record_request(
@@ -192,6 +262,11 @@ async def async_record_request(
         requested_by_name: str | None = None,
         release_mbid: str | None = None,
         initial_status: str = "pending",
+        request_kind: str = "album",
+        track_title: str | None = None,
+        duration_seconds: int | None = None,
+        track_release_group_mbid: str | None = None,
+        content_variant: str = "original",
     ) -> None:
         requested_at = datetime.now(timezone.utc).isoformat()
         normalized_mbid = musicbrainz_id.lower()
@@ -203,8 +278,9 @@ def operation(conn: sqlite3.Connection) -> None:
                     musicbrainz_id_lower, musicbrainz_id, artist_name, album_title,
                     artist_mbid, year, cover_url, requested_at, completed_at, status,
                     monitor_artist, auto_download_artist, user_id, requested_by_name,
-                    release_mbid
-                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?)
+                    release_mbid, request_kind, track_title, duration_seconds,
+                    track_release_group_mbid, content_variant
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                 ON CONFLICT(musicbrainz_id_lower) DO UPDATE SET
                     musicbrainz_id = excluded.musicbrainz_id,
                     artist_name = excluded.artist_name,
@@ -219,7 +295,12 @@ def operation(conn: sqlite3.Connection) -> None:
                     auto_download_artist = excluded.auto_download_artist,
                     user_id = COALESCE(excluded.user_id, request_history.user_id),
                     requested_by_name = COALESCE(excluded.requested_by_name, request_history.requested_by_name),
-                    release_mbid = excluded.release_mbid
+                    release_mbid = excluded.release_mbid,
+                    request_kind = excluded.request_kind,
+                    track_title = excluded.track_title,
+                    duration_seconds = excluded.duration_seconds,
+                    track_release_group_mbid = excluded.track_release_group_mbid,
+                    content_variant = excluded.content_variant
                 """,
                 (
                     normalized_mbid,
@@ -236,8 +317,28 @@ def operation(conn: sqlite3.Connection) -> None:
                     user_id,
                     requested_by_name,
                     release_mbid,
+                    request_kind,
+                    track_title,
+                    duration_seconds,
+                    track_release_group_mbid,
+                    content_variant,
                 ),
             )
+            if user_id is not None:
+                conn.execute(
+                    """
+                    INSERT INTO request_history_requesters (
+                        user_id, musicbrainz_id_lower, requested_at, requested_by_name
+                    ) VALUES (?, ?, ?, ?)
+                    ON CONFLICT (user_id, musicbrainz_id_lower) DO UPDATE SET
+                        requested_at = excluded.requested_at,
+                        requested_by_name = COALESCE(
+                            excluded.requested_by_name,
+                            request_history_requesters.requested_by_name
+                        )
+                    """,
+                    (user_id, normalized_mbid, requested_at, requested_by_name),
+                )
 
         await self._write(operation)
 
@@ -302,10 +403,143 @@ def operation(conn: sqlite3.Connection) -> int:
                 """,
                 rows,
             )
+            if user_id is not None:
+                conn.executemany(
+                    """
+                    INSERT INTO request_history_requesters (
+                        user_id, musicbrainz_id_lower, requested_at, requested_by_name
+                    ) VALUES (?, ?, ?, ?)
+                    ON CONFLICT (user_id, musicbrainz_id_lower) DO UPDATE SET
+                        requested_at = excluded.requested_at,
+                        requested_by_name = COALESCE(
+                            excluded.requested_by_name,
+                            request_history_requesters.requested_by_name
+                        )
+                    """,
+                    [
+                        (
+                            user_id,
+                            item["musicbrainz_id"].lower(),
+                            requested_at,
+                            requested_by_name,
+                        )
+                        for item in items
+                    ],
+                )
             return len(rows)
 
         return await self._write(operation)
 
+    async def async_add_requester(
+        self,
+        musicbrainz_id: str,
+        user_id: str | None,
+        requested_by_name: str | None = None,
+    ) -> None:
+        if user_id is None:
+            return
+        await self.async_add_requesters(
+            [musicbrainz_id], user_id, requested_by_name=requested_by_name
+        )
+
+    async def async_add_requesters(
+        self,
+        musicbrainz_ids: list[str],
+        user_id: str | None,
+        requested_by_name: str | None = None,
+    ) -> None:
+        if user_id is None:
+            return
+        requested_at = datetime.now(timezone.utc).isoformat()
+        normalized = list(
+            dict.fromkeys(value.casefold() for value in musicbrainz_ids if value)
+        )
+        if not normalized:
+            return
+
+        def operation(conn: sqlite3.Connection) -> None:
+            conn.executemany(
+                """
+                INSERT INTO request_history_requesters (
+                    user_id, musicbrainz_id_lower, requested_at, requested_by_name
+                ) VALUES (?, ?, ?, ?)
+                ON CONFLICT (user_id, musicbrainz_id_lower) DO UPDATE SET
+                    requested_at = excluded.requested_at,
+                    requested_by_name = COALESCE(
+                        excluded.requested_by_name,
+                        request_history_requesters.requested_by_name
+                    )
+                """,
+                [
+                    (user_id, musicbrainz_id, requested_at, requested_by_name)
+                    for musicbrainz_id in normalized
+                ],
+            )
+
+        await self._write(operation)
+
+    async def async_is_requester(self, user_id: str, musicbrainz_id: str) -> bool:
+        normalized_mbid = musicbrainz_id.casefold()
+
+        def operation(conn: sqlite3.Connection) -> bool:
+            row = conn.execute(
+                "SELECT 1 FROM request_history_requesters "
+                "WHERE user_id = ? AND musicbrainz_id_lower = ?",
+                (user_id, normalized_mbid),
+            ).fetchone()
+            return row is not None
+
+        return await self._read(operation)
+
+    async def async_requester_count(self, musicbrainz_id: str) -> int:
+        normalized_mbid = musicbrainz_id.casefold()
+
+        def operation(conn: sqlite3.Connection) -> int:
+            row = conn.execute(
+                "SELECT COUNT(*) AS count FROM request_history_requesters "
+                "WHERE musicbrainz_id_lower = ?",
+                (normalized_mbid,),
+            ).fetchone()
+            return int(row["count"] if row is not None else 0)
+
+        return await self._read(operation)
+
+    async def async_remove_requester(self, user_id: str, musicbrainz_id: str) -> bool:
+        """Remove only one listener's interest and transfer primary attribution."""
+        normalized_mbid = musicbrainz_id.casefold()
+
+        def operation(conn: sqlite3.Connection) -> bool:
+            cursor = conn.execute(
+                "DELETE FROM request_history_requesters "
+                "WHERE user_id = ? AND musicbrainz_id_lower = ?",
+                (user_id, normalized_mbid),
+            )
+            if cursor.rowcount <= 0:
+                return False
+            owner = conn.execute(
+                "SELECT user_id FROM request_history " "WHERE musicbrainz_id_lower = ?",
+                (normalized_mbid,),
+            ).fetchone()
+            if owner is not None and owner["user_id"] == user_id:
+                successor = conn.execute(
+                    "SELECT user_id, requested_by_name FROM request_history_requesters "
+                    "WHERE musicbrainz_id_lower = ? ORDER BY requested_at ASC LIMIT 1",
+                    (normalized_mbid,),
+                ).fetchone()
+                if successor is not None:
+                    conn.execute(
+                        "UPDATE request_history SET user_id = ?, requested_by_name = ? "
+                        "WHERE musicbrainz_id_lower = ?",
+                        (
+                            successor["user_id"],
+                            successor["requested_by_name"],
+                            normalized_mbid,
+                        ),
+                    )
+            return True
+
+        return await self._write(operation)
+
     async def async_get_record(
         self, musicbrainz_id: str
     ) -> RequestHistoryRecord | None:
@@ -422,6 +656,18 @@ def operation(conn: sqlite3.Connection) -> int:
                     "DELETE FROM request_history_dismissals WHERE musicbrainz_id_lower = ?",
                     (source_key,),
                 )
+                conn.execute(
+                    "INSERT OR IGNORE INTO request_history_requesters "
+                    "(user_id, musicbrainz_id_lower, requested_at, requested_by_name) "
+                    "SELECT user_id, ?, requested_at, requested_by_name "
+                    "FROM request_history_requesters WHERE musicbrainz_id_lower = ?",
+                    (target_key, source_key),
+                )
+                conn.execute(
+                    "DELETE FROM request_history_requesters "
+                    "WHERE musicbrainz_id_lower = ?",
+                    (source_key,),
+                )
                 conn.execute(
                     "DELETE FROM request_history WHERE musicbrainz_id_lower IN (?, ?)",
                     (source_key, target_key),
@@ -534,7 +780,10 @@ async def async_get_active_count_for_user(self, user_id: str) -> int:
 
         def operation(conn: sqlite3.Connection) -> int:
             row = conn.execute(
-                f"SELECT COUNT(*) AS count FROM request_history WHERE user_id = ? AND status IN ({placeholders})",
+                "SELECT COUNT(*) AS count FROM request_history AS rh "
+                "JOIN request_history_requesters AS rr "
+                "ON rr.musicbrainz_id_lower = rh.musicbrainz_id_lower "
+                f"WHERE rr.user_id = ? AND rh.status IN ({placeholders})",
                 (user_id, *self._USER_ACTIVE_STATUSES),
             ).fetchone()
             return int(row["count"] if row is not None else 0)
@@ -549,7 +798,14 @@ async def async_get_active_requests_for_user(
 
         def operation(conn: sqlite3.Connection) -> list[RequestHistoryRecord]:
             rows = conn.execute(
-                f"SELECT * FROM request_history WHERE user_id = ? AND status IN ({placeholders}) ORDER BY requested_at DESC",
+                "SELECT rh.*, rr.user_id AS requester_user_id, "
+                "rr.requested_by_name AS requester_name, "
+                "rr.requested_at AS requester_requested_at "
+                "FROM request_history AS rh "
+                "JOIN request_history_requesters AS rr "
+                "ON rr.musicbrainz_id_lower = rh.musicbrainz_id_lower "
+                f"WHERE rr.user_id = ? AND rh.status IN ({placeholders}) "
+                "ORDER BY rr.requested_at DESC",
                 (user_id, *self._USER_ACTIVE_STATUSES),
             ).fetchall()
             return [
@@ -647,36 +903,44 @@ async def async_get_history_for_user(
         offset = (safe_page - 1) * safe_page_size
 
         _SORT_MAP = {
-            "newest": "requested_at DESC",
-            "oldest": "requested_at ASC",
-            "status": "status ASC, requested_at DESC",
+            "newest": "rr.requested_at DESC",
+            "oldest": "rr.requested_at ASC",
+            "status": "rh.status ASC, rr.requested_at DESC",
         }
-        order_clause = _SORT_MAP.get(sort or "", "requested_at DESC")
+        order_clause = _SORT_MAP.get(sort or "", "rr.requested_at DESC")
 
         def operation(
             conn: sqlite3.Connection,
         ) -> tuple[list[RequestHistoryRecord], int]:
             dismiss_clause = (
-                "AND musicbrainz_id_lower NOT IN "
+                "AND rh.musicbrainz_id_lower NOT IN "
                 "(SELECT musicbrainz_id_lower FROM request_history_dismissals WHERE user_id = ?)"
             )
             if status_filter == "reimportable":
-                where = (
-                    f"WHERE user_id = ? AND {_REIMPORTABLE_CONDITION} {dismiss_clause}"
-                )
+                where = f"WHERE rr.user_id = ? AND {_REIMPORTABLE_JOIN_CONDITION} {dismiss_clause}"
                 params: tuple = (user_id, user_id)
             elif status_filter:
-                where = f"WHERE user_id = ? AND status = ? {dismiss_clause}"
+                where = f"WHERE rr.user_id = ? AND rh.status = ? {dismiss_clause}"
                 params = (user_id, status_filter, user_id)
             else:
-                where = f"WHERE user_id = ? {dismiss_clause}"
+                where = f"WHERE rr.user_id = ? {dismiss_clause}"
                 params = (user_id, user_id)
 
             total_row = conn.execute(
-                f"SELECT COUNT(*) AS count FROM request_history {where}", params
+                "SELECT COUNT(*) AS count FROM request_history AS rh "
+                "JOIN request_history_requesters AS rr "
+                "ON rr.musicbrainz_id_lower = rh.musicbrainz_id_lower "
+                f"{where}",
+                params,
             ).fetchone()
             rows = conn.execute(
-                f"SELECT * FROM request_history {where} ORDER BY {order_clause} LIMIT ? OFFSET ?",
+                "SELECT rh.*, rr.user_id AS requester_user_id, "
+                "rr.requested_by_name AS requester_name, "
+                "rr.requested_at AS requester_requested_at "
+                "FROM request_history AS rh "
+                "JOIN request_history_requesters AS rr "
+                "ON rr.musicbrainz_id_lower = rh.musicbrainz_id_lower "
+                f"{where} ORDER BY {order_clause} LIMIT ? OFFSET ?",
                 params + (safe_page_size, offset),
             ).fetchall()
             records = [
@@ -802,6 +1066,25 @@ def operation(conn: sqlite3.Connection) -> None:
 
         await self._write(operation)
 
+    async def async_get_record_by_download_task_id(
+        self, download_task_id: str
+    ) -> RequestHistoryRecord | None:
+        """Resolve the request that owns a native task.
+
+        Album requests historically used the release-group MBID as their key.
+        Exact-track requests use the recording MBID, so task ownership is the
+        only identifier that is correct for both request kinds.
+        """
+
+        def operation(conn: sqlite3.Connection) -> RequestHistoryRecord | None:
+            row = conn.execute(
+                "SELECT * FROM request_history WHERE download_task_id = ? LIMIT 1",
+                (download_task_id,),
+            ).fetchone()
+            return self._row_to_record(row)
+
+        return await self._read(operation)
+
     async def async_delete_record(self, musicbrainz_id: str) -> bool:
         normalized_mbid = musicbrainz_id.lower()
 
@@ -814,6 +1097,10 @@ def operation(conn: sqlite3.Connection) -> bool:
                 "DELETE FROM request_history_dismissals WHERE musicbrainz_id_lower = ?",
                 (normalized_mbid,),
             )
+            conn.execute(
+                "DELETE FROM request_history_requesters WHERE musicbrainz_id_lower = ?",
+                (normalized_mbid,),
+            )
             return cursor.rowcount > 0
 
         return await self._write(operation)
@@ -877,6 +1164,10 @@ def operation(conn: sqlite3.Connection) -> int:
                 # no wanted_watches table yet (fresh DB before the store's first
                 # construction) - nothing can be watched, prune unguarded
                 cursor = conn.execute(base, (*terminal_statuses, cutoff_iso))
+            conn.execute(
+                "DELETE FROM request_history_requesters WHERE musicbrainz_id_lower "
+                "NOT IN (SELECT musicbrainz_id_lower FROM request_history)"
+            )
             return cursor.rowcount
 
         return await self._write(operation)
diff --git a/backend/main.py b/backend/main.py
index 7d5d34f2c..268f55f84 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -815,6 +815,7 @@ def _events_poll_time() -> str:
         "/api/v1/discover": (10.0, 20),
         "/api/v1/covers": (15.0, 30),
         "/api/v1/auth/login": (2.0, 5),
+        "/api/v1/auth/device-sessions": (1.0, 5),
         "/api/v1/auth/password-recovery/reset": (1.0, 5),
         "/api/v1/auth/setup": (1.0, 3),
         "/api/v1/auth/plex/poll": (5.0, 10),
diff --git a/backend/models/download.py b/backend/models/download.py
index e88cdeff5..2f6726482 100644
--- a/backend/models/download.py
+++ b/backend/models/download.py
@@ -142,6 +142,9 @@ class DownloadTask(AppStruct):
     # or "upgrade" (a curator-triggered quality upgrade). Orthogonal to ``source``;
     # defaulted so old rows decode as user requests.
     origin: str = "user"
+    # Requested lyrical-content edition. ``clean`` is a safety contract, not a
+    # filename hint: import must fingerprint to the exact requested recording.
+    content_variant: str = "original"
     source_username: str | None = None
     source_directory: str | None = None
     search_query: str | None = None
diff --git a/backend/models/download_manifest.py b/backend/models/download_manifest.py
index 78cb00d38..5ece0d53f 100644
--- a/backend/models/download_manifest.py
+++ b/backend/models/download_manifest.py
@@ -81,6 +81,8 @@ class DownloadManifest(AppStruct):
     # The owning task's origin ('user' | 'retry' | 'upgrade'). Replace-on-import fires
     # only for 'upgrade' (D18); legacy manifests decode as 'user' (add-only, unchanged).
     origin: str = "user"
+    # Durable clean-only intent. Legacy manifests decode as ``original``.
+    content_variant: str = "original"
     # Free Music uses a separate task store. Conversion holding therefore carries
     # the administrator explicitly instead of looking up a built-in download task.
     requested_by_user_id: str | None = None
diff --git a/backend/repositories/coverart_artist.py b/backend/repositories/coverart_artist.py
index f3b665e9a..542826506 100644
--- a/backend/repositories/coverart_artist.py
+++ b/backend/repositories/coverart_artist.py
@@ -28,7 +28,7 @@
 logger = logging.getLogger(__name__)
 LOCAL_SOURCE_TIMEOUT_SECONDS = 1.0
 T = TypeVar("T")
-DEFAULT_EXTERNAL_USER_AGENT = "DroppedNeedle/1.0 (contact@droppedneedle.com; https://www.droppedneedle.com)"
+DEFAULT_EXTERNAL_USER_AGENT = "DroppedNeedleApp/1.0 (contact@droppedneedle.com; https://www.droppedneedle.com)"
 
 
 class TransientImageFetchError(Exception):
diff --git a/backend/repositories/musicbrainz_album.py b/backend/repositories/musicbrainz_album.py
index 309324438..a82d59ae7 100644
--- a/backend/repositories/musicbrainz_album.py
+++ b/backend/repositories/musicbrainz_album.py
@@ -571,11 +571,11 @@ async def _fetch_release_group_by_id(
                 return None
             await self._cache.set(cache_key, result, ttl_seconds=3600)
             return result
-        except Exception as e:  # noqa: BLE001
-            if not isinstance(e, CircuitOpenError):
-                logger.error(f"Failed to fetch release group {mbid}: {e}")
-            _record_mb_degradation(f"release group fetch failed: {e}")
-            return None
+        except (httpx.HTTPError, CircuitOpenError, ExternalServiceError) as error:
+            _record_mb_degradation("release group metadata temporarily unavailable")
+            raise ExternalServiceError(
+                "MusicBrainz album metadata is temporarily unavailable. Try again later."
+            ) from error
 
     async def get_release_group(self, release_group_mbid: str) -> AlbumInfo | None:
         """Fetch a release group and map it to ``AlbumInfo`` (the
@@ -874,11 +874,13 @@ async def _fetch_release_by_id(
                 return None
             await self._cache.set(cache_key, result, ttl_seconds=3600)
             return result
-        except Exception as e:  # noqa: BLE001
-            if not isinstance(e, CircuitOpenError):
-                logger.error(f"Failed to fetch release {release_id}: {e}")
-            _record_mb_degradation(f"release fetch failed: {e}")
-            return None
+        except (httpx.HTTPError, CircuitOpenError, ExternalServiceError) as error:
+            _record_mb_degradation("release metadata temporarily unavailable")
+            # A network outage is not proof that the selected edition does not
+            # exist. Preserve the retryable service failure for acquisition.
+            raise ExternalServiceError(
+                "MusicBrainz release metadata is temporarily unavailable. Try again later."
+            ) from error
 
     async def get_release_group_id_from_release(
         self,
diff --git a/backend/scripts/update_management_genres.py b/backend/scripts/update_management_genres.py
index 4e06c0bf9..b11d12c6d 100644
--- a/backend/scripts/update_management_genres.py
+++ b/backend/scripts/update_management_genres.py
@@ -12,6 +12,7 @@
 import json
 from pathlib import Path
 from urllib.request import Request, urlopen
+from core.config import get_settings
 
 SOURCE_URL = "https://musicbrainz.org/ws/2/genre/all?fmt=txt"
 
@@ -30,12 +31,7 @@ def main() -> None:
     existing = json.loads(args.asset.read_text(encoding="utf-8"))
     request = Request(
         SOURCE_URL,
-        headers={
-            "User-Agent": (
-                "DroppedNeedle/LibraryManagement "
-                "(https://github.com/DroppedNeedle/DroppedNeedle)"
-            )
-        },
+        headers={"User-Agent": get_settings().get_user_agent()},
     )
     with urlopen(request, timeout=30) as response:  # noqa: S310 - fixed HTTPS URL
         names = sorted(
diff --git a/backend/services/acquisition_dispatcher.py b/backend/services/acquisition_dispatcher.py
index a8d581c4a..526b72590 100644
--- a/backend/services/acquisition_dispatcher.py
+++ b/backend/services/acquisition_dispatcher.py
@@ -96,6 +96,7 @@ async def request_album(
         release_mbid: str | None = None,
         release_track_mbid: str | None = None,
         track_count_priority: RequestPriority = RequestPriority.USER_INITIATED,
+        content_variant: str = "original",
     ) -> str:
         if self._ownership is not None:
             release_group_mbid = await self._ownership.provider_album_id(
@@ -134,6 +135,7 @@ async def request_album(
             origin=origin,
             release_mbid=release_mbid,
             release_track_mbid=release_track_mbid,
+            content_variant=content_variant,
         )
 
     async def request_track(
@@ -151,6 +153,7 @@ async def request_track(
         release_track_mbid: str | None = None,
         track_number: int | None = None,
         disc_number: int | None = None,
+        content_variant: str = "original",
     ) -> str:
         if self._ownership is not None:
             recording_mbid = await self._ownership.provider_track_id(recording_mbid)
@@ -161,6 +164,10 @@ async def request_track(
             if artist_mbid is not None:
                 artist_mbid = await self._ownership.provider_artist_id(artist_mbid)
         if self._use_free_music():
+            if content_variant == "clean":
+                raise ProviderIdentityRequiredError(
+                    "Clean-only requests require DroppedNeedle's verified native acquisition backend."
+                )
             if origin != "edition_conversion":
                 return await self._get_free_music_service().request_track(
                     user_id=user_id,
@@ -194,4 +201,5 @@ async def request_track(
             origin=origin,
             release_mbid=release_mbid,
             release_track_mbid=release_track_mbid,
+            content_variant=content_variant,
         )
diff --git a/backend/services/auth_service.py b/backend/services/auth_service.py
index 771b3abb2..678b9b9ca 100644
--- a/backend/services/auth_service.py
+++ b/backend/services/auth_service.py
@@ -439,6 +439,32 @@ async def revoke_user_sessions(self, user_id: str) -> None:
     async def list_sessions(self, user_id: str) -> list[TokenRecord]:
         return await self._store.list_tokens_for_user(user_id)
 
+    @staticmethod
+    def validate_device_session_name(device_name: str) -> str:
+        label = " ".join((device_name or "").split())
+        if not label or len(label) > 80:
+            raise AuthenticationError("Invalid device name")
+        return label
+
+    async def issue_device_session(self, user_id: str, device_name: str) -> str:
+        """Create a distinct, revocable session without copying the caller's token.
+
+        The authenticated caller already proved ownership of the account. The
+        bounded label is stored only as the session user-agent so the existing
+        session roster can identify and revoke the companion independently.
+        """
+        label = self.validate_device_session_name(device_name)
+        await self._require_user(user_id)
+        user_agent = f"Tonarr companion · {label}"
+        raw_token, token_hash = self._store.issue_token()
+        await self._store.replace_companion_token(
+            id=_new_id(),
+            user_id=user_id,
+            token_hash=token_hash,
+            user_agent=user_agent,
+        )
+        return raw_token
+
     async def revoke_session(self, token_id: str, requesting_user_id: str) -> None:
         tokens = await self._store.list_tokens_for_user(requesting_user_id)
         owned = any(token.id == token_id for token in tokens)
diff --git a/backend/services/native/acquisition/strategy.py b/backend/services/native/acquisition/strategy.py
index 2618a70e2..fcaa7276e 100644
--- a/backend/services/native/acquisition/strategy.py
+++ b/backend/services/native/acquisition/strategy.py
@@ -15,6 +15,7 @@
 
 import asyncio
 import logging
+import re
 import time
 from pathlib import Path
 from typing import Protocol, runtime_checkable
@@ -45,6 +46,33 @@
 # skip (blocklisted regardless of age, since propagation can't fix encryption).
 _PASSWORD_MARKERS = ("password", "passworded", "encrypt")
 
+_EXPLICIT_EDITION_MARKER = re.compile(
+    r"(?<![a-z0-9])(explicit|uncensored|unrated)(?![a-z0-9])",
+    re.IGNORECASE,
+)
+
+
+def _clean_candidates(task, candidates):  # noqa: ANN001, ANN201
+    """Drop candidates that explicitly advertise the wrong lyrical edition.
+
+    This improves source selection only. Every clean import is still fingerprinted
+    against the exact requested MusicBrainz recording before it can be published.
+    """
+    if task.content_variant != "clean":
+        return candidates
+    kept = []
+    for candidate in candidates:
+        if candidate.source == "usenet" and candidate.usenet_release is not None:
+            evidence = candidate.usenet_release.title
+        else:
+            evidence = " ".join(
+                [candidate.parent_directory]
+                + [entry.filename for entry in candidate.files]
+            )
+        if not _EXPLICIT_EDITION_MARKER.search(evidence):
+            kept.append(candidate)
+    return kept
+
 
 async def _upgrade_held_tier(library, task) -> "str | None":  # noqa: ANN001
     """The held library tier an ``origin='upgrade'`` task must strictly beat, resolved
@@ -290,13 +318,14 @@ async def search_and_score(self, task, *, timeout, auto, manual):  # noqa: ANN00
                 timeout=timeout,
             )
             results = [r.soulseek for r in indexer_results if r.soulseek is not None]
-            return await self._track_matcher.rank(
+            ranked = await self._track_matcher.rank(
                 target,
                 results,
                 auto_accept_threshold=auto,
                 manual_threshold=manual,
                 held_tier=held_tier,
             )
+            return _clean_candidates(task, ranked)
         # A 1-track release (a single requested as an album) scores per-file via the
         # track matcher, not the folder scorer: folder coherence hands a lone
         # fuzzy-matched file a perfect count_ratio, and only the per-file path carries
@@ -321,13 +350,14 @@ async def search_and_score(self, task, *, timeout, auto, manual):  # noqa: ANN00
                 timeout=timeout,
             )
             results = [r.soulseek for r in indexer_results if r.soulseek is not None]
-            return await self._track_matcher.rank(
+            ranked = await self._track_matcher.rank(
                 target,
                 results,
                 auto_accept_threshold=auto,
                 manual_threshold=manual,
                 held_tier=held_tier,
             )
+            return _clean_candidates(task, ranked)
         target = TargetAlbum(
             artist_name=task.artist_name,
             album_title=task.album_title,
@@ -343,13 +373,14 @@ async def search_and_score(self, task, *, timeout, auto, manual):  # noqa: ANN00
             timeout=timeout,
         )
         results = [r.soulseek for r in indexer_results if r.soulseek is not None]
-        return await self._scorer.rank(
+        ranked = await self._scorer.rank(
             target,
             results,
             auto_accept_threshold=auto,
             manual_threshold=manual,
             held_tier=held_tier,
         )
+        return _clean_candidates(task, ranked)
 
     async def enqueue(
         self, task, candidate, *, strict_track_duration, hold_on_wrong_track=False
@@ -411,6 +442,7 @@ async def enqueue(
             source_username=candidate.username,
             handle=initial_handle,
             origin=task.origin,
+            content_variant=task.content_variant,
             release_group_mbid=task.release_group_mbid,
             release_mbid=release_mbid,
             artist_mbid=task.artist_mbid,
@@ -651,7 +683,7 @@ async def search_and_score(self, task, *, timeout, auto, manual):  # noqa: ANN00
             timeout=timeout,
         )
         releases = [r.usenet for r in indexer_results if r.usenet is not None]
-        return await self._scorer.rank(
+        ranked = await self._scorer.rank(
             target,
             releases,
             auto_accept_threshold=auto,
@@ -659,6 +691,7 @@ async def search_and_score(self, task, *, timeout, auto, manual):  # noqa: ANN00
             track_count=task.track_count,
             held_tier=held_tier,
         )
+        return _clean_candidates(task, ranked)
 
     async def enqueue(
         self, task, candidate, *, strict_track_duration, hold_on_wrong_track=False
@@ -705,6 +738,7 @@ async def enqueue(
             task_id=task.id,
             handle=initial_handle,
             origin=task.origin,
+            content_variant=task.content_variant,
             release_group_mbid=task.release_group_mbid,
             release_mbid=release_mbid,
             artist_mbid=task.artist_mbid,
diff --git a/backend/services/native/download_orchestrator.py b/backend/services/native/download_orchestrator.py
index 2e3f15042..35d2bbe5f 100644
--- a/backend/services/native/download_orchestrator.py
+++ b/backend/services/native/download_orchestrator.py
@@ -2169,6 +2169,8 @@ async def _reimport_task_locked(self, task_id: str):  # noqa: ANN201
         manifest = DownloadManifest(
             task_id=task.id,
             source_username=candidate.username,
+            origin=task.origin,
+            content_variant=task.content_variant,
             release_group_mbid=task.release_group_mbid,
             release_mbid=release_mbid,
             artist_mbid=task.artist_mbid,
diff --git a/backend/services/native/download_service.py b/backend/services/native/download_service.py
index 044cc878d..197b81a10 100644
--- a/backend/services/native/download_service.py
+++ b/backend/services/native/download_service.py
@@ -17,6 +17,7 @@
 from core.exceptions import (
     AutomaticManagementHoldError,
     ConfigurationError,
+    ExternalServiceError,
     PermissionDeniedError,
     ResourceNotFoundError,
     ValidationError,
@@ -323,6 +324,10 @@ async def _resolve_acquisition_identity(
                     release_group_mbid,
                     priority=priority,
                 )
+        except ExternalServiceError:
+            # Preserve a retryable catalog outage; it says nothing about whether
+            # the exact edition exists. No task is created before identity resolves.
+            raise
         except Exception as error:  # noqa: BLE001 - fail closed before any task exists
             raise ValidationError(
                 "The exact MusicBrainz edition could not be verified. No download was started."
@@ -721,6 +726,7 @@ async def request_album(
         origin: str = "user",
         release_mbid: str | None = None,
         release_track_mbid: str | None = None,
+        content_variant: str = "original",
     ) -> str:
         """Create a download task and dispatch the orchestrator. Returns the new
         task id, the existing active task id (dedup), or the ``already_in_library``
@@ -877,6 +883,7 @@ async def request_album(
             track_count=track_count,
             track_duration_seconds=track_duration_seconds,
             origin=origin,
+            content_variant=content_variant,
         )
         self._orchestrator.dispatch(task.id)
         return task.id
@@ -894,10 +901,23 @@ async def request_track(
         origin: str = "user",
         release_mbid: str | None = None,
         release_track_mbid: str | None = None,
+        content_variant: str = "original",
     ) -> str:
         """Request a single track. Orphan tracks (album not in the library) resolve
         the release group via MusicBrainz, auto-create the album folder, and download
         the one track; the album appears partially present."""
+        if content_variant == "clean":
+            if not recording_mbid or not release_group_mbid or not release_mbid:
+                raise ValidationError(
+                    "A clean request requires an exact MusicBrainz recording, "
+                    "release group, and release edition"
+                )
+            # This is server-owned replacement semantics; clients cannot select an
+            # arbitrary origin string to obtain destructive behavior.
+            origin = "clean_replacement"
+        elif content_variant != "original":
+            raise ValidationError("Unsupported content variant")
+
         if self._ownership is not None:
             recording_mbid = await self._ownership.provider_track_id(recording_mbid)
             if release_group_mbid is not None:
@@ -961,6 +981,7 @@ async def request_track(
             origin=origin,
             release_mbid=release_mbid,
             release_track_mbid=release_track_mbid,
+            content_variant=content_variant,
         )
 
     @property
diff --git a/backend/services/native/file_processor.py b/backend/services/native/file_processor.py
index d260bc452..2ca015f82 100644
--- a/backend/services/native/file_processor.py
+++ b/backend/services/native/file_processor.py
@@ -1246,7 +1246,7 @@ async def _place_matched_file(
                 target_tag.track_number,
             )
             occupied_by_other = (
-                manifest.origin != "upgrade"
+                manifest.origin not in {"upgrade", "clean_replacement"}
                 and present is not None
                 and not row_covers_track(
                     present,
@@ -1274,38 +1274,42 @@ async def _place_matched_file(
                 replacement = present
 
         fp = None
-        conversion_verification = manifest.origin == "edition_conversion"
-        if conversion_verification and self._fingerprinter is None:
+        exact_recording_verification = (
+            manifest.origin == "edition_conversion"
+            or manifest.content_variant == "clean"
+        )
+        if exact_recording_verification and self._fingerprinter is None:
             raise VerificationFailed(
-                "Recording verification is unavailable for this edition conversion",
+                "Exact recording verification is unavailable for this request",
                 reason="fingerprint_unavailable",
                 filename=source.name,
             )
         if (
-            self._verify_downloads or conversion_verification
+            self._verify_downloads or exact_recording_verification
         ) and self._fingerprinter is not None:
             fp = await self._fingerprinter.fingerprint(source)
             if _fingerprint_disagrees(fp, track, manifest.artist_name):
-                await self._hold_for_review(
-                    source=source,
-                    manifest=manifest,
-                    reason="fingerprint_mismatch",
-                    evidence_title=getattr(fp, "title", None),
-                    evidence_artist=getattr(fp, "artist", None),
-                    evidence_score=getattr(fp, "score", None),
-                    track_number=track.track_number,
-                    disc_number=track.disc_number or 1,
-                    track_title=track.title,
-                    recording_mbid=track.recording_mbid,
-                    duration_seconds=info.duration_seconds,
-                    file_format=info.file_format,
-                )
+                if manifest.content_variant != "clean":
+                    await self._hold_for_review(
+                        source=source,
+                        manifest=manifest,
+                        reason="fingerprint_mismatch",
+                        evidence_title=getattr(fp, "title", None),
+                        evidence_artist=getattr(fp, "artist", None),
+                        evidence_score=getattr(fp, "score", None),
+                        track_number=track.track_number,
+                        disc_number=track.disc_number or 1,
+                        track_title=track.title,
+                        recording_mbid=track.recording_mbid,
+                        duration_seconds=info.duration_seconds,
+                        file_format=info.file_format,
+                    )
                 raise VerificationFailed(
                     "AcoustID identified a different recording",
                     reason="fingerprint_mismatch",
                     filename=source.name,
                 )
-            if conversion_verification and (
+            if exact_recording_verification and (
                 not track.recording_mbid
                 or getattr(fp, "status", None) != "pass"
                 or (getattr(fp, "recording_id", None) or "").casefold()
@@ -1364,6 +1368,14 @@ def _position_upgrade_target(
         """The occupied slot's old file path when this upgrade import may replace it
         (strictly better + a recycle bin to preserve the old bytes), else ``None``
         (the caller keeps the existing file - today's dedup behaviour)."""
+        if origin == "clean_replacement":
+            if self._recycle_bin is None:
+                raise VerificationFailed(
+                    "Clean replacement requires a configured recycle bin",
+                    reason="replacement_unavailable",
+                    filename=Path(present["file_path"]).name,
+                )
+            return Path(present["file_path"])
         if origin != "upgrade" or self._recycle_bin is None:
             return None
         if _is_strict_upgrade(_row_tier(present), info):
@@ -1387,6 +1399,14 @@ async def _existing_tier_at(self, path: Path) -> str | None:
     async def _same_path_upgrade_applies(
         self, origin: str, target_path: Path, info: AudioInfo
     ) -> bool:
+        if origin == "clean_replacement":
+            if self._recycle_bin is None:
+                raise VerificationFailed(
+                    "Clean replacement requires a configured recycle bin",
+                    reason="replacement_unavailable",
+                    filename=target_path.name,
+                )
+            return True
         if origin != "upgrade" or self._recycle_bin is None:
             return False
         existing_tier = await self._existing_tier_at(target_path)
@@ -1839,7 +1859,7 @@ async def _process_one(
                 target_tag.track_number,
             )
             occupied_by_other = (
-                manifest.origin != "upgrade"
+                manifest.origin not in {"upgrade", "clean_replacement"}
                 and expected_track is not None
                 and present is not None
                 and not row_covers_track(
@@ -1891,7 +1911,7 @@ async def _process_one(
             # A relaxed re-pull (every candidate failed this gate - the MB length is
             # suspect) captures the closest match for HUMAN review instead of
             # importing it silently with the gate off (D9) or discarding it.
-            if manifest.hold_on_wrong_track:
+            if manifest.hold_on_wrong_track and manifest.content_variant != "clean":
                 await self._hold_for_review(
                     source=source,
                     manifest=manifest,
@@ -1923,24 +1943,25 @@ async def _process_one(
         if self._verify_downloads and _tag_conflict_reason(
             tag, info, manifest, expected_track
         ):
-            await self._hold_for_review(
-                source=source,
-                manifest=manifest,
-                reason="tag_mismatch",
-                evidence_title=tag.title,
-                evidence_artist=tag.artist,
-                evidence_score=None,
-                track_number=tag.track_number,
-                disc_number=tag.disc_number or 1,
-                track_title=(expected_track.title if expected_track else tag.title),
-                recording_mbid=(
-                    expected_track.recording_mbid
-                    if expected_track
-                    else tag.musicbrainz_recording_id
-                ),
-                duration_seconds=info.duration_seconds,
-                file_format=info.file_format,
-            )
+            if manifest.content_variant != "clean":
+                await self._hold_for_review(
+                    source=source,
+                    manifest=manifest,
+                    reason="tag_mismatch",
+                    evidence_title=tag.title,
+                    evidence_artist=tag.artist,
+                    evidence_score=None,
+                    track_number=tag.track_number,
+                    disc_number=tag.disc_number or 1,
+                    track_title=(expected_track.title if expected_track else tag.title),
+                    recording_mbid=(
+                        expected_track.recording_mbid
+                        if expected_track
+                        else tag.musicbrainz_recording_id
+                    ),
+                    duration_seconds=info.duration_seconds,
+                    file_format=info.file_format,
+                )
             raise VerificationFailed(
                 "File tags name different content than requested",
                 reason="tag_mismatch",
@@ -1953,38 +1974,42 @@ async def _process_one(
         # different ARTIST is rejected. NOT a release-group check - that false-rejects
         # valid reissue/compilation tracks whose AcoustID RG coverage is incomplete.
         fp = None
-        conversion_verification = manifest.origin == "edition_conversion"
-        if conversion_verification and self._fingerprinter is None:
+        exact_recording_verification = (
+            manifest.origin == "edition_conversion"
+            or manifest.content_variant == "clean"
+        )
+        if exact_recording_verification and self._fingerprinter is None:
             raise VerificationFailed(
-                "Recording verification is unavailable for this edition conversion",
+                "Exact recording verification is unavailable for this request",
                 reason="fingerprint_unavailable",
                 filename=expected.filename,
             )
         if (
-            self._verify_downloads or conversion_verification
+            self._verify_downloads or exact_recording_verification
         ) and self._fingerprinter is not None:
             fp = await self._fingerprinter.fingerprint(source)
             if _fingerprint_disagrees(fp, expected_track, manifest.artist_name):
-                await self._hold_for_review(
-                    source=source,
-                    manifest=manifest,
-                    reason="fingerprint_mismatch",
-                    evidence_title=getattr(fp, "title", None),
-                    evidence_artist=getattr(fp, "artist", None),
-                    evidence_score=getattr(fp, "score", None),
-                    track_number=tag.track_number,
-                    disc_number=tag.disc_number or 1,
-                    track_title=tag.title,
-                    recording_mbid=tag.musicbrainz_recording_id,
-                    duration_seconds=info.duration_seconds,
-                    file_format=info.file_format,
-                )
+                if manifest.content_variant != "clean":
+                    await self._hold_for_review(
+                        source=source,
+                        manifest=manifest,
+                        reason="fingerprint_mismatch",
+                        evidence_title=getattr(fp, "title", None),
+                        evidence_artist=getattr(fp, "artist", None),
+                        evidence_score=getattr(fp, "score", None),
+                        track_number=tag.track_number,
+                        disc_number=tag.disc_number or 1,
+                        track_title=tag.title,
+                        recording_mbid=tag.musicbrainz_recording_id,
+                        duration_seconds=info.duration_seconds,
+                        file_format=info.file_format,
+                    )
                 raise VerificationFailed(
                     "AcoustID identified a different recording",
                     reason="fingerprint_mismatch",
                     filename=expected.filename,
                 )
-            if conversion_verification and (
+            if exact_recording_verification and (
                 expected_track is None
                 or not expected_track.recording_mbid
                 or getattr(fp, "status", None) != "pass"
diff --git a/backend/services/native/target_library_repository.py b/backend/services/native/target_library_repository.py
index a9bcf261b..2507779e2 100644
--- a/backend/services/native/target_library_repository.py
+++ b/backend/services/native/target_library_repository.py
@@ -146,6 +146,15 @@ async def get_artist_mbid_page(self, *, after_mbid: str, limit: int) -> list[str
         )
         return [mbid for mbid in mbids if mbid > cursor][: max(1, limit)]
 
+    async def get_enrichment_candidates(
+        self, *, after_mbid: str | None, limit: int
+    ) -> list[tuple[str, str, dict[str, Any]]]:
+        """Page live provider identities for background metadata enrichment."""
+        return await self._store.target_enrichment_candidates(
+            after_mbid=after_mbid,
+            limit=limit,
+        )
+
     async def existing_album_mbids(self, identifiers: list[str]) -> set[str]:
         normalized = {
             value.strip().casefold() for value in identifiers if value.strip()
diff --git a/backend/services/precache/audiodb_phase.py b/backend/services/precache/audiodb_phase.py
index 00365d61b..68d53bb96 100644
--- a/backend/services/precache/audiodb_phase.py
+++ b/backend/services/precache/audiodb_phase.py
@@ -8,6 +8,7 @@
 
 import httpx
 
+from core.config import get_settings
 from repositories.protocols import CoverArtRepositoryProtocol
 from repositories.coverart_disk_cache import get_cache_filename, VALID_IMAGE_CONTENT_TYPES
 from services.cache_status_service import CacheStatusService
@@ -102,7 +103,7 @@ async def download_bytes(self, url: str, entity_type: str, mbid: str) -> bool:
                 else:
                     logger.debug("audiodb.prewarm action=http_client_fallback entity_type=%s mbid=%s", entity_type, mbid[:8])
                     async with httpx.AsyncClient(timeout=httpx.Timeout(15.0, connect=5.0)) as client:
-                        response = await client.get(url, headers={"User-Agent": "DroppedNeedle/1.0"}, follow_redirects=True)
+                        response = await client.get(url, headers={"User-Agent": get_settings().get_user_agent()}, follow_redirects=True)
 
             if response.status_code != 200:
                 logger.debug(
diff --git a/backend/services/quota_service.py b/backend/services/quota_service.py
index 41eb0c1a2..af33e0647 100644
--- a/backend/services/quota_service.py
+++ b/backend/services/quota_service.py
@@ -4,9 +4,9 @@
 
 - **Layer 1, request-count quota, at submit.** A plain user's ask is recorded in
   ``request_history`` long before any download task exists, so the count gate runs
-  where the ask is made: ``RequestService.request_album``/``request_batch`` and the
-  per-track request route (tracks bypass approval, so their ``download_tasks`` row
-  is the ask). Rolling window (D9); pending asks count.
+  where the ask is made: ``RequestService.request_album``/``request_batch``/
+  ``request_track``. Exact tracks share the same approval gate as albums. Rolling
+  window (D9); pending asks count.
 
 - **Layer 2, byte caps, at every download-task-creation site.** The global library
   cap (all roles) and the per-user storage quota apply when bytes will actually be
diff --git a/backend/services/request_service.py b/backend/services/request_service.py
index 01a53c2d7..244eac2b0 100644
--- a/backend/services/request_service.py
+++ b/backend/services/request_service.py
@@ -9,6 +9,7 @@
     BatchRequestResponse,
     RequestAcceptedResponse,
 )
+from api.v1.schemas.download import TrackRequestResponse
 from core.exceptions import ExternalServiceError, ValidationError
 from infrastructure.queue.priority_queue import RequestPriority
 from services.native.download_service import ALREADY_IN_LIBRARY
@@ -123,12 +124,18 @@ async def request_album(
             musicbrainz_id
         )
 
-        needs_approval = user_role == "user"
+        # Fail closed for any future or malformed role. Only the two roles the
+        # server explicitly grants acquisition authority may dispatch without
+        # owner review.
+        needs_approval = user_role not in ("trusted", "admin")
         initial_status = "awaiting_approval" if needs_approval else "pending"
 
         try:
             existing = await self._request_history.async_get_record(musicbrainz_id)
             if existing and existing.status in ("pending", "downloading"):
+                await self._request_history.async_add_requester(
+                    musicbrainz_id, user_id, requested_by_name
+                )
                 if monitor_artist and not existing.monitor_artist:
                     await self._request_history.async_update_monitoring_flags(
                         musicbrainz_id,
@@ -142,6 +149,9 @@ async def request_album(
                     status=existing.status,
                 )
             if existing and existing.status == "awaiting_approval":
+                await self._request_history.async_add_requester(
+                    musicbrainz_id, user_id, requested_by_name
+                )
                 return RequestAcceptedResponse(
                     success=True,
                     message="Request is awaiting admin approval",
@@ -250,6 +260,115 @@ async def request_album(
             status="pending",
         )
 
+    async def request_track(
+        self,
+        recording_mbid: str,
+        *,
+        artist_name: str,
+        track_title: str,
+        album_title: str | None = None,
+        duration_seconds: int | None = None,
+        release_group_mbid: str | None = None,
+        artist_mbid: str | None = None,
+        release_mbid: str | None = None,
+        content_variant: str = "original",
+        user_id: str,
+        user_role: str,
+        requested_by_name: str | None = None,
+    ) -> TrackRequestResponse:
+        """Record an exact-track ask before dispatching it.
+
+        This deliberately shares the album approval gate. A normal user can
+        request one recording, but only a trusted user or an owner can start
+        acquisition without review. The previous route bypassed approval and
+        made exact-track requests less safe than whole-album requests.
+        """
+        if content_variant not in ("original", "clean"):
+            raise ValidationError("Unknown track content variant")
+        needs_approval = user_role not in ("trusted", "admin")
+        existing = await self._request_history.async_get_record(recording_mbid)
+        if existing and getattr(existing, "content_variant", "original") != content_variant:
+            raise ValidationError("This recording already has a different content request. Choose a verified distinct recording.")
+        if existing and existing.status in (
+            "awaiting_approval",
+            "pending",
+            "queued",
+            "downloading",
+        ):
+            await self._request_history.async_add_requester(
+                recording_mbid, user_id, requested_by_name
+            )
+            return TrackRequestResponse(
+                status=(
+                    "awaiting_approval"
+                    if existing.status == "awaiting_approval"
+                    else "queued"
+                ),
+                task_id=existing.download_task_id,
+            )
+
+        if self._quota is not None:
+            await self._quota.check_request_quota(user_id, user_role)
+            await self._quota.check_storage_admission(user_id, "user")
+
+        await self._request_history.async_record_request(
+            musicbrainz_id=recording_mbid,
+            artist_name=artist_name or "Unknown",
+            album_title=album_title or "Single track",
+            artist_mbid=artist_mbid,
+            user_id=user_id,
+            requested_by_name=requested_by_name,
+            release_mbid=release_mbid,
+            initial_status="awaiting_approval" if needs_approval else "pending",
+            request_kind="track",
+            track_title=track_title,
+            duration_seconds=duration_seconds,
+            track_release_group_mbid=release_group_mbid,
+            content_variant=content_variant,
+        )
+
+        if needs_approval:
+            logger.info(
+                "Exact-track request queued for approval: %s by user %s",
+                recording_mbid,
+                user_id,
+            )
+            return TrackRequestResponse(status="awaiting_approval")
+
+        try:
+            task_id = await self._acquisition.request_track(
+                user_id=user_id,
+                recording_mbid=recording_mbid,
+                artist_name=artist_name,
+                track_title=track_title,
+                album_title=album_title,
+                duration_seconds=duration_seconds,
+                release_group_mbid=release_group_mbid,
+                artist_mbid=artist_mbid,
+                release_mbid=release_mbid,
+                content_variant=content_variant,
+            )
+        except Exception:
+            await self._request_history.async_update_status(
+                recording_mbid,
+                "failed",
+                completed_at=datetime.now(timezone.utc).isoformat(),
+            )
+            raise
+
+        if task_id == ALREADY_IN_LIBRARY:
+            await self._request_history.async_update_status(
+                recording_mbid,
+                "imported",
+                completed_at=datetime.now(timezone.utc).isoformat(),
+            )
+            return TrackRequestResponse(status="already_in_library")
+
+        await self._request_history.async_update_download_task_id(
+            recording_mbid, task_id
+        )
+        return TrackRequestResponse(status="queued", task_id=task_id)
+
     async def request_batch(
         self,
         items: list[dict],
@@ -282,7 +401,7 @@ async def request_batch(
             seen_mbids.add(canonical_key)
             items.append(item)
 
-        needs_approval = user_role == "user"
+        needs_approval = user_role not in ("trusted", "admin")
         initial_status = "awaiting_approval" if needs_approval else "pending"
 
         try:
@@ -290,6 +409,14 @@ async def request_batch(
             new_items = [
                 item for item in items if item["musicbrainz_id"].lower() not in active
             ]
+            existing_items = [
+                item["musicbrainz_id"]
+                for item in items
+                if item["musicbrainz_id"].lower() in active
+            ]
+            await self._request_history.async_add_requesters(
+                existing_items, user_id, requested_by_name
+            )
             skipped = duplicate_count + len(items) - len(new_items)
 
             if not new_items:
@@ -298,6 +425,7 @@ async def request_batch(
                     message="All albums already requested",
                     requested=0,
                     skipped=skipped,
+                    status="already_requested",
                 )
 
             # A batch of N counts as N asks (A4); over-quota rejects the WHOLE batch
@@ -324,6 +452,7 @@ async def request_batch(
                     message="Batch request submitted, awaiting admin approval",
                     requested=len(new_items),
                     skipped=skipped,
+                    status="awaiting_approval",
                 )
 
             # auto-approve: dispatch each item through the native pipeline (mirrors
@@ -367,6 +496,7 @@ async def request_batch(
                 requested=dispatched,
                 skipped=skipped,
                 overflow=0,
+                status="pending" if dispatched else "failed",
             )
         except (ExternalServiceError, ValidationError):
             raise
@@ -388,9 +518,21 @@ async def cancel_batch(
         for mbid in musicbrainz_ids:
             try:
                 record = await self._request_history.async_get_record(mbid)
-                if not is_admin and (record is None or record.user_id != user_id):
-                    failed += 1
-                    continue
+                if not is_admin:
+                    if (
+                        record is None
+                        or not await self._request_history.async_is_requester(
+                            user_id or "", mbid
+                        )
+                    ):
+                        failed += 1
+                        continue
+                    if await self._request_history.async_requester_count(mbid) > 1:
+                        await self._request_history.async_remove_requester(
+                            user_id or "", mbid
+                        )
+                        cancelled += 1
+                        continue
                 # best-effort: a missing/non-cancellable task must not block marking
                 if record is not None and record.download_task_id:
                     try:
diff --git a/backend/services/requests_page_service.py b/backend/services/requests_page_service.py
index 793c25eeb..ae088367a 100644
--- a/backend/services/requests_page_service.py
+++ b/backend/services/requests_page_service.py
@@ -141,6 +141,10 @@ async def get_request_history(
                 download_task_id=r.download_task_id,
                 can_reimport=r.status == "failed"
                 and r.download_task_id in reimportable,
+                request_kind=r.request_kind,
+                track_title=r.track_title,
+                duration_seconds=r.duration_seconds,
+                track_release_group_mbid=r.track_release_group_mbid,
             )
             for r in records
         ]
@@ -181,17 +185,7 @@ async def approve_request(
         # (the 'already_in_library' sentinel is guarded)
         if self._acquisition is not None:
             try:
-                task_id = await self._acquisition.request_album(
-                    user_id=record.user_id or "",
-                    release_group_mbid=musicbrainz_id,
-                    artist_name=record.artist_name or "Unknown",
-                    album_title=record.album_title or "Unknown",
-                    year=record.year,
-                    artist_mbid=record.artist_mbid,
-                    origin="user",
-                    release_mbid=record.release_mbid,
-                    track_count_priority=RequestPriority.USER_INITIATED,
-                )
+                task_id = await self._dispatch_record(record, origin="user")
             except ValidationError as e:
                 # A cap/quota rejection (Feature C) is not a failure of the request:
                 # put it BACK in the approval queue (it would otherwise silently
@@ -211,7 +205,7 @@ async def approve_request(
                 )
                 return CancelRequestResponse(
                     success=False,
-                    message=f"Approved but failed to start: {record.album_title}",
+                    message=f"Approved but failed to start: {self._record_title(record)}",
                 )
             from services.native.download_service import ALREADY_IN_LIBRARY
 
@@ -219,8 +213,14 @@ async def approve_request(
                 await self._request_history.async_update_download_task_id(
                     musicbrainz_id, task_id
                 )
+            else:
+                await self._request_history.async_update_status(
+                    musicbrainz_id,
+                    "imported",
+                    completed_at=datetime.now(timezone.utc).isoformat(),
+                )
         return CancelRequestResponse(
-            success=True, message=f"Approved: {record.album_title}"
+            success=True, message=f"Approved: {self._record_title(record)}"
         )
 
     async def reject_request(
@@ -239,7 +239,7 @@ async def reject_request(
             musicbrainz_id, "rejected", reviewer_id, reviewer_name, completed_at=now_iso
         )
         return CancelRequestResponse(
-            success=True, message=f"Rejected: {record.album_title}"
+            success=True, message=f"Rejected: {self._record_title(record)}"
         )
 
     async def cancel_request(
@@ -248,8 +248,22 @@ async def cancel_request(
         record = await self._request_history.async_get_record(musicbrainz_id)
         if not record:
             return CancelRequestResponse(success=False, message="Request not found")
-        if user_role != "admin" and record.user_id != user_id:
-            raise PermissionDeniedError("Cannot cancel another user's request")
+        if user_role != "admin":
+            if not await self._request_history.async_is_requester(
+                user_id, musicbrainz_id
+            ):
+                raise PermissionDeniedError("Cannot cancel another user's request")
+            if await self._request_history.async_requester_count(musicbrainz_id) > 1:
+                await self._request_history.async_remove_requester(
+                    user_id, musicbrainz_id
+                )
+                return CancelRequestResponse(
+                    success=True,
+                    message=(
+                        "Removed from your requests. The shared server request "
+                        "continues for another listener."
+                    ),
+                )
 
         # awaiting_approval requests never dispatched, cancel directly
         if record.status == "awaiting_approval":
@@ -259,7 +273,7 @@ async def cancel_request(
             )
             return CancelRequestResponse(
                 success=True,
-                message=f"Cancelled request for {record.album_title}",
+                message=f"Cancelled request for {self._record_title(record)}",
             )
 
         if record.status not in _CANCELLABLE_STATUSES:
@@ -293,7 +307,7 @@ async def cancel_request(
 
         return CancelRequestResponse(
             success=True,
-            message=f"Cancelled download of {record.album_title}",
+            message=f"Cancelled download of {self._record_title(record)}",
         )
 
     async def retry_request(
@@ -302,7 +316,9 @@ async def retry_request(
         record = await self._request_history.async_get_record(musicbrainz_id)
         if not record:
             return RetryRequestResponse(success=False, message="Request not found")
-        if user_role != "admin" and record.user_id != user_id:
+        if user_role != "admin" and not await self._request_history.async_is_requester(
+            user_id, musicbrainz_id
+        ):
             raise PermissionDeniedError("Cannot retry another user's request")
 
         if record.status not in _RETRYABLE_STATUSES:
@@ -319,16 +335,11 @@ async def retry_request(
             await self._request_history.async_update_status(musicbrainz_id, "pending")
             # A retry re-dispatches an already-recorded ask, so it is not a new
             # user request for quota purposes (CollectionManagement D20).
-            task_id = await self._acquisition.request_album(
-                user_id=record.user_id or user_id or "",
-                release_group_mbid=musicbrainz_id,
-                artist_name=record.artist_name or "Unknown",
-                album_title=record.album_title or "Unknown",
-                year=record.year,
-                artist_mbid=record.artist_mbid,
+            task_id = await self._dispatch_record(
+                record,
                 origin="retry",
-                release_mbid=record.release_mbid,
-                track_count_priority=RequestPriority.USER_INITIATED,
+                fallback_user_id=user_id,
+                user_id_override=user_id if user_role != "admin" else None,
             )
         except ValidationError as e:
             # cap/quota rejection: restore the pre-retry status (don't strand it as
@@ -347,8 +358,14 @@ async def retry_request(
             await self._request_history.async_update_download_task_id(
                 musicbrainz_id, task_id
             )
+        else:
+            await self._request_history.async_update_status(
+                musicbrainz_id,
+                "imported",
+                completed_at=datetime.now(timezone.utc).isoformat(),
+            )
         return RetryRequestResponse(
-            success=True, message=f"Re-requested {record.album_title}"
+            success=True, message=f"Re-requested {self._record_title(record)}"
         )
 
     async def clear_history_item(
@@ -359,7 +376,9 @@ async def clear_history_item(
             return False
         # ownership checked before clearability so a non-owner gets 403, not a
         # misleading 200/False, on another user's row
-        if user_role != "admin" and record.user_id != user_id:
+        if user_role != "admin" and not await self._request_history.async_is_requester(
+            user_id, musicbrainz_id
+        ):
             raise PermissionDeniedError("Cannot clear another user's request")
         if record.status not in _CLEARABLE_STATUSES:
             return False
@@ -453,8 +472,56 @@ async def _fetch_library_mbids(self) -> set[str]:
                 return self._library_mbids_cache
             return set()
 
+    async def _dispatch_record(
+        self,
+        record: RequestHistoryRecord,
+        *,
+        origin: str,
+        fallback_user_id: str = "",
+        user_id_override: str | None = None,
+    ) -> str:
+        """Dispatch an approved/retried request without widening exact tracks."""
+        user_id = user_id_override or record.user_id or fallback_user_id
+        if record.request_kind == "track":
+            if not record.track_title:
+                raise ValidationError("Exact-track request is missing its track title")
+            return await self._acquisition.request_track(
+                user_id=user_id,
+                recording_mbid=record.musicbrainz_id,
+                artist_name=record.artist_name or "Unknown",
+                track_title=record.track_title,
+                album_title=record.album_title,
+                duration_seconds=record.duration_seconds,
+                release_group_mbid=record.track_release_group_mbid,
+                artist_mbid=record.artist_mbid,
+                release_mbid=record.release_mbid,
+                content_variant=getattr(record, "content_variant", "original"),
+            )
+        return await self._acquisition.request_album(
+            user_id=user_id,
+            release_group_mbid=record.musicbrainz_id,
+            artist_name=record.artist_name or "Unknown",
+            album_title=record.album_title or "Unknown",
+            year=record.year,
+            artist_mbid=record.artist_mbid,
+            origin=origin,
+            release_mbid=record.release_mbid,
+            track_count_priority=RequestPriority.USER_INITIATED,
+        )
+
+    @staticmethod
+    def _record_title(record: RequestHistoryRecord) -> str:
+        if record.request_kind == "track" and record.track_title:
+            return record.track_title
+        return record.album_title
+
     @staticmethod
     def _build_pending_item(record: RequestHistoryRecord) -> ActiveRequestItem:
+        cover_mbid = (
+            record.track_release_group_mbid
+            if record.request_kind == "track" and record.track_release_group_mbid
+            else record.musicbrainz_id
+        )
         return ActiveRequestItem(
             musicbrainz_id=record.musicbrainz_id,
             artist_name=record.artist_name,
@@ -462,7 +529,7 @@ def _build_pending_item(record: RequestHistoryRecord) -> ActiveRequestItem:
             artist_mbid=record.artist_mbid,
             year=record.year,
             cover_url=prefer_release_group_cover_url(
-                record.musicbrainz_id,
+                cover_mbid,
                 record.cover_url,
                 size=500,
             ),
@@ -478,6 +545,10 @@ def _build_pending_item(record: RequestHistoryRecord) -> ActiveRequestItem:
             library_queue_id=None,
             user_id=record.user_id,
             requested_by_name=record.requested_by_name,
+            request_kind=record.request_kind,
+            track_title=record.track_title,
+            duration_seconds=record.duration_seconds,
+            track_release_group_mbid=record.track_release_group_mbid,
         )
 
     async def _check_if_completed(
@@ -487,7 +558,10 @@ async def _check_if_completed(
     ) -> bool:
         now_iso = datetime.now(timezone.utc).isoformat()
 
-        if record.musicbrainz_id.lower() in library_mbids:
+        if (
+            record.request_kind != "track"
+            and record.musicbrainz_id.lower() in library_mbids
+        ):
             await self._request_history.async_update_status(
                 record.musicbrainz_id, "imported", completed_at=now_iso
             )
diff --git a/backend/target_application.py b/backend/target_application.py
index aab226798..d0635d602 100644
--- a/backend/target_application.py
+++ b/backend/target_application.py
@@ -747,6 +747,7 @@ def create_production_target_application() -> FastAPI:
             "/api/v1/discover": (10.0, 20),
             "/api/v1/covers": (15.0, 30),
             "/api/v1/auth/login": (2.0, 5),
+            "/api/v1/auth/device-sessions": (1.0, 5),
             "/api/v1/auth/setup": (1.0, 3),
             "/api/v1/auth/plex/poll": (5.0, 10),
             "/api/v1/auth/jellyfin/login": (2.0, 5),
diff --git a/backend/tests/compat/test_connect_apps_routes.py b/backend/tests/compat/test_connect_apps_routes.py
index 706fa2164..24cbb579c 100644
--- a/backend/tests/compat/test_connect_apps_routes.py
+++ b/backend/tests/compat/test_connect_apps_routes.py
@@ -50,6 +50,7 @@ async def test_get_settings_any_user(app_password_service, tmp_path):
     r = build_test_client(app).get("/connect-apps/settings")
     assert r.status_code == 200
     assert r.json()["subsonic_enabled"] is False
+    assert r.json()["exact_track_approval_supported"] is True
 
 
 async def test_get_settings_unauthenticated_401(app_password_service, tmp_path):
diff --git a/backend/tests/infrastructure/test_auth_store.py b/backend/tests/infrastructure/test_auth_store.py
index 26d2040b0..bbead99d3 100644
--- a/backend/tests/infrastructure/test_auth_store.py
+++ b/backend/tests/infrastructure/test_auth_store.py
@@ -6,6 +6,7 @@
 """
 
 import hashlib
+import sqlite3
 import threading
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
@@ -25,6 +26,38 @@ def test_migration_is_idempotent(tmp_path: Path):
     assert db_path.exists()
 
 
+def test_session_kind_migration_preserves_legacy_tokens_as_standard(tmp_path: Path):
+    db_path = tmp_path / "library.db"
+    with sqlite3.connect(db_path) as connection:
+        connection.execute(
+            """CREATE TABLE auth_tokens (
+                   id TEXT PRIMARY KEY,
+                   user_id TEXT NOT NULL,
+                   token_hash TEXT NOT NULL UNIQUE,
+                   issued_at TEXT NOT NULL,
+                   expires_at TEXT NOT NULL,
+                   last_seen_at TEXT NOT NULL,
+                   revoked INTEGER NOT NULL DEFAULT 0,
+                   user_agent TEXT
+               )"""
+        )
+        connection.execute(
+            """INSERT INTO auth_tokens
+                   (id, user_id, token_hash, issued_at, expires_at, last_seen_at,
+                    revoked, user_agent)
+               VALUES ('legacy-token', 'user-1', 'hash', 'now', 'later', 'now', 0,
+                       'Tonarr companion · Watch')"""
+        )
+
+    AuthStore(db_path)
+
+    with sqlite3.connect(db_path) as connection:
+        session_kind = connection.execute(
+            "SELECT session_kind FROM auth_tokens WHERE id = 'legacy-token'"
+        ).fetchone()[0]
+    assert session_kind == "standard"
+
+
 @pytest.mark.asyncio
 async def test_spotify_state_roundtrip_is_single_use(tmp_path: Path):
     store = AuthStore(tmp_path / "auth.db")
diff --git a/backend/tests/infrastructure/test_request_history_tracks.py b/backend/tests/infrastructure/test_request_history_tracks.py
new file mode 100644
index 000000000..dae58c19b
--- /dev/null
+++ b/backend/tests/infrastructure/test_request_history_tracks.py
@@ -0,0 +1,96 @@
+import sqlite3
+
+import pytest
+
+from infrastructure.persistence.request_history import RequestHistoryStore
+
+
+@pytest.mark.asyncio
+async def test_exact_track_metadata_survives_request_history_round_trip(tmp_path):
+    store = RequestHistoryStore(tmp_path / "droppedneedle.db")
+
+    await store.async_record_request(
+        musicbrainz_id="recording-1",
+        artist_name="Radiohead",
+        album_title="OK Computer",
+        artist_mbid="artist-1",
+        user_id="listener-1",
+        requested_by_name="Listener",
+        release_mbid="release-1",
+        initial_status="awaiting_approval",
+        request_kind="track",
+        track_title="Airbag",
+        duration_seconds=287,
+        track_release_group_mbid="release-group-1",
+        content_variant="clean",
+    )
+
+    record = await store.async_get_record("RECORDING-1")
+
+    assert record is not None
+    assert record.content_variant == "clean"
+    assert record.request_kind == "track"
+    assert record.track_title == "Airbag"
+    assert record.duration_seconds == 287
+    assert record.track_release_group_mbid == "release-group-1"
+
+
+@pytest.mark.asyncio
+async def test_legacy_album_request_defaults_to_album_kind(tmp_path):
+    store = RequestHistoryStore(tmp_path / "droppedneedle.db")
+
+    await store.async_record_request("release-group-1", "Radiohead", "OK Computer")
+
+    record = await store.async_get_record("release-group-1")
+
+    assert record is not None
+    assert record.content_variant == "original"
+    assert record.request_kind == "album"
+    assert record.track_title is None
+
+
+@pytest.mark.asyncio
+async def test_shared_request_remains_visible_and_private_for_each_listener(tmp_path):
+    store = RequestHistoryStore(tmp_path / "droppedneedle.db")
+    await store.async_record_request(
+        "release-group-1",
+        "Artist",
+        "Album",
+        user_id="listener-1",
+        requested_by_name="First listener",
+    )
+    await store.async_add_requester("release-group-1", "listener-2", "Second listener")
+
+    first = await store.async_get_active_requests_for_user("listener-1")
+    second = await store.async_get_active_requests_for_user("listener-2")
+    assert [record.user_id for record in first] == ["listener-1"]
+    assert [record.user_id for record in second] == ["listener-2"]
+    assert [record.requested_by_name for record in second] == ["Second listener"]
+    second_history, total = await store.async_get_history_for_user("listener-2")
+    assert total == 1
+    assert [record.user_id for record in second_history] == ["listener-2"]
+
+    assert await store.async_requester_count("release-group-1") == 2
+    assert await store.async_remove_requester("listener-1", "release-group-1")
+    canonical = await store.async_get_record("release-group-1")
+    assert canonical is not None
+    assert canonical.user_id == "listener-2"
+
+
+@pytest.mark.asyncio
+async def test_existing_user_attribution_is_backfilled_during_upgrade(tmp_path):
+    path = tmp_path / "droppedneedle.db"
+    store = RequestHistoryStore(path)
+    await store.async_record_request(
+        "release-group-1",
+        "Artist",
+        "Album",
+        user_id="legacy-listener",
+        requested_by_name="Legacy listener",
+    )
+    with sqlite3.connect(path) as connection:
+        connection.execute("DROP TABLE request_history_requesters")
+
+    upgraded = RequestHistoryStore(path)
+    active = await upgraded.async_get_active_requests_for_user("legacy-listener")
+    assert [record.musicbrainz_id for record in active] == ["release-group-1"]
diff --git a/backend/tests/repositories/test_coverart_repository_memory_cache.py b/backend/tests/repositories/test_coverart_repository_memory_cache.py
index bfb713542..0bc1460a0 100644
--- a/backend/tests/repositories/test_coverart_repository_memory_cache.py
+++ b/backend/tests/repositories/test_coverart_repository_memory_cache.py
@@ -151,4 +151,4 @@ async def test_artist_fetcher_uses_non_default_user_agent_for_external_requests(
         repo = CoverArtRepository(http_client=http_client, cache=cache, cache_dir=tmp_path)
 
         assert repo._artist_fetcher._external_headers is not None
-        assert repo._artist_fetcher._external_headers['User-Agent'].startswith('DroppedNeedle/')
+        assert repo._artist_fetcher._external_headers['User-Agent'].startswith('DroppedNeedleApp/')
diff --git a/backend/tests/repositories/test_musicbrainz_album_release_group.py b/backend/tests/repositories/test_musicbrainz_album_release_group.py
index 191f29b8d..2550bb469 100644
--- a/backend/tests/repositories/test_musicbrainz_album_release_group.py
+++ b/backend/tests/repositories/test_musicbrainz_album_release_group.py
@@ -5,6 +5,7 @@
 
 import pytest
 
+from core.exceptions import ExternalServiceError
 from models.album import AlbumInfo
 from repositories.musicbrainz_album import MusicBrainzAlbumMixin
 
@@ -72,8 +73,9 @@ async def test_fetch_rg_negative_caches_404_but_not_transient(monkeypatch):
     repo._cache.set.assert_awaited_once_with("ck-404", {}, ttl_seconds=600)
 
     repo._cache.set.reset_mock()
-    monkeypatch.setattr(mod, "mb_api_get", AsyncMock(side_effect=RuntimeError("503")))
-    assert await repo._fetch_release_group_by_id("rg-503", ["artist-credits"], "ck-503") is None
+    monkeypatch.setattr(mod, "mb_api_get", AsyncMock(side_effect=ExternalServiceError("503")))
+    with pytest.raises(ExternalServiceError, match="temporarily unavailable"):
+        await repo._fetch_release_group_by_id("rg-503", ["artist-credits"], "ck-503")
     repo._cache.set.assert_not_called()
 
 
diff --git a/backend/tests/repositories/test_musicbrainz_circuit_open_degradation.py b/backend/tests/repositories/test_musicbrainz_circuit_open_degradation.py
index a45a253d9..346b437a4 100644
--- a/backend/tests/repositories/test_musicbrainz_circuit_open_degradation.py
+++ b/backend/tests/repositories/test_musicbrainz_circuit_open_degradation.py
@@ -8,6 +8,7 @@
 
 import pytest
 
+from core.exceptions import ExternalServiceError
 import repositories.musicbrainz_album as album_module
 from infrastructure.queue.priority_queue import RequestPriority
 from infrastructure.resilience.retry import CircuitOpenError
@@ -61,7 +62,8 @@ async def test_get_release_group_by_id_degrades_quietly_when_breaker_open(
     open_breaker, caplog
 ) -> None:
     with caplog.at_level(logging.ERROR, logger="repositories.musicbrainz_album"):
-        assert await _Repo().get_release_group_by_id("rg-1") is None
+        with pytest.raises(ExternalServiceError, match="temporarily unavailable"):
+            await _Repo().get_release_group_by_id("rg-1")
     assert caplog.records == []
     open_breaker.assert_called_once()
 
@@ -71,10 +73,8 @@ async def test_get_release_by_id_degrades_quietly_when_breaker_open(
     open_breaker, caplog
 ) -> None:
     with caplog.at_level(logging.ERROR, logger="repositories.musicbrainz_album"):
-        assert (
+        with pytest.raises(ExternalServiceError, match="temporarily unavailable"):
             await _Repo().get_release_by_id("release-1", priority=RequestPriority.USER_INITIATED)
-            is None
-        )
     assert caplog.records == []
     open_breaker.assert_called_once()
 
diff --git a/backend/tests/repositories/test_musicbrainz_release_outage.py b/backend/tests/repositories/test_musicbrainz_release_outage.py
new file mode 100644
index 000000000..b4dbd6390
--- /dev/null
+++ b/backend/tests/repositories/test_musicbrainz_release_outage.py
@@ -0,0 +1,48 @@
+"""Transient catalog failures must not masquerade as missing exact editions."""
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+
+from core.exceptions import ExternalServiceError
+from repositories.musicbrainz_album import MusicBrainzAlbumMixin
+
+
+class Repo(MusicBrainzAlbumMixin):
+    def __init__(self):
+        self._cache = AsyncMock()
+        self._cache.get.return_value = None
+
+
+@pytest.mark.asyncio
+async def test_release_network_outage_is_retryable_and_not_cached(monkeypatch):
+    fetch = AsyncMock(side_effect=httpx.ConnectError("TLS connection closed"))
+    monkeypatch.setattr("repositories.musicbrainz_album.mb_api_get", fetch)
+    repo = Repo()
+    with pytest.raises(ExternalServiceError, match="temporarily unavailable"):
+        await repo.get_release_by_id("selected-release")
+    repo._cache.set.assert_not_awaited()
+    fetch.side_effect = None
+    fetch.return_value = {"id": "selected-release", "media": [{"tracks": []}]}
+    assert (await repo.get_release_by_id("selected-release"))["id"] == "selected-release"
+    assert fetch.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_release_missing_remains_distinct_from_outage(monkeypatch):
+    monkeypatch.setattr("repositories.musicbrainz_album.mb_api_get", AsyncMock(return_value={}))
+    assert await Repo().get_release_by_id("missing-release") is None
+
+
+@pytest.mark.asyncio
+async def test_release_group_stream_reset_recovers_without_negative_cache(monkeypatch):
+    fetch = AsyncMock(side_effect=httpx.RemoteProtocolError("HTTP/2 stream reset"))
+    monkeypatch.setattr("repositories.musicbrainz_album.mb_api_get", fetch)
+    repo = Repo()
+    with pytest.raises(ExternalServiceError, match="temporarily unavailable"):
+        await repo.get_release_group_by_id("selected-album")
+    repo._cache.set.assert_not_awaited()
+    fetch.side_effect = None
+    fetch.return_value = {"id": "selected-album", "releases": [{"id": "edition"}]}
+    assert (await repo.get_release_group_by_id("selected-album"))["id"] == "selected-album"
+    assert fetch.await_count == 2
diff --git a/backend/tests/routes/test_auth_username.py b/backend/tests/routes/test_auth_username.py
index 7a16744b6..8c8b33cfd 100644
--- a/backend/tests/routes/test_auth_username.py
+++ b/backend/tests/routes/test_auth_username.py
@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import asyncio
+import sqlite3
 
 import pytest
 from fastapi import FastAPI
@@ -136,6 +137,213 @@ def test_me_returns_username_fields(tmp_path):
     assert body["username_display"] == "Jane"
 
 
+def test_owned_device_session_delete_revokes_bearer(tmp_path):
+    app, service = _app(tmp_path)
+    user, account_token = asyncio.run(
+        service.create_first_admin(
+            display_name="Jane",
+            username="jane",
+            password=PASSWORD,
+        )
+    )
+    app.dependency_overrides[_get_current_user] = lambda: user
+    client = build_test_client(app)
+
+    response = client.post(
+        "/auth/device-sessions",
+        json={"device_name": "Kyle Apple Watch Ultra"},
+    )
+
+    assert response.status_code == 200
+    assert response.json()["token"]
+    sessions = asyncio.run(service.list_sessions(user.id))
+    assert len(sessions) == 2
+    companion_session = next(
+        session
+        for session in sessions
+        if session.user_agent == "Tonarr companion · Kyle Apple Watch Ultra"
+    )
+
+    revoked = client.delete(f"/auth/sessions/{companion_session.id}")
+
+    assert revoked.status_code == 204
+    assert asyncio.run(service.verify_token(response.json()["token"])) is None
+    assert asyncio.run(service.verify_token(account_token)) is not None
+
+
+def test_cross_user_cannot_revoke_device_session(tmp_path):
+    app, service = _app(tmp_path)
+    owner, _ = asyncio.run(
+        service.create_first_admin(
+            display_name="Jane",
+            username="jane",
+            password=PASSWORD,
+        )
+    )
+    other_user = asyncio.run(
+        service.admin_create_user(
+            display_name="Alex",
+            username="alex",
+            password=PASSWORD,
+        )
+    )
+    app.dependency_overrides[_get_current_user] = lambda: owner
+    client = build_test_client(app)
+    created = client.post(
+        "/auth/device-sessions",
+        json={"device_name": "Kyle Apple Watch Ultra"},
+    )
+    companion_session = next(
+        session
+        for session in asyncio.run(service.list_sessions(owner.id))
+        if session.user_agent == "Tonarr companion · Kyle Apple Watch Ultra"
+    )
+
+    app.dependency_overrides[_get_current_user] = lambda: other_user
+    denied = client.delete(f"/auth/sessions/{companion_session.id}")
+
+    assert denied.status_code == 403
+    assert asyncio.run(service.verify_token(created.json()["token"])) is not None
+
+
+def test_device_label_collision_does_not_revoke_ordinary_session(tmp_path):
+    app, service = _app(tmp_path)
+    client = build_test_client(app)
+    setup = client.post(
+        "/auth/setup",
+        headers={"User-Agent": b"Tonarr companion \xb7 Kyle Apple Watch Ultra"},
+        json={"display_name": "Jane", "username": "jane", "password": PASSWORD},
+    )
+    account_token = setup.json()["token"]
+    verified = asyncio.run(service.verify_token(account_token))
+    assert verified is not None
+    user, _ = verified
+    app.dependency_overrides[_get_current_user] = lambda: user
+
+    companion = client.post(
+        "/auth/device-sessions",
+        json={"device_name": "Kyle Apple Watch Ultra"},
+    )
+
+    assert companion.status_code == 200
+    assert asyncio.run(service.verify_token(account_token)) is not None
+    assert asyncio.run(service.verify_token(companion.json()["token"])) is not None
+    assert len(asyncio.run(service.list_sessions(user.id))) == 2
+
+
+def test_same_label_replacement_invalidates_old_bearer(tmp_path):
+    app, service = _app(tmp_path)
+    user, _ = asyncio.run(
+        service.create_first_admin(
+            display_name="Jane",
+            username="jane",
+            password=PASSWORD,
+        )
+    )
+    app.dependency_overrides[_get_current_user] = lambda: user
+    client = build_test_client(app)
+    original = client.post(
+        "/auth/device-sessions",
+        json={"device_name": "Kyle Apple Watch Ultra"},
+    )
+
+    replacement = client.post(
+        "/auth/device-sessions",
+        json={"device_name": "Kyle Apple Watch Ultra"},
+    )
+
+    assert replacement.status_code == 200
+    assert replacement.json()["token"] != original.json()["token"]
+    assert asyncio.run(service.verify_token(original.json()["token"])) is None
+    assert asyncio.run(service.verify_token(replacement.json()["token"])) is not None
+    assert len(asyncio.run(service.list_sessions(user.id))) == 2
+
+
+def test_device_session_preflight_failure_preserves_old_bearer(tmp_path, monkeypatch):
+    app, service = _app(tmp_path)
+    user, _ = asyncio.run(
+        service.create_first_admin(
+            display_name="Jane",
+            username="jane",
+            password=PASSWORD,
+        )
+    )
+    app.dependency_overrides[_get_current_user] = lambda: user
+    client = build_test_client(app)
+    original = client.post(
+        "/auth/device-sessions",
+        json={"device_name": "Kyle Apple Watch Ultra"},
+    )
+
+    async def _provider_failure(_user_ids):
+        raise RuntimeError("forced provider lookup failure")
+
+    monkeypatch.setattr(service, "get_provider_names_for_users", _provider_failure)
+    invalid = client.post("/auth/device-sessions", json={"device_name": "  "})
+    failed = client.post(
+        "/auth/device-sessions",
+        json={"device_name": "Kyle Apple Watch Ultra"},
+    )
+
+    assert invalid.status_code == 400
+    assert failed.status_code == 500
+    assert asyncio.run(service.verify_token(original.json()["token"])) is not None
+    assert len(asyncio.run(service.list_sessions(user.id))) == 2
+
+
+def test_failed_same_label_replacement_preserves_old_bearer(tmp_path):
+    app, service = _app(tmp_path)
+    user, _ = asyncio.run(
+        service.create_first_admin(
+            display_name="Jane",
+            username="jane",
+            password=PASSWORD,
+        )
+    )
+    app.dependency_overrides[_get_current_user] = lambda: user
+    client = build_test_client(app)
+    original = client.post(
+        "/auth/device-sessions",
+        json={"device_name": "Kyle Apple Watch Ultra"},
+    )
+    assert original.status_code == 200
+    with sqlite3.connect(tmp_path / "library.db") as connection:
+        connection.execute(
+            """CREATE TRIGGER fail_device_session_replacement
+               BEFORE UPDATE OF revoked ON auth_tokens
+               WHEN OLD.user_agent = 'Tonarr companion · Kyle Apple Watch Ultra'
+                 AND NEW.revoked = 1
+               BEGIN
+                 SELECT RAISE(ABORT, 'forced replacement failure');
+               END"""
+        )
+
+    failed = client.post(
+        "/auth/device-sessions",
+        json={"device_name": "Kyle Apple Watch Ultra"},
+    )
+
+    assert failed.status_code == 500
+    assert asyncio.run(service.verify_token(original.json()["token"])) is not None
+    sessions = asyncio.run(service.list_sessions(user.id))
+    assert len(sessions) == 2
+    assert sum(
+        session.user_agent == "Tonarr companion · Kyle Apple Watch Ultra"
+        for session in sessions
+    ) == 1
+
+
+def test_device_session_rejects_empty_or_unbounded_label(tmp_path):
+    app, _ = _app(tmp_path)
+    app.dependency_overrides[_get_current_user] = lambda: UserRecord(
+        id="u-watch", display_name="Jane", role="user", created_at="t"
+    )
+    client = build_test_client(app)
+
+    assert client.post("/auth/device-sessions", json={"device_name": "  "}).status_code == 400
+    assert client.post("/auth/device-sessions", json={"device_name": "x" * 81}).status_code == 400
+
+
 def test_admin_create_user_with_username_and_duplicate_conflict(tmp_path):
     app, _ = _app(tmp_path)
     app.dependency_overrides[_get_current_admin] = mock_admin_user
diff --git a/backend/tests/routes/test_request_routes.py b/backend/tests/routes/test_request_routes.py
index 3f0c8d8a8..e08beb6e5 100644
--- a/backend/tests/routes/test_request_routes.py
+++ b/backend/tests/routes/test_request_routes.py
@@ -11,7 +11,7 @@
 from fastapi import FastAPI
 
 from api.v1.routes import requests, tracks
-from core.dependencies import get_acquisition_dispatcher, get_request_service
+from core.dependencies import get_request_service
 from middleware import _get_current_user
 from services.native.download_service import ALREADY_IN_LIBRARY
 from services.request_service import RequestService
@@ -40,11 +40,11 @@ def _requests_app(service: RequestService, role: str) -> FastAPI:
     return app
 
 
-def _tracks_app(download_service: AsyncMock) -> FastAPI:
+def _tracks_app(service: RequestService, role: str) -> FastAPI:
     app = FastAPI()
     app.include_router(tracks.router)
-    app.dependency_overrides[get_acquisition_dispatcher] = lambda: download_service
-    app.dependency_overrides[_get_current_user] = lambda: mock_user(role="user", user_id="u1")
+    app.dependency_overrides[get_request_service] = lambda: service
+    app.dependency_overrides[_get_current_user] = lambda: mock_user(role=role, user_id="u1")
     return app
 
 
@@ -89,34 +89,54 @@ def test_request_new_unauthenticated_401():
     assert response.status_code == 401
 
 
-def test_track_request_returns_task_id():
+def test_track_request_user_role_awaits_approval_without_dispatch():
     ds = AsyncMock()
-    ds.request_track.return_value = "task-track-1"
-    response = build_test_client(_tracks_app(ds)).post(
+    service, history = _request_service(ds)
+    response = build_test_client(_tracks_app(service, "user")).post(
         "/tracks/rec-1/request", json={"artist_name": "Radiohead", "track_title": "Airbag"}
     )
     assert response.status_code == 200
     body = response.json()
-    assert body["status"] == "queued"
-    assert body["task_id"] == "task-track-1"
+    assert body["status"] == "awaiting_approval"
+    assert body["task_id"] is None
+    ds.request_track.assert_not_awaited()
+    assert history.async_record_request.await_args.kwargs["request_kind"] == "track"
+
+
+def test_track_request_trusted_returns_task_id():
+    ds = AsyncMock()
+    ds.request_track.return_value = "task-track-1"
+    service, history = _request_service(ds)
+    response = build_test_client(_tracks_app(service, "trusted")).post(
+        "/tracks/rec-1/request", json={"artist_name": "Radiohead", "track_title": "Airbag"}
+    )
+    assert response.status_code == 200
+    assert response.json()["status"] == "queued"
+    assert response.json()["task_id"] == "task-track-1"
     ds.request_track.assert_awaited_once()
+    history.async_update_download_task_id.assert_awaited_once_with(
+        "rec-1", "task-track-1"
+    )
 
 
 def test_track_request_already_in_library():
     ds = AsyncMock()
     ds.request_track.return_value = ALREADY_IN_LIBRARY
-    response = build_test_client(_tracks_app(ds)).post(
+    service, history = _request_service(ds)
+    response = build_test_client(_tracks_app(service, "admin")).post(
         "/tracks/rec-1/request", json={"artist_name": "Radiohead", "track_title": "Airbag"}
     )
     assert response.status_code == 200
     assert response.json()["status"] == "already_in_library"
+    history.async_update_status.assert_awaited_once()
 
 
 def test_track_request_unauthenticated_401():
     ds = AsyncMock()
     app = FastAPI()
     app.include_router(tracks.router)
-    app.dependency_overrides[get_acquisition_dispatcher] = lambda: ds
+    service, _history = _request_service(ds)
+    app.dependency_overrides[get_request_service] = lambda: service
     response = build_test_client(app).post(
         "/tracks/rec-1/request", json={"artist_name": "Radiohead", "track_title": "Airbag"}
     )
diff --git a/backend/tests/routes/test_tracks_routes.py b/backend/tests/routes/test_tracks_routes.py
index 89ad40593..6f74b1d40 100644
--- a/backend/tests/routes/test_tracks_routes.py
+++ b/backend/tests/routes/test_tracks_routes.py
@@ -5,24 +5,26 @@
 from fastapi import FastAPI
 
 from api.v1.routes import tracks
-from core.dependencies import get_acquisition_dispatcher, get_quota_service
+from api.v1.schemas.download import TrackRequestResponse
+from core.dependencies import get_request_service
+from core.exceptions import ValidationError
 from middleware import _get_current_user
-from services.native.download_service import ALREADY_IN_LIBRARY
 from tests.helpers import build_test_client, mock_user
 
 
-def _app(service, quota=None) -> FastAPI:
+def _app(service, role="user") -> FastAPI:
     app = FastAPI()
     app.include_router(tracks.router)
-    app.dependency_overrides[get_acquisition_dispatcher] = lambda: service
-    app.dependency_overrides[get_quota_service] = lambda: quota or AsyncMock()
-    app.dependency_overrides[_get_current_user] = lambda: mock_user(role="user", user_id="u1")
+    app.dependency_overrides[get_request_service] = lambda: service
+    app.dependency_overrides[_get_current_user] = lambda: mock_user(role=role, user_id="u1")
     return app
 
 
 def test_request_track_queued():
     service = AsyncMock()
-    service.request_track.return_value = "task-1"
+    service.request_track.return_value = TrackRequestResponse(
+        status="queued", task_id="task-1"
+    )
     response = build_test_client(_app(service)).post(
         "/tracks/rec-1/request",
         json={"artist_name": "Radiohead", "track_title": "Airbag"},
@@ -33,13 +35,16 @@ def test_request_track_queued():
     assert body["task_id"] == "task-1"
     service.request_track.assert_awaited_once()
     kwargs = service.request_track.await_args.kwargs
-    assert kwargs["recording_mbid"] == "rec-1"
+    assert service.request_track.await_args.args[0] == "rec-1"
     assert kwargs["user_id"] == "u1"
+    assert kwargs["user_role"] == "user"
 
 
 def test_request_track_already_in_library():
     service = AsyncMock()
-    service.request_track.return_value = ALREADY_IN_LIBRARY
+    service.request_track.return_value = TrackRequestResponse(
+        status="already_in_library"
+    )
     response = build_test_client(_app(service)).post(
         "/tracks/rec-1/request",
         json={"artist_name": "Radiohead", "track_title": "Airbag"},
@@ -54,8 +59,7 @@ def test_request_track_unauthenticated_401():
     service = AsyncMock()
     app = FastAPI()
     app.include_router(tracks.router)
-    app.dependency_overrides[get_acquisition_dispatcher] = lambda: service
-    app.dependency_overrides[get_quota_service] = lambda: AsyncMock()
+    app.dependency_overrides[get_request_service] = lambda: service
     response = build_test_client(app).post(
         "/tracks/rec-1/request",
         json={"artist_name": "Radiohead", "track_title": "Airbag"},
@@ -64,20 +68,17 @@ def test_request_track_unauthenticated_401():
 
 
 def test_request_track_over_quota_rejected_at_submit():
-    """Track asks bypass approval but still count toward the request quota (D20):
-    an over-quota user is rejected before the download service is touched."""
-    from core.exceptions import ValidationError
-
+    """Request-service quota failures remain a clear 400 at the route boundary."""
     service = AsyncMock()
-    quota = AsyncMock()
-    quota.check_request_quota.side_effect = ValidationError("Request limit reached (5 per 7 days)")
+    service.request_track.side_effect = ValidationError(
+        "Request limit reached (5 per 7 days)"
+    )
 
-    response = build_test_client(_app(service, quota)).post(
+    response = build_test_client(_app(service)).post(
         "/tracks/rec-1/request",
         json={"artist_name": "Radiohead", "track_title": "Airbag"},
     )
 
     assert response.status_code == 400
     assert "Request limit reached" in response.json()["error"]["message"]
-    service.request_track.assert_not_awaited()
-    quota.check_request_quota.assert_awaited_once_with("u1", "user")
+    service.request_track.assert_awaited_once()
diff --git a/backend/tests/services/native/test_target_consumer_services.py b/backend/tests/services/native/test_target_consumer_services.py
index 6d2d2508b..12fcd4463 100644
--- a/backend/tests/services/native/test_target_consumer_services.py
+++ b/backend/tests/services/native/test_target_consumer_services.py
@@ -545,6 +545,34 @@ async def test_target_repository_resolves_active_provider_and_local_album_ids(
     assert await repository.resolve_library_album_identifier("missing") is None
 
 
+@pytest.mark.asyncio
+async def test_target_repository_pages_live_enrichment_candidates(
+    target_services,
+) -> None:
+    store, _view, _favorites, _history, _root = target_services
+    repository = TargetLibraryRepository(store)
+
+    first = await repository.get_enrichment_candidates(after_mbid=None, limit=1)
+    first_cursor = f"{first[0][0]}:{first[0][1]}"
+    second = await repository.get_enrichment_candidates(
+        after_mbid=first_cursor,
+        limit=1,
+    )
+    complete = await repository.get_enrichment_candidates(after_mbid=None, limit=10)
+
+    assert first == [
+        (
+            "album",
+            RELEASE_GROUP_MBID,
+            {"title": "Identified", "artist_name": "Identified Artist"},
+        )
+    ]
+    assert second == [
+        ("artist", ARTIST_MBID, {"name": "Identified Artist"})
+    ]
+    assert complete == first + second
+
+
 @pytest.mark.asyncio
 async def test_album_service_selects_by_active_target_album_file_count(
     target_services,
diff --git a/backend/tests/services/test_acquisition_strategy_singles.py b/backend/tests/services/test_acquisition_strategy_singles.py
index 2c842e1b6..5b65782e8 100644
--- a/backend/tests/services/test_acquisition_strategy_singles.py
+++ b/backend/tests/services/test_acquisition_strategy_singles.py
@@ -16,7 +16,7 @@
 from models.download import DownloadTask, ScoredCandidate
 from models.download_manifest import ManifestCodec
 from repositories.protocols.download_client import DownloadSearchResult, TaskHandle
-from services.native.acquisition.strategy import SoulseekStrategy
+from services.native.acquisition.strategy import SoulseekStrategy, _clean_candidates
 
 _CANONICAL = 155.556  # "the arrival" (recording 180ceef5...), seconds
 
@@ -55,6 +55,26 @@ def _single_task(**overrides) -> DownloadTask:
     return DownloadTask(**kwargs)
 
 
+def test_clean_candidate_filter_rejects_explicit_markers_only_for_clean_tasks():
+    explicit = ScoredCandidate(
+        username="peer",
+        parent_directory="Artist - Album [Explicit]",
+        files=[_search_result(filename="Artist - Album [Explicit]/01.flac")],
+        tier="auto",
+    )
+    unmarked = ScoredCandidate(
+        username="peer",
+        parent_directory="Artist - Album",
+        files=[_search_result(filename="Artist - Album/01.flac")],
+        tier="auto",
+    )
+
+    assert _clean_candidates(
+        _single_task(content_variant="clean"), [explicit, unmarked]
+    ) == [unmarked]
+    assert _clean_candidates(_single_task(), [explicit, unmarked]) == [explicit, unmarked]
+
+
 def _strategy(tmp_path: Path):
     indexer = MagicMock()
     indexer.search_album = AsyncMock(
diff --git a/backend/tests/services/test_download_service.py b/backend/tests/services/test_download_service.py
index ebf051e0e..be75c6a29 100644
--- a/backend/tests/services/test_download_service.py
+++ b/backend/tests/services/test_download_service.py
@@ -847,6 +847,58 @@ async def test_request_track_persists_exact_release_track_mapping():
     assert (kwargs["disc_number"], kwargs["track_number"]) == (1, 1)
 
 
+@pytest.mark.asyncio
+async def test_clean_request_is_exact_and_persists_fail_closed_intent():
+    album_service = _single_album_service(
+        tracks=[
+            SimpleNamespace(
+                position=1,
+                disc_number=1,
+                title="Song",
+                recording_id="recording-clean",
+                release_track_id="release-track-clean",
+                length=180_000,
+            )
+        ]
+    )
+    service, store, *_ = _make_service(album_service=album_service)
+    service._library.has_track.return_value = False
+    store.get_active_task_for_track.return_value = None
+
+    await service.request_track(
+        "u1",
+        "recording-clean",
+        "Artist",
+        "Song",
+        release_group_mbid="group-clean",
+        release_mbid="release-clean",
+        content_variant="clean",
+    )
+
+    kwargs = store.create_task.await_args.kwargs
+    assert kwargs["origin"] == "clean_replacement"
+    assert kwargs["content_variant"] == "clean"
+    assert kwargs["release_mbid"] == "release-clean"
+    assert kwargs["release_track_mbid"] == "release-track-clean"
+
+
+@pytest.mark.asyncio
+async def test_clean_request_without_exact_release_is_rejected():
+    service, store, *_ = _make_service(album_service=_single_album_service())
+
+    with pytest.raises(ValidationError, match="exact MusicBrainz"):
+        await service.request_track(
+            "u1",
+            "recording-clean",
+            "Artist",
+            "Song",
+            release_group_mbid="group-clean",
+            content_variant="clean",
+        )
+
+    store.create_task.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_request_track_dedup_is_recording_keyed_not_album_keyed():
     # A second, different track of the same album must NOT be swallowed by the
@@ -1836,3 +1888,22 @@ async def test_upgrade_origin_never_fetches_an_unheld_recording():
 
     assert result == ALREADY_IN_LIBRARY
     store.create_task.assert_not_called()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("explicit_edition", [False, True])
+async def test_metadata_outage_is_retryable_and_never_starts_acquisition(explicit_edition):
+    from core.exceptions import ExternalServiceError
+
+    albums = _single_album_service()
+    outage = ExternalServiceError("MusicBrainz metadata is temporarily unavailable")
+    albums.get_album_tracks_info.side_effect = outage
+    albums.get_exact_edition_tracks_info.side_effect = outage
+    service, store, *_ = _make_service(album_service=albums)
+    store.get_active_task_for_album.return_value = None
+    with pytest.raises(ExternalServiceError, match="temporarily unavailable"):
+        await service.request_album(
+            "u1", "rg", "A", "B",
+            release_mbid="selected-edition" if explicit_edition else None,
+        )
+    store.create_task.assert_not_awaited()
diff --git a/backend/tests/services/test_file_processor.py b/backend/tests/services/test_file_processor.py
index dcc076e61..ce976feb0 100644
--- a/backend/tests/services/test_file_processor.py
+++ b/backend/tests/services/test_file_processor.py
@@ -187,6 +187,7 @@ def _manifest(
     is_track=False,
     expected_tracks=(),
     origin="user",
+    content_variant="original",
 ) -> DownloadManifest:
     return DownloadManifest(
         task_id=task_id,
@@ -202,6 +203,7 @@ def _manifest(
         is_track=is_track,
         expected_tracks=list(expected_tracks),
         origin=origin,
+        content_variant=content_variant,
     )
 
 
@@ -234,6 +236,110 @@ async def test_edition_conversion_requires_recording_fingerprint_proof(
     assert result.failed[0].reason == "fingerprint_unavailable"
 
 
+@pytest.mark.asyncio
+async def test_clean_import_requires_recording_fingerprint_proof(tmp_path: Path) -> None:
+    fp, _manager, _client, _library, downloads = _make_processor(
+        tmp_path, verify=False, fingerprinter=None
+    )
+    _place(downloads, "A/track.flac")
+    manifest = _manifest(
+        ExpectedFile(filename="A/track.flac", size=1),
+        release_mbid="release-clean",
+        is_track=True,
+        expected_tracks=[
+            ExpectedTrack(
+                track_number=1,
+                title="Airbag",
+                recording_mbid="recording-clean",
+                release_track_mbid="release-track-clean",
+            )
+        ],
+        origin="clean_replacement",
+        content_variant="clean",
+    )
+
+    result = await fp.process_downloaded(manifest)
+
+    assert result.succeeded == []
+    assert result.failed[0].reason == "fingerprint_unavailable"
+
+
+@pytest.mark.asyncio
+async def test_clean_import_rejects_unproven_recording(tmp_path: Path) -> None:
+    fingerprinter = MagicMock()
+    fingerprinter.fingerprint = AsyncMock(
+        return_value=FingerprintResult(
+            status="pass",
+            score=0.99,
+            artist="Radiohead",
+            title="Airbag",
+            recording_id="recording-explicit",
+        )
+    )
+    fp, _manager, _client, _library, downloads = _make_processor(
+        tmp_path, verify=False, fingerprinter=fingerprinter
+    )
+    _place(downloads, "A/track.flac")
+    manifest = _manifest(
+        ExpectedFile(filename="A/track.flac", size=1),
+        release_mbid="release-clean",
+        is_track=True,
+        expected_tracks=[
+            ExpectedTrack(
+                track_number=1,
+                title="Airbag",
+                recording_mbid="recording-clean",
+                release_track_mbid="release-track-clean",
+            )
+        ],
+        origin="clean_replacement",
+        content_variant="clean",
+    )
+
+    result = await fp.process_downloaded(manifest)
+
+    assert result.succeeded == []
+    assert result.failed[0].reason == "fingerprint_unverified"
+
+
+@pytest.mark.asyncio
+async def test_clean_import_accepts_only_exact_recording_proof(tmp_path: Path) -> None:
+    fingerprinter = MagicMock()
+    fingerprinter.fingerprint = AsyncMock(
+        return_value=FingerprintResult(
+            status="pass",
+            score=0.99,
+            artist="Radiohead",
+            title="Airbag",
+            recording_id="recording-clean",
+        )
+    )
+    fp, manager, _client, _library, downloads = _make_processor(
+        tmp_path, verify=False, fingerprinter=fingerprinter
+    )
+    _place(downloads, "A/track.flac")
+    manifest = _manifest(
+        ExpectedFile(filename="A/track.flac", size=1),
+        release_mbid="release-clean",
+        is_track=True,
+        expected_tracks=[
+            ExpectedTrack(
+                track_number=1,
+                title="Airbag",
+                recording_mbid="recording-clean",
+                release_track_mbid="release-track-clean",
+            )
+        ],
+        origin="clean_replacement",
+        content_variant="clean",
+    )
+
+    result = await fp.process_downloaded(manifest)
+
+    assert len(result.succeeded) == 1
+    assert await manager.has_album("rg-1") is True
+
+
 def _make_processor(
     tmp_path: Path,
     *,
diff --git a/backend/tests/services/test_musicbrainz_rate_cap.py b/backend/tests/services/test_musicbrainz_rate_cap.py
index 5adae70fd..980c95390 100644
--- a/backend/tests/services/test_musicbrainz_rate_cap.py
+++ b/backend/tests/services/test_musicbrainz_rate_cap.py
@@ -136,8 +136,20 @@ def test_instance_id_in_user_agent(self, tmp_path):
             root_app_dir=tmp_path,
         )
         ua = settings.get_user_agent()
+        assert ua.startswith("DroppedNeedleApp/")
         assert "a1b2c3d4" in ua
-        assert "DroppedNeedle/1.0" in ua
+
+    def test_user_agent_uses_default_contact_when_empty(self, tmp_path):
+        from core.config import Settings
+
+        settings = Settings(
+            instance_id="a1b2c3d4-e5f6-7890-abcd-ef1234567890",
+            contact_email="",
+            root_app_dir=tmp_path,
+        )
+        ua = settings.get_user_agent()
+        assert "contact@droppedneedle.com" in ua
+        assert "; ;" not in ua
 
     def test_user_agent_unknown_when_no_instance_id(self, tmp_path):
         from core.config import Settings
@@ -145,3 +157,30 @@ def test_user_agent_unknown_when_no_instance_id(self, tmp_path):
         settings = Settings(instance_id="", root_app_dir=tmp_path)
         ua = settings.get_user_agent()
         assert "unknown" in ua
+
+
+@pytest.mark.parametrize("tag", [None, "", "   "])
+def test_user_agent_always_has_a_version(tmp_path, monkeypatch, tag):
+    from core.config import Settings
+    if tag is None:
+        monkeypatch.delenv("COMMIT_TAG", raising=False)
+    else:
+        monkeypatch.setenv("COMMIT_TAG", tag)
+    settings = Settings(root_app_dir=tmp_path)
+    assert settings.get_user_agent().split()[0] == "DroppedNeedleApp/dev"
+
+
+def test_maintained_integration_can_identify_itself_without_changing_rate_limits(tmp_path):
+    from core.config import Settings
+    agent = "TonarrRequests/1.0 (self-hosted Dropped Needle integration; support@example.test)"
+    settings = Settings(root_app_dir=tmp_path, http_user_agent=agent)
+    assert settings.get_user_agent() == agent
+    defaults = MusicBrainzConnectionSettings()
+    assert defaults.rate_limit == 1.0
+
+
+def test_user_agent_rejects_header_injection(tmp_path):
+    from core.config import Settings
+    from pydantic import ValidationError
+    with pytest.raises(ValidationError):
+        Settings(root_app_dir=tmp_path, http_user_agent="TonarrRequests/1.0\r\nX-Extra: value")
diff --git a/backend/tests/services/test_request_service.py b/backend/tests/services/test_request_service.py
index bb93f6e8f..7fb5b66e5 100644
--- a/backend/tests/services/test_request_service.py
+++ b/backend/tests/services/test_request_service.py
@@ -17,9 +17,15 @@ def _make_service() -> tuple[RequestService, MagicMock, MagicMock]:
     request_history.async_update_status = AsyncMock()
     request_history.async_update_download_task_id = AsyncMock()
     request_history.async_bulk_record_requests = AsyncMock()
+    request_history.async_add_requester = AsyncMock()
+    request_history.async_add_requesters = AsyncMock()
+    request_history.async_is_requester = AsyncMock(return_value=True)
+    request_history.async_requester_count = AsyncMock(return_value=1)
+    request_history.async_remove_requester = AsyncMock(return_value=True)
     request_history.async_get_active_mbids = AsyncMock(return_value=set())
     request_history.async_get_requested_mbids = AsyncMock(return_value=set())
     download_service.request_album = AsyncMock(return_value="task-1")
+    download_service.request_track = AsyncMock(return_value="track-task-1")
     download_service.cancel_task = AsyncMock()
 
     get_ds = lambda: download_service  # noqa: E731
@@ -96,6 +102,7 @@ async def test_request_album_canonicalizes_release_alias_before_history_and_disp
         origin="user",
         release_mbid="release-edition",
         release_track_mbid=None,
+        content_variant="original",
     )
 
 
@@ -139,6 +146,106 @@ async def test_request_album_user_role_awaits_approval_without_dispatch():
     request_history.async_update_download_task_id.assert_not_awaited()
 
 
+@pytest.mark.asyncio
+async def test_existing_request_is_attributed_to_each_listener_without_redispatch():
+    service, request_history, download_service = _make_service()
+    request_history.async_get_record.return_value = SimpleNamespace(
+        status="pending", monitor_artist=False
+    )
+
+    response = await service.request_album(
+        "rg-123",
+        user_id="listener-2",
+        user_role="user",
+        requested_by_name="Second listener",
+    )
+
+    assert response.status == "pending"
+    request_history.async_add_requester.assert_awaited_once_with(
+        "rg-123", "listener-2", "Second listener"
+    )
+    download_service.request_album.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_request_track_user_role_records_exact_metadata_and_awaits_approval():
+    service, request_history, download_service = _make_service()
+
+    response = await service.request_track(
+        "recording-1",
+        user_id="listener-1",
+        user_role="user",
+        requested_by_name="Listener",
+        artist_name="Radiohead",
+        track_title="Airbag",
+        album_title="OK Computer",
+        duration_seconds=287,
+        release_group_mbid="release-group-1",
+        artist_mbid="artist-1",
+        release_mbid="release-1",
+    )
+
+    assert response.status == "awaiting_approval"
+    download_service.request_track.assert_not_awaited()
+    request_history.async_record_request.assert_awaited_once_with(
+        musicbrainz_id="recording-1",
+        artist_name="Radiohead",
+        album_title="OK Computer",
+        artist_mbid="artist-1",
+        user_id="listener-1",
+        requested_by_name="Listener",
+        release_mbid="release-1",
+        initial_status="awaiting_approval",
+        request_kind="track",
+        track_title="Airbag",
+        duration_seconds=287,
+        track_release_group_mbid="release-group-1",
+        content_variant="original",
+    )
+
+
+@pytest.mark.asyncio
+async def test_request_track_unknown_role_fails_closed_to_owner_approval():
+    service, request_history, download_service = _make_service()
+
+    response = await service.request_track(
+        "recording-unknown-role",
+        user_id="future-role-1",
+        user_role="future-role",
+        artist_name="Radiohead",
+        track_title="Airbag",
+    )
+
+    assert response.status == "awaiting_approval"
+    download_service.request_track.assert_not_awaited()
+    assert (
+        request_history.async_record_request.await_args.kwargs["initial_status"]
+        == "awaiting_approval"
+    )
+
+
+@pytest.mark.asyncio
+async def test_request_track_trusted_dispatches_exact_recording_and_links_task():
+    service, request_history, download_service = _make_service()
+
+    response = await service.request_track(
+        "recording-1",
+        user_id="trusted-1",
+        user_role="trusted",
+        artist_name="Radiohead",
+        track_title="Airbag",
+        album_title="OK Computer",
+        release_group_mbid="release-group-1",
+    )
+
+    assert response.status == "queued"
+    assert response.task_id == "track-task-1"
+    download_service.request_track.assert_awaited_once()
+    request_history.async_update_download_task_id.assert_awaited_once_with(
+        "recording-1", "track-task-1"
+    )
+
+
 @pytest.mark.asyncio
 async def test_request_album_already_in_library_not_linked_as_task_id():
     service, request_history, download_service = _make_service()
@@ -296,6 +403,7 @@ async def test_request_batch_does_not_overwrite_an_approval_pending_request():
 
     assert response.requested == 0
     assert response.skipped == 1
+    assert response.status == "already_requested"
     request_history.async_bulk_record_requests.assert_not_awaited()
     download_service.request_album.assert_not_awaited()
 
@@ -310,6 +418,7 @@ async def test_request_batch_user_role_awaits_approval_without_dispatch():
     resp = await service.request_batch(items, user_role="user", user_id="u1")
 
     assert "approval" in resp.message.lower()
+    assert resp.status == "awaiting_approval"
     download_service.request_album.assert_not_awaited()
 
 
@@ -356,6 +465,9 @@ async def test_cancel_batch_user_only_cancels_owned_requests():
     request_history.async_get_record = AsyncMock(
         side_effect=lambda mbid: records.get(mbid)
     )
+    request_history.async_is_requester = AsyncMock(
+        side_effect=lambda _user_id, mbid: mbid == "rg-mine"
+    )
 
     response = await service.cancel_batch(["rg-mine", "rg-theirs"], user_id="alice")
 
@@ -437,6 +549,7 @@ async def test_request_batch_quota_counts_only_new_items():
 
     assert response.success is True
     assert quota.check_request_quota.await_args.args == ("u1", "user", 1)
+    request_history.async_add_requesters.assert_awaited_once_with(["RG-1"], "u1", None)
 
 
 @pytest.mark.asyncio
@@ -486,3 +599,28 @@ async def test_request_album_resolves_download_service_per_dispatch():
 
     ds_a.request_album.assert_awaited_once()  # first dispatch used the original engine
     ds_b.request_album.assert_awaited_once()  # second used the NEW one (fails if captured)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("role", ["user", "admin"])
+async def test_track_route_content_variant_survives_approval_boundary(role):
+    service, history, downloads = _make_service()
+    response = await service.request_track("clean-recording", artist_name="Artist", track_title="Clean song", user_id="listener", user_role=role, content_variant="clean")
+    assert history.async_record_request.await_args.kwargs["content_variant"] == "clean"
+    if role == "user":
+        assert response.status == "awaiting_approval"
+        downloads.request_track.assert_not_awaited()
+    else:
+        assert response.status == "queued"
+        assert downloads.request_track.await_args.kwargs["content_variant"] == "clean"
+
+
+@pytest.mark.asyncio
+async def test_clean_request_does_not_reuse_unverified_original_request():
+    service, history, downloads = _make_service()
+    history.async_get_record.return_value = SimpleNamespace(status="pending", content_variant="original")
+    from core.exceptions import ValidationError
+    with pytest.raises(ValidationError):
+        await service.request_track("recording", artist_name="Artist", track_title="Song", user_id="listener", user_role="admin", content_variant="clean")
+    downloads.request_track.assert_not_awaited()
+    history.async_add_requester.assert_not_awaited()
diff --git a/backend/tests/services/test_requests_page_approve.py b/backend/tests/services/test_requests_page_approve.py
index cc092807b..99301ebb8 100644
--- a/backend/tests/services/test_requests_page_approve.py
+++ b/backend/tests/services/test_requests_page_approve.py
@@ -7,6 +7,7 @@
 import pytest
 
 from infrastructure.queue.priority_queue import RequestPriority
+from infrastructure.persistence.request_history import RequestHistoryRecord
 from services.requests_page_service import RequestsPageService
 from tests.helpers import make_builtin_dispatcher
 
@@ -15,7 +16,9 @@ def _make(
     record_status="awaiting_approval",
     *,
     request_album_result="task-9",
+    request_track_result="track-task-9",
     download_task_id=None,
+    request_kind="album",
 ):
     request_history = MagicMock()
     request_history.async_get_record = AsyncMock(
@@ -28,14 +31,25 @@ def _make(
             user_id="u1",
             download_task_id=download_task_id,
             release_mbid="release-edition",
+            musicbrainz_id="mbid-1",
+            request_kind=request_kind,
+            track_title="Airbag" if request_kind == "track" else None,
+            duration_seconds=287 if request_kind == "track" else None,
+            track_release_group_mbid="release-group-1"
+            if request_kind == "track"
+            else None,
         )
     )
     request_history.async_record_review = AsyncMock()
     request_history.async_update_download_task_id = AsyncMock()
     request_history.async_update_status = AsyncMock()
+    request_history.async_is_requester = AsyncMock(return_value=True)
+    request_history.async_requester_count = AsyncMock(return_value=1)
+    request_history.async_remove_requester = AsyncMock(return_value=True)
 
     download_service = MagicMock()
     download_service.request_album = AsyncMock(return_value=request_album_result)
+    download_service.request_track = AsyncMock(return_value=request_track_result)
     download_service.cancel_task = AsyncMock()
 
     async def _mbids() -> set[str]:
@@ -95,6 +109,24 @@ async def test_approve_rejects_non_awaiting_record():
     download_service.request_album.assert_not_awaited()
 
 
+@pytest.mark.asyncio
+async def test_approve_exact_track_dispatches_track_without_widening_to_album():
+    service, history, download_service = _make(request_kind="track")
+
+    resp = await service.approve_request("mbid-1", "admin-id", "Admin")
+
+    assert resp.success is True
+    download_service.request_album.assert_not_awaited()
+    download_service.request_track.assert_awaited_once()
+    kwargs = download_service.request_track.await_args.kwargs
+    assert kwargs["recording_mbid"] == "mbid-1"
+    assert kwargs["track_title"] == "Airbag"
+    assert kwargs["release_group_mbid"] == "release-group-1"
+    history.async_update_download_task_id.assert_awaited_once_with(
+        "mbid-1", "track-task-9"
+    )
+
+
 @pytest.mark.asyncio
 async def test_cancel_request_cancels_linked_native_task():
     service, history, download_service = _make(
@@ -108,6 +140,22 @@ async def test_cancel_request_cancels_linked_native_task():
     history.async_update_status.assert_awaited()
 
 
+@pytest.mark.asyncio
+async def test_cancel_shared_request_removes_only_current_listener():
+    service, history, download_service = _make(
+        record_status="downloading", download_task_id="task-9"
+    )
+    history.async_requester_count.return_value = 2
+
+    resp = await service.cancel_request("mbid-1", user_id="u1", user_role="user")
+
+    assert resp.success is True
+    assert "another listener" in resp.message
+    history.async_remove_requester.assert_awaited_once_with("u1", "mbid-1")
+    download_service.cancel_task.assert_not_awaited()
+    history.async_update_status.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_retry_request_redispatches_native_and_links():
     service, history, download_service = _make(
@@ -131,6 +179,49 @@ async def test_retry_request_redispatches_native_and_links():
     history.async_update_download_task_id.assert_awaited_once_with("mbid-1", "task-9")
 
 
+@pytest.mark.asyncio
+async def test_retry_exact_track_preserves_exact_track_semantics():
+    service, history, download_service = _make(
+        record_status="failed",
+        request_kind="track",
+        download_task_id="old-track-task",
+    )
+
+    resp = await service.retry_request("mbid-1", user_id="u1", user_role="user")
+
+    assert resp.success is True
+    download_service.request_album.assert_not_awaited()
+    download_service.request_track.assert_awaited_once()
+    assert (
+        download_service.request_track.await_args.kwargs["recording_mbid"] == "mbid-1"
+    )
+    history.async_update_download_task_id.assert_awaited_once_with(
+        "mbid-1", "track-task-9"
+    )
+
+
+def test_pending_exact_track_response_exposes_kind_title_and_release_artwork():
+    item = RequestsPageService._build_pending_item(
+        RequestHistoryRecord(
+            musicbrainz_id="recording-1",
+            artist_name="Radiohead",
+            album_title="OK Computer",
+            requested_at="2026-08-24T12:00:00+00:00",
+            status="awaiting_approval",
+            request_kind="track",
+            track_title="Airbag",
+            duration_seconds=287,
+            track_release_group_mbid="7b0032d0-09b3-4f21-a207-9eb26b746c4f",
+        )
+    )
+
+    assert item.request_kind == "track"
+    assert item.track_title == "Airbag"
+    assert item.duration_seconds == 287
+    assert item.track_release_group_mbid == "7b0032d0-09b3-4f21-a207-9eb26b746c4f"
+    assert "7b0032d0-09b3-4f21-a207-9eb26b746c4f" in (item.cover_url or "")
+
+
 @pytest.mark.asyncio
 async def test_sync_reconciles_request_from_native_download_task():
     """The rewritten reconciler reads the native download task (not the dead Lidarr
@@ -195,3 +286,12 @@ async def test_retry_over_cap_restores_prior_status_with_reason():
     assert "storage budget" in resp.message
     # flipped to 'pending' for the attempt, then restored to the pre-retry status
     assert history.async_update_status.await_args_list[-1].args == ("mbid-1", "failed")
+
+
+@pytest.mark.asyncio
+async def test_approved_clean_track_keeps_clean_variant():
+    service, history, downloads = _make(request_kind="track")
+    record = history.async_get_record.return_value
+    record.content_variant = "clean"
+    await service._dispatch_record(record, origin="approval")
+    assert downloads.request_track.await_args.kwargs["content_variant"] == "clean"
diff --git a/docs/tonarr-metadata-recovery.md b/docs/tonarr-metadata-recovery.md
new file mode 100644
index 000000000..526ae2166
--- /dev/null
+++ b/docs/tonarr-metadata-recovery.md
@@ -0,0 +1,67 @@
+# Tonarr metadata connection recovery — 2026-09-07
+
+The deployed Dropped Needle service could authenticate users but failed exact-song
+requests before a download task existed. Earlier diagnosis described this as a
+TLS/network outage. Controlled tests isolated a narrower cause: outbound
+application identification.
+
+From the same container, network and destination, the configured Dropped Needle
+User-Agent caused HTTP/2 stream resets and HTTP/1.1 disconnects. A diagnostic
+identifier succeeded. Adding a missing version alone did not resolve it. The
+maintained integration's truthful identifier, including its product/version,
+Dropped Needle integration description, operator contact and Tonarr URL, returned
+HTTP 200 from both MusicBrainz and ListenBrainz.
+
+## Change
+
+- Add optional `HTTP_USER_AGENT` for a maintained integration to identify itself
+  accurately. Reject control characters/header injection and bound its length.
+- Preserve the standard upstream identity by default, with a nonempty development
+  version even when Docker supplies an empty build argument.
+- Keep transient release-group and exact-edition lookup errors retryable instead
+  of misreporting them as nonexistent editions. Do not create an acquisition task
+  until the exact recording and edition are verified.
+- Keep provider rate limits, TLS verification, VPN routing, owner sources, quotas
+  and approval rules intact. No other container was recreated.
+
+Production image: `local/droppedneedle:7bfd742`, source
+`7bfd742eaf9e7dd822451e95035c696d3f227292`, image digest
+`sha256:bcc8ed1b42438244c0fbdb2ba61740c6136fbc787b1c0eb0d9840339ace2c5fb`.
+The existing Compose file has a protected rollback beside it. Only Dropped
+Needle's image and application-identification environment setting changed.
+
+## Verification
+
+- 780 repository/download/edition/rate-limit tests passed.
+- The restarted service is healthy; its metadata-health endpoint reports no
+  MusicBrainz or ListenBrainz degradation after the successful request.
+- Retrying the original “Make It Right” / BTS request returned success in 8.4 s.
+- Durable task `c95b4a0b6b6c4a90aaaae958ef692f6e` retains recording
+  `0228077b-505d-4224-a882-d8d044bc8ed5`, edition
+  `64160dc9-9841-4301-a0a2-537ec74472de`, and release-track
+  `42b30c94-65ea-4211-9447-43e3f870e85f`.
+- The task completed: one file imported, zero failures, at
+  `2026-09-07T00:44:01.432821Z`. The owner's configured Soulseek fallback
+  supplied the recording; no acquisition-source policy changed.
+- The imported FLAC is 25,646,191 bytes, stereo 16-bit/44.1 kHz, duration
+  226.321020 seconds. `ffprobe` and a full `ffmpeg` decode-to-null both exited
+  zero without audio errors. SHA-256:
+  `226be34ad830b1609005c83c79227864efb91f9c871083c0bd0ce98e3fa44671`.
+- Tonarr's real iOS library found the recording automatically. Native playback
+  ran through its natural 3:46 ending; pause/resume and adding/playing the next
+  track also worked. Source: `951ff79`.
+- Signed Google Store Tonarr 1000011 played the same recording through a public
+  HTTPS Navidrome reviewer connection on Android TV API 36, paused/resumed via
+  media controls, then automatically advanced to the next track. This proves
+  emulator playback, not physical Fire TV certification or later builds.
+
+Sanitized receipts are in `/root/artifacts/tonarr-request-acceptance/`:
+`dn-network-recovery-request.json`, `dn-recovered-track-status.json`,
+`dn-completed-download-proof.json`, and `dn-android-tv-playback.json`.
+iOS observations are in
+`/root/artifacts/tonarr-production-apple/dn-native-playback.json`.
+
+Upstream had previously addressed blocked application identification in
+[commit a4aa4ad](https://github.com/DroppedNeedle/DroppedNeedle/commit/a4aa4ad48f03319bf5950becf936c9aefec49b76).
+This deployment still reproduced header-dependent failures with that identity;
+we did not establish the metadata provider's internal blocking rule.