feat(search): answer search from a local index; tier downloads by lifetime

Search's instant leg read only downloaded items, so with no downloads it
returned nothing and every keystroke fell through to a full Recursive=true
server query. It now reads the whole synced catalog through the same
availability CTE get_items uses, gated on the same include_catalog_browse
flag so search and browse cannot diverge. (UR-065, DR-108)

Also fixes three defects found while confirming that:

- items_fts grew by a full duplicate index every catalog pass. INSERT OR
  REPLACE fires no AFTER DELETE trigger without recursive_triggers, so the
  old index row was orphaned, and a TEXT PRIMARY KEY meant the replacement
  took a fresh rowid and inserted a second entry. Now a real upsert, with
  migration 021 rebuilding existing indexes. (DR-110)
- DELETE FROM items existed nowhere, so server-side deletions never
  propagated. Adds a post-crawl mark-and-sweep, scoped to crawled types,
  skipping downloaded items, and refusing to run after a partial crawl
  because items.parent_id cascades. (DR-110)
- The index omitted MusicArtist, Playlist and People, which search groups
  results by. Adds them plus people_fts (migration 022). (DR-111)

Re-indexing moves from a frontend startup call to a Rust background task
with a 6h TTL, so a long session no longer searches a stale catalog and a
restart no longer forces a crawl regardless of freshness. (DR-109, IR-030)

Downloads gain a lifetime tier. Eviction selected every completed row by
age with no download_source filter, so hitting the storage limit deleted
the oldest download -- typically one saved deliberately for offline -- to
make room for a precached track. It now reclaims only 'auto' rows, and
expired ones are reclaimed first, before live cache is evicted.
(DR-126, DR-127)

Downloaded video and audio-only handoffs now play from disk instead of
streaming; the video path had never consulted downloads at all. No
transcode is involved: MPV runs video=no and ExoPlayer has no surface for
an Audio item. (DR-123 in part, DR-128)

FTS queries are built as quoted phrases so apostrophes, hyphens and
slashes are data rather than operator syntax, and the item-type filter is
bound rather than interpolated.

Specs: docs/specs/catalog-index-search.md,
docs/specs/read-through-media-cache.md

Includes concurrently-developed favourites browsing and background-audio
stream-end handling; the two workstreams share offline.rs, lib.rs and
online.rs, so no subset of files builds independently.

src-tauri/src/commands/catalog.rs

@@ -14,10 +14,12 @@
 //!   that were queued offline (they have `stream_url IS NULL`), mirroring the
 //!   heal-and-pump pattern in `player_preload_upcoming`.
 
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
+use std::time::Duration;
 
 use log::{info, warn};
-use tauri::State;
+use tauri::{Emitter, Manager, State};
 
 use crate::commands::download::{pump_download_queue, DownloadManagerWrapper};
 use crate::commands::repository::RepositoryManagerWrapper;
@@ -29,16 +31,79 @@ use crate::storage::db_service::{DatabaseService, Query, QueryParam};
 /// full-catalog sync.
 const LAST_CATALOG_SYNC_KEY: &str = "last_catalog_sync";
 
+/// How long an index stays fresh before a re-index is due.
+///
+/// This lives in Rust rather than being a frontend constant because it decides
+/// *whether the local cache is authoritative* — the same class of decision as
+/// `include_catalog_browse`, and squarely the "sync policy" the spec review
+/// checklist keeps out of the presentation layer. If it later becomes
+/// user-configurable it stays a Rust-owned setting edited through a command.
+const CATALOG_INDEX_TTL: Duration = Duration::from_secs(6 * 60 * 60);
+
+/// How often the scheduler wakes to *check* staleness. Far shorter than the TTL
+/// because a tick is nearly free — one indexed `app_settings` lookup — and it is
+/// what makes the indexer responsive to events it cannot subscribe to: signing
+/// in, and coming back online. The TTL, not the tick, decides whether a crawl
+/// actually happens.
+const CATALOG_INDEX_TICK: Duration = Duration::from_secs(5 * 60);
+
+/// Delay before the first staleness check, to let sign-in complete and the
+/// repository be registered. Without it the first check runs against an empty
+/// repository manager and a fresh install would sit unindexed until the next
+/// tick.
+const CATALOG_INDEX_FIRST_CHECK: Duration = Duration::from_secs(15);
+
+/// Kebab-case, per the project's event convention.
+pub const CATALOG_INDEX_EVENT: &str = "catalog-index-event";
+
+/// Guards against two passes running at once. Replaces the frontend's
+/// `syncInProgress` boolean in `offlineCatalog.ts`, which could not see a pass
+/// started by the scheduler.
+static INDEX_IN_PROGRESS: AtomicBool = AtomicBool::new(false);
+
+/// Clears [`INDEX_IN_PROGRESS`] however the pass leaves — including on the `?`
+/// early return when `get_libraries` fails, which a plain store at the end of
+/// the function would leak.
+struct IndexPassGuard;
+
+impl Drop for IndexPassGuard {
+    fn drop(&mut self) {
+        INDEX_IN_PROGRESS.store(false, Ordering::SeqCst);
+    }
+}
+
+/// Progress of a background index pass, for the staleness hint in the UI.
+#[derive(specta::Type, Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CatalogIndexEvent {
+    /// `started` | `finished` | `failed`
+    pub state: String,
+    pub items_cached: usize,
+    pub items_pruned: usize,
+    pub libraries_failed: usize,
+    /// Present on `failed`.
+    pub error: Option<String>,
+}
+
 /// Item types worth caching for offline browsing: containers the library
 /// landing pages render plus the playable leaves users queue for download.
+/// `MusicArtist` and `Playlist` are here because search groups results by them
+/// (UR-060's Artists group). Without them in the crawl, the local index can
+/// never answer an artist query and those groups can only ever be filled by the
+/// server leg. Keep this in step with what `prune_stale_catalog` is allowed to
+/// sweep — the crawl is only authoritative for the types it asks for.
+///
+/// TRACES: UR-065, UR-060 | DR-111
 const CATALOG_ITEM_TYPES: &[&str] = &[
     "MusicAlbum",
+    "MusicArtist",
     "Movie",
     "Series",
     "Season",
     "Episode",
     "Audio",
     "BoxSet",
+    "Playlist",
 ];
 
 #[derive(specta::Type, Debug, Clone, serde::Serialize, serde::Deserialize)]
@@ -48,6 +113,9 @@ pub struct CatalogSyncResult {
     pub items_cached: usize,
     /// Libraries that failed to sync (e.g. server hiccup); best-effort.
     pub libraries_failed: usize,
+    /// Entries removed because the server no longer has them. Always 0 when any
+    /// library failed, since a partial crawl cannot prove an item is gone.
+    pub items_pruned: usize,
 }
 
 #[derive(specta::Type, Debug, Clone, serde::Serialize, serde::Deserialize)]
@@ -71,8 +139,6 @@ pub async fn sync_full_catalog(
     db: State<'_, DatabaseWrapper>,
     handle: String,
 ) -> Result<CatalogSyncResult, String> {
-    use crate::repository::MediaRepository;
-
     let repo = repository.0.get(&handle).ok_or("Repository not found")?;
 
     let db_service = {
@@ -80,6 +146,27 @@ pub async fn sync_full_catalog(
         Arc::new(database.service())
     };
 
+    run_index_pass(repo, db_service).await
+}
+
+/// One full-catalog indexing pass, shared by the [`sync_full_catalog`] command
+/// and the background scheduler (DR-109) so there is exactly one implementation
+/// and one concurrency guard.
+///
+/// TRACES: UR-065 | DR-109, DR-110
+pub(crate) async fn run_index_pass(
+    repo: Arc<crate::repository::HybridRepository>,
+    db_service: Arc<crate::storage::db_service::RusqliteService>,
+) -> Result<CatalogSyncResult, String> {
+    use crate::repository::MediaRepository;
+
+    // One pass at a time. The command and the scheduler can both land here, and
+    // two concurrent crawls would double the server load and race on the sweep.
+    if INDEX_IN_PROGRESS.swap(true, Ordering::SeqCst) {
+        return Err("A catalog index pass is already running".to_string());
+    }
+    let _guard = IndexPassGuard;
+
     let libraries = repo.get_libraries().await.map_err(|e| e.to_string())?;
     info!(
         "[Catalog] Full sync starting across {} libraries",
@@ -88,6 +175,10 @@ pub async fn sync_full_catalog(
 
     let include_types: Vec<String> = CATALOG_ITEM_TYPES.iter().map(|s| s.to_string()).collect();
 
+    // Taken before the crawl: every row the crawl writes gets a `synced_at`
+    // newer than this, so anything still older afterwards is gone server-side.
+    let pass_started_at = chrono::Utc::now().to_rfc3339();
+
     let mut items_cached = 0usize;
     let mut libraries_failed = 0usize;
 
@@ -118,6 +209,35 @@ pub async fn sync_full_catalog(
         }
     }
 
+    // Propagate server-side deletions — but only after a *complete* crawl.
+    // `sync_full_catalog` is best-effort per library, and `items.parent_id` is
+    // ON DELETE CASCADE, so sweeping when a library failed to fetch could
+    // cascade a whole series away because one request timed out.
+    let mut items_pruned = 0usize;
+    if libraries_failed == 0 && !libraries.is_empty() {
+        match repo
+            .prune_stale_catalog(&pass_started_at, &include_types)
+            .await
+        {
+            Ok(removed) => {
+                items_pruned = removed;
+                if removed > 0 {
+                    info!(
+                        "[Catalog] Pruned {} entries no longer on the server",
+                        removed
+                    );
+                }
+            }
+            Err(e) => warn!("[Catalog] Prune of stale catalog entries failed: {:?}", e),
+        }
+    } else if libraries_failed > 0 {
+        info!(
+            "[Catalog] Skipping stale-entry prune: {} librar{} failed to sync, so the crawl is not authoritative",
+            libraries_failed,
+            if libraries_failed == 1 { "y" } else { "ies" }
+        );
+    }
+
     // Record the sync time so callers can skip re-syncing too eagerly.
     let now = chrono::Utc::now().to_rfc3339();
     let upsert = Query::with_params(
@@ -133,16 +253,169 @@ pub async fn sync_full_catalog(
     }
 
     info!(
-        "[Catalog] Full sync complete: {} items cached, {} libraries failed",
-        items_cached, libraries_failed
+        "[Catalog] Full sync complete: {} items cached, {} pruned, {} libraries failed",
+        items_cached, items_pruned, libraries_failed
     );
 
     Ok(CatalogSyncResult {
         items_cached,
         libraries_failed,
+        items_pruned,
     })
 }
 
+/// Whether an index pass is due, given when one last completed.
+///
+/// Pure so the policy is unit-testable without a clock, a server, or a database.
+/// `None` (never indexed) and an unparseable stored value both mean "due" — a
+/// corrupt timestamp should trigger a re-index, not silently freeze the catalog.
+///
+/// TRACES: UR-065 | DR-109 | UT-115
+pub(crate) fn index_is_due(
+    last_synced_at: Option<&str>,
+    now: chrono::DateTime<chrono::Utc>,
+    ttl: Duration,
+) -> bool {
+    let Some(raw) = last_synced_at else {
+        return true;
+    };
+    let Ok(last) = chrono::DateTime::parse_from_rfc3339(raw) else {
+        return true;
+    };
+    now.signed_duration_since(last.with_timezone(&chrono::Utc))
+        .to_std()
+        .map(|elapsed| elapsed >= ttl)
+        // Negative elapsed => the stored stamp is in the future (clock skew).
+        // Not due; a future stamp will age into due-ness on its own.
+        .unwrap_or(false)
+}
+
+/// Read the last-sync timestamp straight from `app_settings`.
+async fn read_last_sync(
+    db_service: &Arc<crate::storage::db_service::RusqliteService>,
+) -> Option<String> {
+    db_service
+        .query_optional(
+            Query::with_params(
+                "SELECT value FROM app_settings WHERE key = ?",
+                vec![QueryParam::String(LAST_CATALOG_SYNC_KEY.to_string())],
+            ),
+            |row| row.get(0),
+        )
+        .await
+        .ok()
+        .flatten()
+}
+
+/// Start the background catalog indexer.
+///
+/// Replaces the frontend's startup-only `syncCatalog()` call: index freshness is
+/// sync policy and belongs in Rust (see the layer assignment in
+/// docs/specs/catalog-index-search.md). Ticks every [`CATALOG_INDEX_TICK`] and
+/// runs a pass when a repository exists, the server is reachable, and the index
+/// is older than [`CATALOG_INDEX_TTL`].
+///
+/// TRACES: UR-065 | DR-109, IR-030
+pub fn spawn_catalog_indexer(app: tauri::AppHandle) {
+    tauri::async_runtime::spawn(async move {
+        // Check shortly after launch, then on every tick — not tick-then-check,
+        // which would leave a fresh install unindexed for a full tick.
+        tokio::time::sleep(CATALOG_INDEX_FIRST_CHECK).await;
+
+        loop {
+            if let Err(e) = maybe_run_scheduled_pass(&app).await {
+                // Never fatal — a failed pass leaves the existing index in place
+                // and we retry on the next tick.
+                warn!("[Catalog] Scheduled index pass skipped: {}", e);
+            }
+
+            tokio::time::sleep(CATALOG_INDEX_TICK).await;
+        }
+    });
+}
+
+/// One scheduler tick: check the preconditions, then index if due.
+async fn maybe_run_scheduled_pass(app: &tauri::AppHandle) -> Result<(), String> {
+    if INDEX_IN_PROGRESS.load(Ordering::SeqCst) {
+        return Ok(());
+    }
+
+    let db_service = {
+        let db = app.state::<DatabaseWrapper>();
+        let database = db.0.lock().map_err(|e| e.to_string())?;
+        Arc::new(database.service())
+    };
+
+    if !index_is_due(
+        read_last_sync(&db_service).await.as_deref(),
+        chrono::Utc::now(),
+        CATALOG_INDEX_TTL,
+    ) {
+        return Ok(());
+    }
+
+    // Offline: leave the index alone. The crawl would fail every library and,
+    // more importantly, a partial crawl must never reach the deletion sweep.
+    {
+        let monitor = app.state::<crate::commands::connectivity::ConnectivityMonitorWrapper>();
+        let monitor = monitor.0.lock().await;
+        if !monitor.get_status().await.is_server_reachable {
+            return Ok(());
+        }
+    }
+
+    let repo = {
+        let manager = app.state::<RepositoryManagerWrapper>();
+        let handles = manager.0.handles();
+        let Some(handle) = handles.first() else {
+            // Not signed in yet.
+            return Ok(());
+        };
+        manager.0.get(handle).ok_or("Repository not found")?
+    };
+
+    info!("[Catalog] Index is stale; starting a scheduled pass");
+    let _ = app.emit(
+        CATALOG_INDEX_EVENT,
+        CatalogIndexEvent {
+            state: "started".to_string(),
+            items_cached: 0,
+            items_pruned: 0,
+            libraries_failed: 0,
+            error: None,
+        },
+    );
+
+    match run_index_pass(repo, db_service).await {
+        Ok(result) => {
+            let _ = app.emit(
+                CATALOG_INDEX_EVENT,
+                CatalogIndexEvent {
+                    state: "finished".to_string(),
+                    items_cached: result.items_cached,
+                    items_pruned: result.items_pruned,
+                    libraries_failed: result.libraries_failed,
+                    error: None,
+                },
+            );
+            Ok(())
+        }
+        Err(e) => {
+            let _ = app.emit(
+                CATALOG_INDEX_EVENT,
+                CatalogIndexEvent {
+                    state: "failed".to_string(),
+                    items_cached: 0,
+                    items_pruned: 0,
+                    libraries_failed: 0,
+                    error: Some(e.clone()),
+                },
+            );
+            Err(e)
+        }
+    }
+}
+
 /// Report the last-synced timestamp so the UI can show a hint / decide whether
 /// to trigger a fresh sync.
 #[tauri::command]
@@ -378,6 +651,43 @@ mod tests {
     use rusqlite::Connection;
     use std::sync::Mutex;
 
+    /// The re-index policy. Pure, so it is testable without a clock, a server or
+    /// a database — which is the reason it was factored out of the scheduler.
+    ///
+    /// TRACES: UR-065 | DR-109 | UT-115
+    #[test]
+    fn test_index_is_due() {
+        let ttl = Duration::from_secs(6 * 60 * 60);
+        let now = chrono::DateTime::parse_from_rfc3339("2026-08-04T12:00:00+00:00")
+            .unwrap()
+            .with_timezone(&chrono::Utc);
+
+        // Never indexed => due. This is the first-run case.
+        assert!(index_is_due(None, now, ttl));
+
+        // Indexed 7 hours ago => past the 6h TTL => due.
+        assert!(index_is_due(Some("2026-08-04T05:00:00+00:00"), now, ttl));
+
+        // Indexed 1 hour ago => fresh => not due. This is what stops the
+        // scheduler re-crawling every tick.
+        assert!(!index_is_due(Some("2026-08-04T11:00:00+00:00"), now, ttl));
+
+        // Exactly at the TTL boundary counts as due.
+        assert!(index_is_due(Some("2026-08-04T06:00:00+00:00"), now, ttl));
+
+        // A corrupt stored value must trigger a re-index, not freeze the
+        // catalog forever behind an unparseable timestamp.
+        assert!(index_is_due(Some("not-a-timestamp"), now, ttl));
+        assert!(index_is_due(Some(""), now, ttl));
+
+        // A timestamp in the future (clock skew, or a restored backup) is not
+        // due — it ages into due-ness rather than causing a crawl every tick.
+        assert!(!index_is_due(Some("2026-08-05T00:00:00+00:00"), now, ttl));
+
+        // Offsets other than UTC are compared as instants, not as strings.
+        assert!(!index_is_due(Some("2026-08-04T13:30:00+02:00"), now, ttl));
+    }
+
     fn test_db() -> Arc<RusqliteService> {
         let conn = Connection::open_in_memory().unwrap();
         conn.execute_batch(

src-tauri/src/commands/download/mod.rs

@@ -270,6 +270,19 @@ pub async fn download_item(
         if !can_download {
             warn!("Storage limit reached. Attempting to free space...");
 
+            // Reclaim expired temporary entries first: they are dead weight, so
+            // freeing them may avoid evicting cache that is still within its
+            // life. Best-effort — a failure here just means eviction does more.
+            // TRACES: UR-071 | DR-127
+            match cache_arc
+                .reclaim_expired_async(&db_service, &user_id, &chrono::Utc::now().to_rfc3339())
+                .await
+            {
+                Ok(n) if n > 0 => info!("Reclaimed {} expired cache entries", n),
+                Ok(_) => {}
+                Err(e) => warn!("Expired-entry reclaim failed: {}", e),
+            }
+
             // Try to evict LRU items to make space
             match cache_arc
                 .evict_lru_async(&db_service, &user_id, size as u64)

src-tauri/src/commands/favorites.rs

@@ -0,0 +1,367 @@
+//! Pushing favourite toggles made while the server was unreachable.
+//!
+//! Favouriting works offline: `storage_toggle_favorite` writes the local
+//! `user_data` row and sets `pending_sync = 1`. Until DR-120 nothing ever
+//! cleared that flag — the offline `mark_favorite`/`unmark_favorite` are no-ops
+//! and `syncService.queueFavorite` had no callers — so an offline toggle was
+//! silently lost.
+//!
+//! The drain lives in Rust, not the frontend, because it must run whether or
+//! not any view is mounted; a drain started by a component dies with it.
+//!
+//! TRACES: UR-069 | DR-120 | UT-103
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use log::{debug, info, warn};
+use tauri::{Emitter, Listener, Manager};
+
+use crate::repository::types::RepoError;
+use crate::repository::MediaRepository;
+use crate::storage::db_service::{DatabaseService, Query, QueryParam, RusqliteService};
+
+/// The subset of the repository the drain needs.
+///
+/// Narrow on purpose: a test double for `MediaRepository` would be forty
+/// unimplemented methods, which is how a drain ends up untested.
+#[async_trait]
+pub trait FavoriteSink: Send + Sync {
+    async fn push_favorite(&self, item_id: &str, is_favorite: bool) -> Result<(), RepoError>;
+}
+
+#[async_trait]
+impl<T: MediaRepository + ?Sized> FavoriteSink for T {
+    async fn push_favorite(&self, item_id: &str, is_favorite: bool) -> Result<(), RepoError> {
+        if is_favorite {
+            self.mark_favorite(item_id).await
+        } else {
+            self.unmark_favorite(item_id).await
+        }
+    }
+}
+
+/// A local favourite change still waiting to reach the server.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct PendingFavorite {
+    pub item_id: String,
+    pub is_favorite: bool,
+}
+
+/// Read every favourite change this user has pending.
+async fn read_pending(
+    db: &Arc<RusqliteService>,
+    user_id: &str,
+) -> Result<Vec<PendingFavorite>, String> {
+    db.query_many(
+        Query::with_params(
+            "SELECT item_id, is_favorite FROM user_data \
+             WHERE user_id = ? AND pending_sync = 1 AND is_favorite IS NOT NULL",
+            vec![QueryParam::String(user_id.to_string())],
+        ),
+        |row| {
+            Ok(PendingFavorite {
+                item_id: row.get::<_, String>(0)?,
+                is_favorite: row.get::<_, Option<i32>>(1)?.unwrap_or(0) != 0,
+            })
+        },
+    )
+    .await
+}
+
+/// Push pending favourite changes to the server and clear their flags.
+///
+/// Returns the ids that reached the server, for the `favorites-changed` event.
+/// A row whose push fails keeps `pending_sync = 1` and is retried on the next
+/// reconnect rather than being dropped.
+///
+/// TRACES: UR-069 | DR-120 | UT-103
+pub async fn drain_pending_favorites(
+    db: &Arc<RusqliteService>,
+    sink: &dyn FavoriteSink,
+    user_id: &str,
+) -> Result<Vec<String>, String> {
+    let pending = read_pending(db, user_id).await?;
+    if pending.is_empty() {
+        return Ok(Vec::new());
+    }
+
+    info!(
+        "[Favorites] Pushing {} favourite change(s) queued while offline",
+        pending.len()
+    );
+
+    let mut pushed = Vec::new();
+    for change in pending {
+        match sink
+            .push_favorite(&change.item_id, change.is_favorite)
+            .await
+        {
+            Ok(()) => {
+                let cleared = db
+                    .execute(Query::with_params(
+                        "UPDATE user_data SET pending_sync = 0, synced_at = ? \
+                         WHERE user_id = ? AND item_id = ?",
+                        vec![
+                            QueryParam::String(chrono::Utc::now().to_rfc3339()),
+                            QueryParam::String(user_id.to_string()),
+                            QueryParam::String(change.item_id.clone()),
+                        ],
+                    ))
+                    .await;
+
+                match cleared {
+                    Ok(_) => pushed.push(change.item_id),
+                    // The server took it; failing to clear the flag only means
+                    // we push it again next time, which is harmless.
+                    Err(e) => warn!(
+                        "[Favorites] Pushed {} but could not clear pending_sync: {}",
+                        change.item_id, e
+                    ),
+                }
+            }
+            Err(e) => {
+                // Still pending — retried on the next reconnect.
+                debug!(
+                    "[Favorites] Deferring {}, server rejected the push: {:?}",
+                    change.item_id, e
+                );
+            }
+        }
+    }
+
+    Ok(pushed)
+}
+
+/// Drain on every offline→online transition.
+///
+/// Hooks the `connectivity:reconnected` event the `ConnectivityMonitor`
+/// already emits, rather than polling — reachability is derived from real
+/// traffic (DR-055) and this just reacts to it.
+///
+/// TRACES: UR-069 | DR-120
+pub fn spawn_favorites_drain(app: tauri::AppHandle) {
+    let handle = app.clone();
+    app.listen("connectivity:reconnected", move |_event| {
+        let app = handle.clone();
+        tauri::async_runtime::spawn(async move {
+            if let Err(e) = run_drain(&app).await {
+                warn!("[Favorites] Drain skipped: {}", e);
+            }
+        });
+    });
+}
+
+async fn run_drain(app: &tauri::AppHandle) -> Result<(), String> {
+    let db_service: Arc<RusqliteService> = {
+        let db = app.state::<crate::commands::storage::DatabaseWrapper>();
+        let database = db.0.lock().map_err(|e| e.to_string())?;
+        Arc::new(database.service())
+    };
+
+    let (repo, user_id) = {
+        let manager = app.state::<crate::commands::repository::RepositoryManagerWrapper>();
+        let handles = manager.0.handles();
+        let Some(handle) = handles.first() else {
+            // Not signed in — nothing to push on behalf of.
+            return Ok(());
+        };
+        let repo = manager.0.get(handle).ok_or("Repository not found")?;
+        let user_id = repo.user_id().to_string();
+        (repo, user_id)
+    };
+
+    let pushed = drain_pending_favorites(&db_service, repo.as_ref(), &user_id).await?;
+
+    if !pushed.is_empty() {
+        let event = crate::commands::repository::FavoritesChangedEvent { item_ids: pushed };
+        if let Err(e) = app.emit(crate::commands::repository::FAVORITES_CHANGED_EVENT, &event) {
+            warn!("[Favorites] Failed to emit change event: {}", e);
+        }
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rusqlite::Connection;
+    use std::sync::Mutex;
+
+    /// Records what the server was asked to do, and can be told to fail.
+    struct RecordingSink {
+        calls: Mutex<Vec<(String, bool)>>,
+        fail_for: Option<String>,
+    }
+
+    impl RecordingSink {
+        fn new() -> Self {
+            Self {
+                calls: Mutex::new(Vec::new()),
+                fail_for: None,
+            }
+        }
+
+        fn failing_for(item_id: &str) -> Self {
+            Self {
+                calls: Mutex::new(Vec::new()),
+                fail_for: Some(item_id.to_string()),
+            }
+        }
+
+        fn calls(&self) -> Vec<(String, bool)> {
+            let mut calls = self.calls.lock().unwrap().clone();
+            calls.sort();
+            calls
+        }
+    }
+
+    #[async_trait]
+    impl FavoriteSink for RecordingSink {
+        async fn push_favorite(&self, item_id: &str, is_favorite: bool) -> Result<(), RepoError> {
+            if self.fail_for.as_deref() == Some(item_id) {
+                return Err(RepoError::Offline);
+            }
+            self.calls
+                .lock()
+                .unwrap()
+                .push((item_id.to_string(), is_favorite));
+            Ok(())
+        }
+    }
+
+    fn test_db() -> Arc<RusqliteService> {
+        let conn = Connection::open_in_memory().unwrap();
+        conn.execute_batch(
+            r#"
+            CREATE TABLE user_data (
+                user_id TEXT NOT NULL,
+                item_id TEXT NOT NULL,
+                is_favorite INTEGER,
+                synced_at TEXT,
+                pending_sync INTEGER DEFAULT 0,
+                PRIMARY KEY (user_id, item_id)
+            );
+            "#,
+        )
+        .unwrap();
+        Arc::new(RusqliteService::new(Arc::new(Mutex::new(conn))))
+    }
+
+    async fn seed(db: &Arc<RusqliteService>, rows: &[(&str, &str, i32, i32)]) {
+        for (user, item, fav, pending) in rows {
+            db.execute(Query::with_params(
+                "INSERT INTO user_data (user_id, item_id, is_favorite, pending_sync) \
+                 VALUES (?, ?, ?, ?)",
+                vec![
+                    QueryParam::String(user.to_string()),
+                    QueryParam::String(item.to_string()),
+                    QueryParam::Int(*fav),
+                    QueryParam::Int(*pending),
+                ],
+            ))
+            .await
+            .unwrap();
+        }
+    }
+
+    async fn pending_flag(db: &Arc<RusqliteService>, item_id: &str) -> Option<i32> {
+        db.query_optional(
+            Query::with_params(
+                "SELECT pending_sync FROM user_data WHERE item_id = ?",
+                vec![QueryParam::String(item_id.to_string())],
+            ),
+            |row| row.get::<_, Option<i32>>(0),
+        )
+        .await
+        .unwrap()
+        .flatten()
+    }
+
+    /// UT-103 — the core of the bug: a favourite toggled while offline reaches
+    /// the server on reconnect, and stops being pending.
+    ///
+    /// TRACES: UR-069 | DR-120 | UT-103
+    #[tokio::test]
+    async fn test_drain_pushes_pending_favorites_and_clears_the_flag() {
+        let db = test_db();
+        seed(
+            &db,
+            &[
+                ("u1", "marked-offline", 1, 1),
+                ("u1", "unmarked-offline", 0, 1),
+                // Already synced — must not be pushed again.
+                ("u1", "already-synced", 1, 0),
+            ],
+        )
+        .await;
+
+        let sink = RecordingSink::new();
+        let pushed = drain_pending_favorites(&db, &sink, "u1").await.unwrap();
+
+        assert_eq!(
+            sink.calls(),
+            vec![
+                ("marked-offline".to_string(), true),
+                ("unmarked-offline".to_string(), false),
+            ],
+            "both pending changes push, with their direction preserved"
+        );
+        assert_eq!(pushed.len(), 2);
+        assert_eq!(pending_flag(&db, "marked-offline").await, Some(0));
+        assert_eq!(pending_flag(&db, "unmarked-offline").await, Some(0));
+    }
+
+    /// A push that fails keeps its row pending, so the change is retried rather
+    /// than dropped on the floor.
+    ///
+    /// TRACES: UR-069 | DR-120 | UT-103
+    #[tokio::test]
+    async fn test_drain_leaves_failed_pushes_pending() {
+        let db = test_db();
+        seed(&db, &[("u1", "ok", 1, 1), ("u1", "boom", 1, 1)]).await;
+
+        let sink = RecordingSink::failing_for("boom");
+        let pushed = drain_pending_favorites(&db, &sink, "u1").await.unwrap();
+
+        assert_eq!(pushed, vec!["ok".to_string()]);
+        assert_eq!(pending_flag(&db, "ok").await, Some(0));
+        assert_eq!(
+            pending_flag(&db, "boom").await,
+            Some(1),
+            "a failed push must stay queued for the next reconnect"
+        );
+    }
+
+    /// Another user's queued changes are not pushed with this user's token.
+    ///
+    /// TRACES: UR-069 | DR-120 | UT-103
+    #[tokio::test]
+    async fn test_drain_only_touches_the_given_user() {
+        let db = test_db();
+        seed(&db, &[("u1", "mine", 1, 1), ("u2", "theirs", 1, 1)]).await;
+
+        let sink = RecordingSink::new();
+        let pushed = drain_pending_favorites(&db, &sink, "u1").await.unwrap();
+
+        assert_eq!(pushed, vec!["mine".to_string()]);
+        assert_eq!(pending_flag(&db, "theirs").await, Some(1));
+    }
+
+    /// Nothing pending means no server calls at all — a reconnect must not
+    /// generate traffic just because it happened.
+    ///
+    /// TRACES: UR-069 | DR-120 | UT-103
+    #[tokio::test]
+    async fn test_drain_is_a_noop_when_nothing_is_pending() {
+        let db = test_db();
+        seed(&db, &[("u1", "synced", 1, 0)]).await;
+
+        let sink = RecordingSink::new();
+        let pushed = drain_pending_favorites(&db, &sink, "u1").await.unwrap();
+
+        assert!(pushed.is_empty());
+        assert!(sink.calls().is_empty());
+    }
+}

src-tauri/src/commands/mod.rs

@@ -7,6 +7,7 @@ pub mod connectivity;
 pub mod conversions;
 pub mod device;
 pub mod download;
+pub mod favorites;
 pub mod offline;
 pub mod playback_mode;
 pub mod playback_reporting;

src-tauri/src/commands/player/mod.rs

@@ -379,16 +379,49 @@ pub(super) async fn create_media_item(
     })
 }
 
-/// Check if an item has a completed download
-pub(super) async fn check_for_local_download(
-    db: &DatabaseWrapper,
+/// Pick the source for an audio-only handoff.
+///
+/// A downloaded file wins over the audio-only stream URL. No transcode or audio
+/// extraction is involved or wanted: the native backends already play a video
+/// container without decoding its video — the Linux MPV backend is configured
+/// with `video: no`, and ExoPlayer simply has no surface to render to when the
+/// item is `MediaType::Audio`. Producing a separate audio-only file would cost
+/// CPU and battery, need an encoder the project does not ship, and leave a
+/// second artifact to keep in step with the first.
+///
+/// TRACES: UR-071 | DR-128 | UT-119
+pub(super) fn background_audio_source(
+    local_path: Option<String>,
+    stream_url: String,
+    item_id: &str,
+) -> MediaSource {
+    match local_path {
+        Some(path) => MediaSource::Local {
+            file_path: PathBuf::from(path),
+            jellyfin_item_id: Some(item_id.to_string()),
+        },
+        None => MediaSource::Remote {
+            stream_url,
+            jellyfin_item_id: item_id.to_string(),
+        },
+    }
+}
+
+/// Resolve the on-disk file backing a completed download, if there is one.
+///
+/// A `downloads` row is not proof of a file: it can outlive the bytes (manual
+/// deletion, a cleared cache directory, a restored database). Every caller wants
+/// "can I play this from disk right now", so existence is checked here rather
+/// than trusted from the row.
+///
+/// Split out from [`check_for_local_download`] so the resolution is testable
+/// without a `DatabaseWrapper`, and reusable by the video path.
+///
+/// TRACES: UR-071 | DR-123 | UT-116
+pub(super) async fn resolve_local_media_path<S: DatabaseService>(
+    db_service: &Arc<S>,
     item_id: &str,
 ) -> Result<Option<String>, String> {
-    let db_service = {
-        let database = db.0.lock().map_err(|e| e.to_string())?;
-        Arc::new(database.service())
-    };
-
     let query = Query::with_params(
         "SELECT file_path FROM downloads WHERE item_id = ? AND status = 'completed' LIMIT 1",
         vec![QueryParam::String(item_id.to_string())],
@@ -399,22 +432,58 @@ pub(super) async fn check_for_local_download(
         .await
         .map_err(|e| e.to_string())?;
 
-    // Verify the file actually exists on disk
-    if let Some(ref file_path) = path {
-        if std::path::Path::new(file_path).exists() {
-            Ok(path)
-        } else {
+    match path {
+        Some(ref file_path) if std::path::Path::new(file_path).exists() => Ok(path),
+        Some(file_path) => {
             warn!(
                 "[Player] Download entry exists in DB but file not found: {}",
                 file_path
             );
             Ok(None)
         }
-    } else {
-        Ok(None)
+        None => Ok(None),
     }
 }
 
+/// Check if an item has a completed download
+pub(super) async fn check_for_local_download(
+    db: &DatabaseWrapper,
+    item_id: &str,
+) -> Result<Option<String>, String> {
+    let db_service = {
+        let database = db.0.lock().map_err(|e| e.to_string())?;
+        Arc::new(database.service())
+    };
+
+    resolve_local_media_path(&db_service, item_id).await
+}
+
+/// The on-disk path for a downloaded item, for playback surfaces that resolve
+/// their own source rather than going through the queue.
+///
+/// The video player is the reason this exists: audio has preferred local files
+/// since queue construction, but video asks the repository for a stream URL and
+/// never consults `downloads`, so a downloaded film was still streamed — costing
+/// bandwidth that had already been spent and failing outright when offline.
+///
+/// Returns `None` when nothing is downloaded *or* the file is missing, so the
+/// caller falls back to streaming.
+///
+/// TRACES: UR-071 | DR-123 | UT-116
+#[tauri::command]
+#[specta::specta]
+pub async fn player_local_media_path(
+    db: State<'_, DatabaseWrapper>,
+    item_id: String,
+) -> Result<Option<String>, String> {
+    let db_service = {
+        let database = db.0.lock().map_err(|e| e.to_string())?;
+        Arc::new(database.service())
+    };
+
+    resolve_local_media_path(&db_service, &item_id).await
+}
+
 /// Re-point queued streaming items at completed local downloads.
 ///
 /// Sources are resolved once when the queue is built, so downloads that finish
@@ -574,6 +643,7 @@ pub async fn player_play_item(
 pub async fn player_enter_background_audio(
     player: State<'_, PlayerStateWrapper>,
     session: State<'_, MediaSessionManagerWrapper>,
+    db: State<'_, DatabaseWrapper>,
     item: PlayItemRequest,
     position_seconds: f64,
 ) -> Result<PlayerStatus, String> {
@@ -582,6 +652,19 @@ pub async fn player_enter_background_audio(
         item.title, position_seconds
     );
 
+    // Prefer the downloaded file over the audio-only stream URL the frontend
+    // resolved. Handing the native backend a local video container yields
+    // audio-only playback for free — no transcode, no second artifact.
+    // TRACES: UR-071 | DR-128
+    let local_path = check_for_local_download(&db, &item.id).await?;
+    if local_path.is_some() {
+        info!(
+            "player_enter_background_audio: using downloaded file for {}",
+            item.id
+        );
+    }
+    let source = background_audio_source(local_path, item.stream_url, &item.id);
+
     // Build an AUDIO media item pointing at the audio-only stream. We do not use
     // create_media_item() because that hardcodes MediaType::Video; background
     // audio must be Audio so no video decode is started.
@@ -605,10 +688,7 @@ pub async fn player_enter_background_audio(
         duration: item.duration_seconds,
         artwork_url: None,
         media_type: MediaType::Audio,
-        source: MediaSource::Remote {
-            stream_url: item.stream_url,
-            jellyfin_item_id: item.id.clone(),
-        },
+        source,
         video_codec: None,
         needs_transcoding: false,
         video_width: None,
@@ -2379,6 +2459,128 @@ pub async fn player_disable_jellyfin(player: State<'_, PlayerStateWrapper>) -> R
 
 #[cfg(test)]
 mod tests {
+    /// The audio-only handoff must play a downloaded file when there is one,
+    /// rather than fetching an audio-only stream for media already on disk.
+    ///
+    /// TRACES: UR-071 | DR-128 | UT-119
+    #[test]
+    fn test_background_audio_source_prefers_local_file() {
+        use super::background_audio_source;
+        use crate::player::MediaSource;
+        use std::path::PathBuf;
+
+        let local = background_audio_source(
+            Some("/downloads/ep1.mkv".to_string()),
+            "https://server/audio-only".to_string(),
+            "ep-1",
+        );
+        match local {
+            MediaSource::Local {
+                file_path,
+                jellyfin_item_id,
+            } => {
+                assert_eq!(file_path, PathBuf::from("/downloads/ep1.mkv"));
+                // The Jellyfin id must survive so progress still syncs back.
+                assert_eq!(jellyfin_item_id.as_deref(), Some("ep-1"));
+            }
+            other => panic!("expected a local source, got {:?}", other),
+        }
+
+        let remote = background_audio_source(None, "https://server/audio-only".to_string(), "ep-1");
+        match remote {
+            MediaSource::Remote {
+                stream_url,
+                jellyfin_item_id,
+            } => {
+                assert_eq!(stream_url, "https://server/audio-only");
+                assert_eq!(jellyfin_item_id, "ep-1");
+            }
+            other => panic!("expected a remote source, got {:?}", other),
+        }
+    }
+
+    /// A downloaded item must resolve to its file, and a `downloads` row whose
+    /// file has gone must resolve to `None` so the caller falls back to
+    /// streaming instead of handing the player a path that cannot be opened.
+    ///
+    /// TRACES: UR-071 | DR-123 | UT-116
+    #[tokio::test]
+    async fn test_resolve_local_media_path() {
+        use super::resolve_local_media_path;
+        use crate::storage::db_service::{DatabaseService, Query, RusqliteService};
+        use rusqlite::Connection;
+        use std::sync::{Arc, Mutex};
+
+        let conn = Connection::open_in_memory().unwrap();
+        conn.execute(
+            "CREATE TABLE downloads (id INTEGER PRIMARY KEY, item_id TEXT, status TEXT, file_path TEXT)",
+            [],
+        )
+        .unwrap();
+        let db_service = Arc::new(RusqliteService::new(Arc::new(Mutex::new(conn))));
+
+        // A real file on disk, so the existence check passes.
+        let present = std::env::temp_dir().join("jellytau-resolve-local-test.mp4");
+        std::fs::write(&present, b"x").unwrap();
+        let present_str = present.to_string_lossy().to_string();
+
+        for (item, status, path) in [
+            ("downloaded", "completed", present_str.as_str()),
+            ("still-going", "downloading", present_str.as_str()),
+            (
+                "file-gone",
+                "completed",
+                "/nonexistent/jellytau/missing.mp4",
+            ),
+        ] {
+            db_service
+                .execute(Query::with_params(
+                    "INSERT INTO downloads (item_id, status, file_path) VALUES (?, ?, ?)",
+                    vec![
+                        crate::storage::db_service::QueryParam::String(item.to_string()),
+                        crate::storage::db_service::QueryParam::String(status.to_string()),
+                        crate::storage::db_service::QueryParam::String(path.to_string()),
+                    ],
+                ))
+                .await
+                .unwrap();
+        }
+
+        assert_eq!(
+            resolve_local_media_path(&db_service, "downloaded")
+                .await
+                .unwrap()
+                .as_deref(),
+            Some(present_str.as_str()),
+            "a completed download with its file present must resolve"
+        );
+
+        assert_eq!(
+            resolve_local_media_path(&db_service, "still-going")
+                .await
+                .unwrap(),
+            None,
+            "an in-progress download is not playable from disk"
+        );
+
+        assert_eq!(
+            resolve_local_media_path(&db_service, "file-gone")
+                .await
+                .unwrap(),
+            None,
+            "a row whose file has gone must fall back to streaming, not hand over a dead path"
+        );
+
+        assert_eq!(
+            resolve_local_media_path(&db_service, "never-heard-of-it")
+                .await
+                .unwrap(),
+            None
+        );
+
+        let _ = std::fs::remove_file(&present);
+    }
+
     /// Queue items enqueued as Remote must flip to Local once a completed
     /// download exists on disk — this is what makes preloaded tracks (and
     /// offline playback after a connection drop) actually use the cache.

src-tauri/src/commands/player/timers.rs

@@ -142,7 +142,7 @@ pub async fn player_play_next_episode(
 /// - Frontend when audio track ends via backend event - no itemId/repositoryHandle needed
 /// - Android JNI callback also triggers this logic directly
 ///
-/// TRACES: UR-023, UR-026, UR-040 | DR-047, DR-052
+/// TRACES: UR-023, UR-026, UR-040 | DR-047, DR-052, DR-129
 #[tauri::command]
 #[specta::specta]
 pub async fn player_on_playback_ended(
@@ -257,6 +257,23 @@ pub async fn player_on_playback_ended(
                     .await;
             }
         }
+        AutoplayDecision::ResumeStream { position } => {
+            // The stream was cut short by the network, not by the media ending.
+            // Re-open it where it died — no queue clearing, no PlaybackEnded, and
+            // above all no leaving the player parked in ExoPlayer's STATE_ENDED,
+            // where the next play intent restarts the item from 0:00.
+            log::info!(
+                "[Autoplay] Decision: Resume truncated stream at {:.1}s",
+                position
+            );
+            let controller = controller_arc.lock().await;
+            if let Err(e) = controller.resume_stream_at(position).await {
+                log::error!("[Autoplay] Failed to resume truncated stream: {}", e);
+                if let Some(emitter) = controller.event_emitter() {
+                    emitter.emit(PlayerStatusEvent::PlaybackEnded);
+                }
+            }
+        }
     }
 
     Ok(())

src-tauri/src/commands/repository.rs

@@ -41,6 +41,19 @@ impl RepositoryManager {
         repos.get(handle).cloned()
     }
 
+    /// Handles of every live repository.
+    ///
+    /// The background catalog indexer (DR-109) runs outside any command, so it
+    /// has no handle passed in and needs to discover one. In practice there is a
+    /// single signed-in repository; returning all of them avoids inventing an
+    /// "active" concept the rest of the code does not have.
+    ///
+    /// TRACES: UR-065 | DR-109
+    pub fn handles(&self) -> Vec<String> {
+        let repos = self.repositories.lock_safe();
+        repos.keys().cloned().collect()
+    }
+
     pub fn destroy(&self, handle: &str) {
         let mut repos = self.repositories.lock_safe();
         repos.remove(handle);
@@ -780,6 +793,108 @@ pub async fn repository_mark_favorite(
         .map_err(|e| format!("{:?}", e))
 }
 
+/// Tauri event announcing that favourite state changed behind the UI's back —
+/// either because the server disagreed with the cache on a background refresh,
+/// or because pending offline toggles were pushed on reconnect.
+///
+/// TRACES: UR-069 | DR-120
+pub const FAVORITES_CHANGED_EVENT: &str = "favorites-changed";
+
+/// Payload for [`FAVORITES_CHANGED_EVENT`] — the ids whose favourite state
+/// actually flipped, so the frontend refreshes those rather than everything.
+///
+/// TRACES: UR-069 | DR-120 | UT-107
+#[derive(specta::Type, Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct FavoritesChangedEvent {
+    pub item_ids: Vec<String>,
+}
+
+/// Ids whose favourite state differs between what we showed and what the server
+/// has — favourited elsewhere since the cache was written, or un-favourited
+/// elsewhere.
+///
+/// Pulled out of the command so the "emit nothing when nothing changed" rule is
+/// testable: an unchanged set must leave a quiet page quiet rather than
+/// triggering a refetch on every visit.
+///
+/// TRACES: UR-069 | DR-120 | UT-107
+fn changed_favorite_ids(
+    cached: &std::collections::HashSet<String>,
+    server: &std::collections::HashSet<String>,
+) -> Vec<String> {
+    let mut changed: Vec<String> = server.symmetric_difference(cached).cloned().collect();
+    // Deterministic order so the event payload does not depend on hash seeding.
+    changed.sort();
+    changed
+}
+
+/// Everything the viewer has favourited, across libraries, narrowed by scope.
+///
+/// Two-phase like `repository_search`: the local answer returns immediately and
+/// a background server pass emits `favorites-changed` when the server's set
+/// differs. Without the second phase a favourite marked in another client shows
+/// up only on the *second* visit to the page, since the cache-first read hands
+/// back local rows and the refresh is invisible to the frontend.
+///
+/// TRACES: UR-067 | DR-115, DR-120, JA-033 | UT-107
+#[tauri::command]
+#[specta::specta]
+pub async fn repository_get_favorites(
+    app: AppHandle,
+    manager: State<'_, RepositoryManagerWrapper>,
+    handle: String,
+    scope: SearchScope,
+    options: Option<GetItemsOptions>,
+) -> Result<SearchResult, String> {
+    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
+
+    let cache_result = repo
+        .get_favorites_cache_only(scope, options.clone())
+        .await
+        .unwrap_or_else(|e| {
+            debug!("[Favorites] Cache miss/timeout: {:?}", e);
+            SearchResult {
+                items: Vec::new(),
+                total_record_count: 0,
+            }
+        });
+
+    // With "Show all server media" off the local answer is authoritative
+    // (DR-080) — don't go behind the user's back to the server.
+    if !crate::repository::offline::include_catalog_browse() {
+        return Ok(cache_result);
+    }
+
+    let repo_bg = repo.clone();
+    let cached_ids: std::collections::HashSet<String> =
+        cache_result.items.iter().map(|i| i.id.clone()).collect();
+    tauri::async_runtime::spawn(async move {
+        match repo_bg.get_favorites_server_only(scope, options).await {
+            Ok(server_result) => {
+                let server_ids: std::collections::HashSet<String> =
+                    server_result.items.iter().map(|i| i.id.clone()).collect();
+                let changed = changed_favorite_ids(&cached_ids, &server_ids);
+
+                if !changed.is_empty() {
+                    let event = FavoritesChangedEvent { item_ids: changed };
+                    if let Err(e) = app.emit(FAVORITES_CHANGED_EVENT, &event) {
+                        error!("[Favorites] Failed to emit change event: {}", e);
+                    }
+                }
+            }
+            Err(e) => {
+                warn!(
+                    "[Favorites] Server refresh failed, keeping cached favourites: {:?}",
+                    e
+                );
+            }
+        }
+    });
+
+    Ok(cache_result)
+}
+
 /// Unmark an item as favorite
 #[tauri::command]
 #[specta::specta]
@@ -853,6 +968,44 @@ mod tests {
         assert!(manager.get("any-handle").is_none());
     }
 
+    fn ids(values: &[&str]) -> std::collections::HashSet<String> {
+        values.iter().map(|v| v.to_string()).collect()
+    }
+
+    /// UT-107 — the background refresh reports only what actually changed.
+    ///
+    /// TRACES: UR-069 | DR-120 | UT-107
+    #[test]
+    fn test_changed_favorite_ids_reports_both_directions() {
+        // Favourited in another client since we cached.
+        assert_eq!(
+            changed_favorite_ids(&ids(&["a"]), &ids(&["a", "b"])),
+            vec!["b".to_string()]
+        );
+
+        // Un-favourited in another client.
+        assert_eq!(
+            changed_favorite_ids(&ids(&["a", "b"]), &ids(&["a"])),
+            vec!["b".to_string()]
+        );
+
+        // Both at once, in a stable order.
+        assert_eq!(
+            changed_favorite_ids(&ids(&["a", "b"]), &ids(&["b", "c"])),
+            vec!["a".to_string(), "c".to_string()]
+        );
+    }
+
+    /// An unchanged set emits nothing — otherwise every visit to the page would
+    /// fire an event and trigger a pointless refetch.
+    ///
+    /// TRACES: UR-069 | DR-120 | UT-107
+    #[test]
+    fn test_changed_favorite_ids_is_empty_when_nothing_moved() {
+        assert!(changed_favorite_ids(&ids(&["a", "b"]), &ids(&["b", "a"])).is_empty());
+        assert!(changed_favorite_ids(&ids(&[]), &ids(&[])).is_empty());
+    }
+
     #[test]
     fn test_repository_manager_wrapper_structure() {
         let manager = RepositoryManager::new();

src-tauri/src/commands/storage/people.rs

@@ -47,10 +47,25 @@ pub async fn storage_save_person(
     };
 
     let query = Query::with_params(
-        "INSERT OR REPLACE INTO people (
+        // A real UPSERT, not INSERT OR REPLACE — `people` is now backed by the
+        // `people_fts` index (migration 022), and REPLACE would orphan an index
+        // entry on every re-cache: it fires no AFTER DELETE trigger without
+        // `recursive_triggers`, and reassigns the rowid that `content_rowid`
+        // refers to. Same defect as DR-110 fixed for `items`.
+        //
+        // TRACES: UR-065 | DR-110, DR-111
+        "INSERT INTO people (
             id, server_id, name, overview, primary_image_tag,
             premiere_date, end_date, synced_at
-        ) VALUES (?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)",
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
+        ON CONFLICT(id) DO UPDATE SET
+            server_id = excluded.server_id,
+            name = excluded.name,
+            overview = excluded.overview,
+            primary_image_tag = excluded.primary_image_tag,
+            premiere_date = excluded.premiere_date,
+            end_date = excluded.end_date,
+            synced_at = CURRENT_TIMESTAMP",
         vec![
             QueryParam::String(person.id),
             QueryParam::String(person.server_id),

src-tauri/src/download/cache.rs

@@ -26,6 +26,12 @@ pub struct CacheConfig {
     pub storage_limit: u64,
     /// Only cache on WiFi
     pub wifi_only: bool,
+    /// How long a temporary (`download_source = 'auto'`) download lives before
+    /// it is reclaimed, in hours. 0 disables expiry, leaving space pressure as
+    /// the only reclaim trigger.
+    ///
+    /// TRACES: UR-071 | DR-127
+    pub temporary_ttl_hours: u64,
 }
 
 impl Default for CacheConfig {
@@ -37,6 +43,10 @@ impl Default for CacheConfig {
             album_affinity_threshold: 3,
             storage_limit: 10 * 1024 * 1024 * 1024, // 10GB
             wifi_only: false,                       // Allow preloading on any connection by default
+            // A week: long enough that re-watching over a weekend still hits
+            // disk, short enough that a one-off play does not hold space
+            // indefinitely.
+            temporary_ttl_hours: 24 * 7,
         }
     }
 }
@@ -183,7 +193,92 @@ impl SmartCache {
         current_size + new_size <= storage_limit
     }
 
-    /// Evict least recently used items to make space (async version)
+    /// Reclaim temporary downloads whose life limit has passed.
+    ///
+    /// The time-based half of the temporary tier (DR-127); [`evict_lru_async`]
+    /// is the space-pressure half. A row is reclaimed by whichever fires first.
+    ///
+    /// Scoped to `download_source = 'auto'` for the same reason eviction is: a
+    /// `'user'` row is someone's own download and has no expiry. `COALESCE`
+    /// guards rows predating migration 012, whose source is NULL and whose
+    /// provenance must therefore be treated as the user's.
+    ///
+    /// Expiry is normally *derived* — `completed_at` plus the configured TTL —
+    /// rather than stamped at completion. That means a TTL change applies to
+    /// entries already on disk instead of only to future ones, and entries
+    /// predating the column expire without a backfill. `expires_at` is honoured
+    /// as a per-row override when something sets it.
+    ///
+    /// `now` is passed in rather than read from the clock so the policy is
+    /// testable without sleeping. Both sides go through SQLite's `datetime()`
+    /// because `completed_at` is written as `CURRENT_TIMESTAMP`
+    /// (`YYYY-MM-DD HH:MM:SS`) while callers pass RFC-3339 (`…T…+00:00`) — a raw
+    /// string comparison between the two formats is wrong, since `' ' < 'T'`.
+    ///
+    /// Returns the number of entries reclaimed.
+    ///
+    /// TRACES: UR-071 | DR-127 | UT-120
+    pub async fn reclaim_expired_async<S: DatabaseService>(
+        &self,
+        db_service: &Arc<S>,
+        user_id: &str,
+        now: &str,
+    ) -> Result<usize, String> {
+        let ttl_hours = {
+            let config = self.config.lock().map_err(|e| e.to_string())?;
+            config.temporary_ttl_hours
+        };
+        // 0 disables time-based reclaim; space pressure remains the only trigger.
+        if ttl_hours == 0 {
+            return Ok(0);
+        }
+
+        let expired: Vec<(i64, String)> = db_service
+            .query_many(
+                Query::with_params(
+                    "SELECT id, file_path FROM downloads
+                      WHERE user_id = ?
+                        AND COALESCE(download_source, 'user') = 'auto'
+                        AND status = 'completed'
+                        AND datetime(
+                              COALESCE(expires_at, datetime(completed_at, '+' || ? || ' hours'))
+                            ) < datetime(?)",
+                    vec![
+                        QueryParam::String(user_id.to_string()),
+                        QueryParam::String(ttl_hours.to_string()),
+                        QueryParam::String(now.to_string()),
+                    ],
+                ),
+                |row| Ok((row.get(0)?, row.get(1)?)),
+            )
+            .await
+            .map_err(|e| e.to_string())?;
+
+        let mut reclaimed = 0usize;
+        for (id, file_path) in expired {
+            // Best-effort on the file: a missing one still needs its row gone,
+            // or the sweep retries it forever.
+            let _ = std::fs::remove_file(&file_path);
+            db_service
+                .execute(Query::with_params(
+                    "DELETE FROM downloads WHERE id = ?",
+                    vec![QueryParam::Int64(id)],
+                ))
+                .await
+                .map_err(|e| e.to_string())?;
+            reclaimed += 1;
+        }
+
+        if reclaimed > 0 {
+            info!("[SmartCache] Reclaimed {} expired entries", reclaimed);
+        }
+        Ok(reclaimed)
+    }
+
+    /// Evict least recently used items to make space (async version).
+    ///
+    /// The space-pressure half of the temporary tier; [`reclaim_expired_async`]
+    /// is the time-based half.
     pub async fn evict_lru_async<S: DatabaseService>(
         &self,
         db_service: &Arc<S>,
@@ -207,10 +302,26 @@ impl SmartCache {
         let to_free = (current_size + space_needed) - limit;
         let mut freed: u64 = 0;
 
-        // Get downloads ordered by last access (oldest first)
+        // Only the *temporary* tier is evictable. `download_source = 'auto'` is
+        // precache — the cache put it there, the cache may reclaim it. A 'user'
+        // row is a download someone explicitly asked for; deleting it to make
+        // room for a predictive fetch is data loss, and because the old query
+        // ordered purely by `completed_at ASC` it took the oldest — typically
+        // exactly the film saved for a flight.
+        //
+        // COALESCE, not `= 'auto'` alone: migration 012 added the column with a
+        // 'user' default, but rows predating it can be NULL, and an unknown
+        // provenance must be treated as the user's, never as disposable.
+        //
+        // Freeing less than requested is the correct outcome when only user
+        // downloads remain — the caller surfaces "unable to free enough space"
+        // rather than silently deleting them.
+        //
+        // TRACES: UR-071 | DR-126 | UT-108
         let query = Query::with_params(
             "SELECT id, file_size, file_path FROM downloads
              WHERE user_id = ? AND status = 'completed'
+               AND COALESCE(download_source, 'user') = 'auto'
              ORDER BY completed_at ASC",
             vec![QueryParam::String(user_id.to_string())],
         );
@@ -304,6 +415,205 @@ mod tests {
         assert!(cache.should_precache_queue());
     }
 
+    /// Expiry reclaims only temporary entries that are actually past their life
+    /// limit — never a user's download (which has no expiry), and never a
+    /// temporary entry still within its life.
+    ///
+    /// TRACES: UR-071 | DR-127 | UT-120
+    #[tokio::test]
+    async fn test_reclaim_expired_only_takes_expired_temporary_entries() {
+        use crate::storage::db_service::{DatabaseService, RusqliteService};
+        use rusqlite::Connection;
+        use std::sync::{Arc, Mutex};
+
+        let conn = Connection::open_in_memory().unwrap();
+        conn.execute(
+            "CREATE TABLE downloads (
+                id INTEGER PRIMARY KEY,
+                user_id TEXT,
+                status TEXT,
+                file_size INTEGER,
+                file_path TEXT,
+                completed_at TEXT,
+                download_source TEXT DEFAULT 'user',
+                expires_at TEXT
+            )",
+            [],
+        )
+        .unwrap();
+
+        // `completed_at` is in SQLite's CURRENT_TIMESTAMP format (space, not
+        // 'T'), deliberately: the query has to compare it against an RFC-3339
+        // "now" and must not do so as raw strings.
+        for (path, source, completed, expires) in [
+            // Completed long ago, no override => derived expiry has passed.
+            ("/tmp/jt-expired.mp4", "auto", "2026-01-01 00:00:00", None),
+            // Completed yesterday => still inside the 7-day default TTL.
+            ("/tmp/jt-fresh.mp4", "auto", "2026-05-31 00:00:00", None),
+            // Old, but an explicit override keeps it alive.
+            (
+                "/tmp/jt-override.mp4",
+                "auto",
+                "2026-01-01 00:00:00",
+                Some("2026-12-01T00:00:00+00:00"),
+            ),
+            // A user download must never carry an expiry, but assert the sweep
+            // ignores it even if one were somehow set.
+            (
+                "/tmp/jt-user.mp4",
+                "user",
+                "2026-01-01 00:00:00",
+                Some("2026-01-01T00:00:00+00:00"),
+            ),
+            (
+                "/tmp/jt-user-noexp.mp4",
+                "user",
+                "2026-01-01 00:00:00",
+                None,
+            ),
+        ] {
+            conn.execute(
+                "INSERT INTO downloads (user_id, status, file_size, file_path, download_source, completed_at, expires_at)
+                 VALUES ('user1', 'completed', 10, ?1, ?2, ?3, ?4)",
+                rusqlite::params![path, source, completed, expires],
+            )
+            .unwrap();
+        }
+
+        let conn_arc = Arc::new(Mutex::new(conn));
+        let db_service = Arc::new(RusqliteService::new(conn_arc.clone()));
+        let cache = SmartCache::new(CacheConfig::default());
+
+        let reclaimed = cache
+            .reclaim_expired_async(&db_service, "user1", "2026-06-01T00:00:00+00:00")
+            .await
+            .unwrap();
+        assert_eq!(
+            reclaimed, 1,
+            "only the expired temporary entry is reclaimed"
+        );
+
+        let surviving: Vec<String> = {
+            let guard = conn_arc.lock_safe();
+            let mut stmt = guard
+                .prepare("SELECT file_path FROM downloads ORDER BY id")
+                .unwrap();
+            let rows = stmt
+                .query_map([], |row| row.get::<_, String>(0))
+                .unwrap()
+                .map(|r| r.unwrap())
+                .collect();
+            rows
+        };
+        assert_eq!(
+            surviving,
+            vec![
+                "/tmp/jt-fresh.mp4".to_string(),
+                "/tmp/jt-override.mp4".to_string(),
+                "/tmp/jt-user.mp4".to_string(),
+                "/tmp/jt-user-noexp.mp4".to_string(),
+            ],
+            "entries within their life, those with a later override, and every user download must survive"
+        );
+
+        // TTL of 0 disables time-based reclaim entirely.
+        let cache_no_ttl = SmartCache::new(CacheConfig {
+            temporary_ttl_hours: 0,
+            ..Default::default()
+        });
+        assert_eq!(
+            cache_no_ttl
+                .reclaim_expired_async(&db_service, "user1", "2027-01-01T00:00:00+00:00")
+                .await
+                .unwrap(),
+            0,
+            "a zero TTL leaves space pressure as the only reclaim trigger"
+        );
+    }
+
+    /// Eviction must only reclaim *temporary* (`download_source = 'auto'`)
+    /// downloads — the precache tier. A download the user explicitly asked for
+    /// is their file: it may be deleted by them, never by the cache making room
+    /// for a predictive fetch.
+    ///
+    /// Before the fix, `evict_lru_async` selected every completed row ordered by
+    /// `completed_at ASC` with no source filter, so hitting the storage limit
+    /// deleted the *oldest* download — typically the film someone downloaded for
+    /// a flight — in favour of a newer auto-precached track.
+    ///
+    /// TRACES: UR-071 | DR-126 | UT-108
+    #[tokio::test]
+    async fn test_evict_lru_never_deletes_user_downloads() {
+        use crate::storage::db_service::RusqliteService;
+        use rusqlite::Connection;
+        use std::sync::{Arc, Mutex};
+
+        let conn = Connection::open_in_memory().unwrap();
+        conn.execute(
+            "CREATE TABLE downloads (
+                id INTEGER PRIMARY KEY,
+                user_id TEXT,
+                status TEXT,
+                file_size INTEGER,
+                file_path TEXT,
+                completed_at TEXT,
+                download_source TEXT DEFAULT 'user'
+            )",
+            [],
+        )
+        .unwrap();
+
+        // The user's own download is the OLDEST, so a purely time-ordered
+        // eviction would take it first.
+        conn.execute(
+            "INSERT INTO downloads (user_id, status, file_size, file_path, completed_at, download_source)
+             VALUES ('user1', 'completed', 600, '/tmp/jellytau-test-user.mp4', '2026-01-01', 'user')",
+            [],
+        )
+        .unwrap();
+        // A newer, auto-precached item.
+        conn.execute(
+            "INSERT INTO downloads (user_id, status, file_size, file_path, completed_at, download_source)
+             VALUES ('user1', 'completed', 600, '/tmp/jellytau-test-auto.mp4', '2026-06-01', 'auto')",
+            [],
+        )
+        .unwrap();
+
+        let conn_arc = Arc::new(Mutex::new(conn));
+        let db_service = Arc::new(RusqliteService::new(conn_arc.clone()));
+
+        let cache = SmartCache::new(CacheConfig {
+            storage_limit: 1000,
+            ..Default::default()
+        });
+
+        // 1200 bytes held against a 1000 limit: eviction must free something.
+        let freed = cache
+            .evict_lru_async(&db_service, "user1", 0)
+            .await
+            .unwrap();
+        assert!(freed > 0, "eviction should have reclaimed the auto entry");
+
+        let surviving: Vec<String> = {
+            let guard = conn_arc.lock_safe();
+            let mut stmt = guard
+                .prepare("SELECT download_source FROM downloads ORDER BY id")
+                .unwrap();
+            let rows = stmt
+                .query_map([], |row| row.get::<_, String>(0))
+                .unwrap()
+                .map(|r| r.unwrap())
+                .collect();
+            rows
+        };
+
+        assert_eq!(
+            surviving,
+            vec!["user".to_string()],
+            "the user's own download must survive; only the 'auto' entry is evictable"
+        );
+    }
+
     #[tokio::test]
     async fn test_storage_limit_check() {
         use crate::storage::db_service::RusqliteService;

src-tauri/src/lib.rs

@@ -128,6 +128,8 @@ use commands::{
     player_get_sleep_timer,
     player_get_status,
     player_get_video_settings,
+    // Preload commands
+    player_local_media_path,
     player_move_in_queue,
     player_next,
     player_on_playback_ended,
@@ -138,7 +140,6 @@ use commands::{
     player_play_next_episode,
     player_play_queue,
     player_play_tracks,
-    // Preload commands
     player_preload_upcoming,
     player_previous,
     player_remove_from_queue,
@@ -187,6 +188,7 @@ use commands::{
     repository_get_download_disk_usage,
     repository_get_downloaded_items,
     repository_get_downloaded_libraries,
+    repository_get_favorites,
     repository_get_genres,
     repository_get_image_url,
     repository_get_item,
@@ -700,6 +702,7 @@ fn specta_builder() -> Builder<tauri::Wry> {
             player_report_position,
             player_report_media_loaded,
             // Preload commands
+            player_local_media_path,
             player_preload_upcoming,
             player_set_cache_config,
             player_get_cache_config,
@@ -893,6 +896,7 @@ fn specta_builder() -> Builder<tauri::Wry> {
             repository_get_image_url,
             repository_mark_favorite,
             repository_unmark_favorite,
+            repository_get_favorites,
             repository_get_person,
             repository_get_items_by_person,
             repository_get_similar_items,
@@ -1288,6 +1292,21 @@ pub fn run() {
             let playback_reporter_wrapper = PlaybackReporterWrapper(playback_reporter.clone());
             app.manage(playback_reporter_wrapper);
 
+            // Keep the local search index fresh. Ownership of *when* to re-index
+            // sits here rather than in the frontend: it is sync policy over
+            // domain data, and a startup-only trigger left a long session
+            // searching a stale catalog.
+            // TRACES: UR-065 | DR-109, IR-030
+            info!("[INIT] Starting background catalog indexer...");
+            commands::catalog::spawn_catalog_indexer(app.handle().clone());
+
+            // Push favourite toggles made while the server was unreachable, on
+            // every reconnect. In Rust rather than the frontend so it runs
+            // whether or not the screen that made the change is still mounted.
+            // TRACES: UR-069 | DR-120
+            info!("[INIT] Starting favourites drain...");
+            commands::favorites::spawn_favorites_drain(app.handle().clone());
+
             info!("[INIT] Application setup completed successfully");
             Ok(())
         })

src-tauri/src/player/android/mod.rs

@@ -930,6 +930,25 @@ pub extern "system" fn Java_com_dtourolle_jellytau_player_JellyTauPlayer_nativeO
                             .await;
                     }
                 }
+                Ok(AutoplayDecision::ResumeStream { position }) => {
+                    // ExoPlayer reported ENDED because the progressive transcode's
+                    // connection dropped, not because the episode finished. This
+                    // is the arm that matters while backgrounded: it needs no
+                    // frontend echo, so the stream re-opens even with the webview
+                    // suspended — and playback never parks in STATE_ENDED, where
+                    // the next lockscreen/Bluetooth play restarts the item at 0:00.
+                    log::info!(
+                        "[Autoplay] Decision: Resume truncated stream at {:.1}s",
+                        position
+                    );
+                    let ctrl = controller.lock().await;
+                    if let Err(e) = ctrl.resume_stream_at(position).await {
+                        log::error!("[Autoplay] Failed to resume truncated stream: {}", e);
+                        if let Some(emitter) = EVENT_EMITTER.get() {
+                            emitter.emit(PlayerStatusEvent::PlaybackEnded);
+                        }
+                    }
+                }
                 Err(e) => {
                     log::error!("[Autoplay] Decision failed: {}", e);
                     // Emit PlaybackEnded event on error
@@ -974,11 +993,58 @@ pub extern "system" fn Java_com_dtourolle_jellytau_player_JellyTauPlayer_nativeO
         .get_string(&message)
         .map(|s| s.into())
         .unwrap_or_else(|_| "Unknown error".to_string());
+    let recoverable = recoverable != 0;
+
+    // A background audio-only handoff is an mp3 the device was already decoding,
+    // so a recoverable failure part-way through is the network. Surfacing it as a
+    // player error stops playback for good (the frontend's handler calls
+    // player_stop); re-opening the stream where it died is the "buffer and
+    // resume" this actually is. Everything else keeps reporting the error.
+    if recoverable {
+        if let Some(controller) = PLAYER_CONTROLLER.get() {
+            let controller = controller.clone();
+            let message_str = message_str.clone();
+            tauri::async_runtime::spawn(async move {
+                let resume = controller.lock().await.recoverable_error_resume();
+                let Some((position, delay_secs)) = resume else {
+                    if let Some(emitter) = EVENT_EMITTER.get() {
+                        emitter.emit(PlayerStatusEvent::Error {
+                            message: message_str,
+                            recoverable: true,
+                        });
+                    }
+                    return;
+                };
+
+                log::warn!(
+                    "[ExoPlayer] Recoverable stream error ({}) — re-opening at {:.1}s in {}s",
+                    message_str,
+                    position,
+                    delay_secs
+                );
+                // Give a brief outage time to clear before asking the server for
+                // the stream again; retrying instantly just burns the budget.
+                tokio::time::sleep(std::time::Duration::from_secs(delay_secs)).await;
+
+                let ctrl = controller.lock().await;
+                if let Err(e) = ctrl.resume_stream_at(position).await {
+                    log::error!("[ExoPlayer] Failed to resume after error: {}", e);
+                    if let Some(emitter) = EVENT_EMITTER.get() {
+                        emitter.emit(PlayerStatusEvent::Error {
+                            message: message_str,
+                            recoverable: true,
+                        });
+                    }
+                }
+            });
+            return;
+        }
+    }
 
     if let Some(emitter) = EVENT_EMITTER.get() {
         emitter.emit(PlayerStatusEvent::Error {
             message: message_str,
-            recoverable: recoverable != 0,
+            recoverable,
         });
     }
 }

src-tauri/src/player/autoplay.rs

@@ -11,6 +11,10 @@ pub enum AutoplayDecision {
     Stop,
     /// Advance to next track in queue (for audio/movies)
     AdvanceToNext,
+    /// The stream ended well short of the item's runtime — the connection
+    /// dropped, not the media. Re-open the same stream at `position` instead of
+    /// running any end-of-item logic (UR-040).
+    ResumeStream { position: f64 },
     /// Show next episode popup with countdown
     ShowNextEpisodePopup {
         current_episode: MediaItem,

src-tauri/src/player/mod.rs

@@ -11,6 +11,7 @@ pub mod seek;
 pub mod session;
 pub mod sleep_timer;
 pub mod state;
+pub mod stream_end;
 
 #[cfg(test)]
 mod mpv_backend_test;
@@ -54,6 +55,16 @@ pub use android::{
     set_remote_volume_handler, MediaCommandHandler, RemoteVolumeHandler,
 };
 
+/// Seconds added per attempt before retrying a stream that failed with an error.
+///
+/// Attempt 1 waits this long, attempt 2 twice as long, and so on — a spread that
+/// covers roughly a quarter-minute of outage across the retry budget without
+/// leaving the user staring at a dead notification when the network is truly gone.
+/// Only *read* by the Android error callback (`#[cfg(android)]`), but compiled
+/// and unit-tested on the host, hence `allow(dead_code)` off-Android.
+#[cfg_attr(not(target_os = "android"), allow(dead_code))]
+const RESUME_BACKOFF_STEP_SECS: u64 = 2;
+
 /// Metadata for the lockscreen / media notification.
 ///
 /// Used to drive the Android MediaSession from Rust in remote (cast) mode, where
@@ -168,6 +179,15 @@ pub struct PlayerController {
     // TRACES: UR-040 | DR-052
     background_audio_base: Arc<Mutex<f64>>,
 
+    // Budget for re-opening a stream that ended short of the item's runtime.
+    //
+    // A resume re-requests the same URL, so a server that is genuinely gone would
+    // otherwise end → resume → end without limit. The tracker only bounds retries
+    // that make no progress; a resume that plays on refills it.
+    //
+    // TRACES: UR-040 | DR-129
+    stream_resume: Arc<Mutex<stream_end::ResumeTracker>>,
+
     // Last state reported by a webview-rendered HTML5 <video>/<audio> element.
     //
     // Webview-rendered media is played by an element the native backend cannot
@@ -201,6 +221,7 @@ impl PlayerController {
             end_reason: Arc::new(Mutex::new(None)),
             autoplay_episode_count: Arc::new(Mutex::new(0)),
             background_audio_base: Arc::new(Mutex::new(0.0)),
+            stream_resume: Arc::new(Mutex::new(stream_end::ResumeTracker::default())),
             html5_playing: Arc::new(Mutex::new(None)),
         };
 
@@ -271,6 +292,16 @@ impl PlayerController {
         self.end_reason.lock_safe().take()
     }
 
+    /// Read the end reason WITHOUT consuming it.
+    ///
+    /// `take_end_reason` has an owner: on Android the JNI ended-callback consumes
+    /// the `NewTrackLoaded` every load sets, and the frontend's echoed call is the
+    /// one that sees `None` and decides. The truncated-stream check runs in both
+    /// calls and must not disturb that hand-off, so it peeks.
+    fn peek_end_reason(&self) -> Option<EndReason> {
+        *self.end_reason.lock_safe()
+    }
+
     /// Record that playback is being stopped by an expiring sleep timer.
     ///
     /// Stopping the backend makes it fire its ended callback (ExoPlayer does on
@@ -1034,6 +1065,15 @@ impl PlayerController {
     /// Only triggers autoplay if the track finished naturally (EndReason::Finished or None).
     /// If EndReason is NewTrackLoaded, UserStop, UserSkip, or Error, returns Stop without autoplay.
     pub async fn on_playback_ended(&self) -> Result<AutoplayDecision, String> {
+        // A truncated stream is not an end at all, so this is decided BEFORE the
+        // end-reason gate below — which returns early for the `NewTrackLoaded`
+        // that every load sets, and would therefore swallow the whole question on
+        // Android's JNI callback: the one call guaranteed to run while the app is
+        // backgrounded and the webview cannot echo anything back.
+        if let Some(position) = self.truncated_stream_resume_position() {
+            return Ok(AutoplayDecision::ResumeStream { position });
+        }
+
         // Check why playback ended
         let end_reason = self.take_end_reason();
 
@@ -1248,6 +1288,194 @@ impl PlayerController {
         self.start_autoplay_countdown(next_episode, countdown_seconds);
     }
 
+    /// A video item played through the native *audio* path — i.e. the background
+    /// audio-only handoff, the only place a length-less progressive transcode is
+    /// used. Jellyfin's item-type taxonomy stays in Rust (CLAUDE.md).
+    fn is_audio_only_video(item: &MediaItem) -> bool {
+        item.media_type == MediaType::Audio
+            && matches!(item.item_type.as_deref(), Some("Episode") | Some("Movie"))
+    }
+
+    /// Claim a resume attempt for the current stream, returning the absolute
+    /// position to re-open at and the 1-based attempt number. `None` when the
+    /// current item cannot meaningfully be re-requested, or when retrying at this
+    /// position has stopped helping.
+    ///
+    /// Only `Remote` sources qualify. A downloaded file cannot fail because of
+    /// the network, so re-opening one would paper over a real read error; a
+    /// `DirectUrl` is a plugin's endpoint with no Jellyfin item behind it.
+    ///
+    /// The player's position is relative to the stream's own zero (the handoff
+    /// URL's `StartTimeTicks`), so the base is added back to get an absolute one.
+    /// It is zero for everything else, where positions are already absolute.
+    ///
+    /// TRACES: UR-040, UR-004 | DR-129 | UT-117
+    fn claim_stream_resume(&self) -> Option<(f64, u32)> {
+        let current = {
+            let queue = self.queue.lock_safe();
+            queue.current().cloned()
+        }?;
+        if !matches!(current.source, MediaSource::Remote { .. }) {
+            return None;
+        }
+
+        let base = *self.background_audio_base.lock_safe();
+        let absolute = (base + self.position()).max(0.0);
+
+        match self.stream_resume.lock_safe().allow_attempt(absolute) {
+            Some(attempt) => Some((absolute, attempt)),
+            None => {
+                warn!(
+                    "[PlayerController] Stream for {} keeps failing at {:.1}s — giving up on resuming",
+                    current.id, absolute
+                );
+                None
+            }
+        }
+    }
+
+    /// The absolute position to re-open the current stream at, when the reported
+    /// end was really a dropped connection — `None` when the end looks genuine,
+    /// when this is not an audio-only handoff, or when retrying has stopped
+    /// helping.
+    ///
+    /// TRACES: UR-040 | DR-129 | UT-117
+    fn truncated_stream_resume_position(&self) -> Option<f64> {
+        // An explicit user intent already explains the end; never resume over it.
+        if matches!(
+            self.peek_end_reason(),
+            Some(EndReason::UserStop) | Some(EndReason::UserSkip) | Some(EndReason::Error)
+        ) {
+            return None;
+        }
+
+        // One lock at a time — `position()` reaches into the backend, and nesting
+        // that inside the queue lock would invent a lock order nothing else here
+        // takes.
+        let item_duration = {
+            let queue = self.queue.lock_safe();
+            let current = queue.current()?;
+            if !Self::is_audio_only_video(current) {
+                return None;
+            }
+            current.duration
+        };
+        let base = *self.background_audio_base.lock_safe();
+        let absolute = (base + self.position()).max(0.0);
+
+        // Only spend a resume attempt once the runtime says this really was cut
+        // short — a genuine end must stay a genuine end.
+        if !stream_end::is_truncated_end(
+            absolute,
+            item_duration,
+            stream_end::TRUNCATED_STREAM_TOLERANCE_SECS,
+        ) {
+            return None;
+        }
+
+        self.claim_stream_resume().map(|(position, _)| position)
+    }
+
+    /// Where to re-open the current stream after a *recoverable* playback error,
+    /// plus how many seconds to wait first.
+    ///
+    /// The media was decoding fine a moment ago, so a mid-playback failure on a
+    /// server stream is the network — and stopping the player (the previous
+    /// behaviour, via the frontend's error handler) turns a hiccup into "playback
+    /// just died". Applies to every streamed item, not only the audio-only
+    /// handoff: music and video reach here instead of the truncation path because
+    /// their streams declare a length, so a cut connection surfaces as an error
+    /// rather than a phantom end.
+    ///
+    /// The wait grows with the attempt number so a short outage has time to
+    /// clear, and the shared budget stops the retries when it doesn't.
+    ///
+    /// Only *called* from the Android error callback (`#[cfg(android)]`), but
+    /// compiled and unit-tested on the host, hence `allow(dead_code)` off-Android.
+    ///
+    /// TRACES: UR-040, UR-004 | DR-129 | UT-117
+    #[cfg_attr(not(target_os = "android"), allow(dead_code))]
+    pub fn recoverable_error_resume(&self) -> Option<(f64, u64)> {
+        self.claim_stream_resume()
+            .map(|(position, attempt)| (position, attempt as u64 * RESUME_BACKOFF_STEP_SECS))
+    }
+
+    /// Re-open the current stream at `position` after the network cut it short.
+    ///
+    /// Single place every dispatcher agrees on, for the same reason
+    /// `auto_advance_to_next_episode` is: the Android JNI callbacks and the
+    /// frontend-invoked command must not disagree about what a failed stream
+    /// means. None of them emits `PlaybackEnded` for this, so nothing downstream
+    /// clears the queue or tears the session down — from the outside this is a
+    /// buffering hiccup, which is what it actually was.
+    ///
+    /// Reloads the item **in place** rather than through `play_item`, which
+    /// replaces the queue with a single item: recovering a track that way would
+    /// throw away the rest of the album, turning a network blip into lost state.
+    ///
+    /// Two shapes of stream, two ways back to `position`:
+    ///
+    /// - The audio-only handoff's `/Audio/{id}/universal` transcode is chunked
+    ///   with no length, so it cannot be seeked. Its URL is rewritten to start at
+    ///   the position instead — edited, not rebuilt from the repository, since it
+    ///   already carries the user's audio track and media source and recovering
+    ///   from a network failure must not itself need a network round-trip.
+    /// - Everything else (a static file with byte ranges, an HLS playlist)
+    ///   declares its whole timeline, so re-preparing the URL it already has and
+    ///   seeking lands in the right place — and leaves any transcode session
+    ///   behind it alone.
+    ///
+    /// TRACES: UR-040, UR-004 | DR-129 | UT-117
+    pub async fn resume_stream_at(&self, position: f64) -> Result<(), String> {
+        let current = {
+            let queue = self.queue.lock_safe();
+            queue.current().cloned()
+        }
+        .ok_or_else(|| "No current item to resume".to_string())?;
+
+        let MediaSource::Remote { stream_url, .. } = &current.source else {
+            return Err(format!(
+                "Cannot resume a non-remote source for {}",
+                current.id
+            ));
+        };
+
+        info!(
+            "[PlayerController] Stream for {} failed — re-opening at {:.1}s",
+            current.id, position
+        );
+
+        if !Self::is_audio_only_video(&current) {
+            self.load_and_play(&current).map_err(|e| e.to_string())?;
+            if position > 0.5 {
+                self.seek(position).map_err(|e| e.to_string())?;
+            }
+            return Ok(());
+        }
+
+        let restarted_url = stream_end::with_start_time(stream_url, position);
+        {
+            let queue_arc = self.queue.clone();
+            let mut queue = queue_arc.lock_safe();
+            if !queue.update_current_stream_url(restarted_url) {
+                return Err(format!("Failed to update stream URL for {}", current.id));
+            }
+        }
+        let resumed = {
+            let queue = self.queue.lock_safe();
+            queue.current().cloned()
+        }
+        .ok_or_else(|| "Current item vanished mid-resume".to_string())?;
+
+        // The re-opened stream's timeline starts at `position` (StartTimeTicks),
+        // so that is its zero: the exit-to-foreground maths and the lockscreen
+        // scrubber both read absolute positions off this base.
+        self.set_background_audio_base(position);
+        let _ = set_lockscreen_position_offset(position.max(0.0));
+
+        self.load_and_play(&resumed).map_err(|e| e.to_string())
+    }
+
     /// Advance to the next episode while playing audio-only in the background.
     ///
     /// The normal autoplay-next path navigates the frontend to `/player/<id>`,
@@ -1322,6 +1550,9 @@ impl PlayerController {
         // back to the foreground) and the lockscreen scrubber's matching shift.
         self.set_background_audio_base(0.0);
         let _ = set_lockscreen_position_offset(0.0);
+        // Different stream entirely: whatever was stuck about the last one is not
+        // this one's problem.
+        self.stream_resume.lock_safe().reset();
 
         self.play_item(media_item).map_err(|e| e.to_string())
     }
@@ -2862,6 +3093,13 @@ mod tests {
         async fn unmark_favorite(&self, _: &str) -> Result<(), repo_types::RepoError> {
             unimplemented!()
         }
+        async fn get_favorites(
+            &self,
+            _: repo_types::SearchScope,
+            _: Option<repo_types::GetItemsOptions>,
+        ) -> Result<repo_types::SearchResult, repo_types::RepoError> {
+            unimplemented!()
+        }
         async fn clear_watch_history(&self, _: &str) -> Result<(), repo_types::RepoError> {
             unimplemented!()
         }
@@ -3018,6 +3256,7 @@ mod tests {
             item_type: Some("Episode".to_string()),
             media_type: MediaType::Audio, // audio-only handoff, not Video
             series_id: Some("series1".to_string()),
+            duration: Some(180.0),
             source: MediaSource::Remote {
                 stream_url: "http://example.com/ep2-audio.m3u8".to_string(),
                 jellyfin_item_id: "ep2".to_string(),
@@ -3026,6 +3265,8 @@ mod tests {
         };
         controller.play_queue(vec![episode], 0).unwrap();
 
+        // Played through to the end — a natural finish, not a stream cut short.
+        controller.seek(180.0).unwrap();
         // Clear the NewTrackLoaded reason to simulate natural track end.
         controller.take_end_reason();
 
@@ -3158,6 +3399,328 @@ mod tests {
         assert!(controller.current_is_audio_episode());
     }
 
+    /// Build the audio-only episode the background handoff loads: a video item
+    /// played through the native audio path, with a known runtime and a stream
+    /// URL carrying the handoff position.
+    fn audio_only_episode(runtime_seconds: f64) -> MediaItem {
+        MediaItem {
+            id: "ep2".to_string(),
+            item_type: Some("Episode".to_string()),
+            media_type: MediaType::Audio,
+            series_id: Some("series1".to_string()),
+            duration: Some(runtime_seconds),
+            source: MediaSource::Remote {
+                stream_url:
+                    "http://s/Audio/ep2/universal?api_key=k&AudioStreamIndex=2&StartTimeTicks=0"
+                        .to_string(),
+                jellyfin_item_id: "ep2".to_string(),
+            },
+            ..create_test_items(1).remove(0)
+        }
+    }
+
+    /// A flaky connection truncates the progressive mp3 transcode that carries
+    /// background audio-only playback. ExoPlayer sees end-of-input on a stream
+    /// with no reliable length, so it reports STATE_ENDED ten minutes into a
+    /// twenty-five minute episode — indistinguishable, to the player, from the
+    /// real end.
+    ///
+    /// Treating that as "the episode finished" is what the user experiences as
+    /// the episode randomly restarting: playback parks in STATE_ENDED and the
+    /// next play intent (lockscreen, notification, Bluetooth reconnect) seeks an
+    /// ended player to position 0 before playing. The runtime we already know
+    /// says the stream died early, so the decision must be to resume it.
+    #[tokio::test]
+    async fn test_truncated_background_audio_stream_resumes_instead_of_ending() {
+        let controller = PlayerController::default();
+        controller.set_repository(Arc::new(MockEpisodeRepo::season(3)));
+
+        controller
+            .play_queue(vec![audio_only_episode(1500.0)], 0)
+            .unwrap();
+        // The connection dropped 10 minutes into a 25-minute episode.
+        controller.seek(600.0).unwrap();
+        controller.take_end_reason();
+
+        let decision = controller.on_playback_ended().await.unwrap();
+
+        match decision {
+            AutoplayDecision::ResumeStream { position } => {
+                assert_eq!(position, 600.0, "must resume where the stream died");
+            }
+            other => panic!(
+                "a stream that ended 15 minutes short of the runtime must resume, \
+                 not run end-of-episode logic; got {:?}",
+                other
+            ),
+        }
+    }
+
+    /// The handoff stream's timeline starts at the handoff position, so the
+    /// player reports a *relative* position. The runtime it is compared against
+    /// is absolute — the base has to be added back, or every handoff looks like a
+    /// truncation.
+    #[tokio::test]
+    async fn test_truncated_check_uses_the_absolute_position() {
+        let controller = PlayerController::default();
+        controller.set_repository(Arc::new(MockEpisodeRepo::season(3)));
+
+        controller
+            .play_queue(vec![audio_only_episode(1500.0)], 0)
+            .unwrap();
+        // Handed off at 24:00; the stream then played its last 56 seconds out.
+        controller.set_background_audio_base(1440.0);
+        controller.seek(56.0).unwrap();
+        controller.take_end_reason();
+
+        let decision = controller.on_playback_ended().await.unwrap();
+
+        assert!(
+            matches!(decision, AutoplayDecision::ShowNextEpisodePopup { .. }),
+            "24:56 of a 25:00 episode is the real end, not a truncation; got {:?}",
+            decision
+        );
+    }
+
+    /// The resume re-opens the same URL, so a server that is actually gone would
+    /// otherwise end → resume → end forever. After the budget runs out the
+    /// decision falls back to normal end-of-item handling.
+    #[tokio::test]
+    async fn test_repeated_truncation_at_the_same_position_gives_up() {
+        let controller = PlayerController::default();
+        controller.set_repository(Arc::new(MockEpisodeRepo::season(3)));
+
+        controller
+            .play_queue(vec![audio_only_episode(1500.0)], 0)
+            .unwrap();
+        controller.seek(600.0).unwrap();
+
+        for attempt in 1..=stream_end::MAX_STALLED_RESUME_ATTEMPTS {
+            controller.take_end_reason();
+            let decision = controller.on_playback_ended().await.unwrap();
+            assert!(
+                matches!(decision, AutoplayDecision::ResumeStream { .. }),
+                "attempt {} should still resume, got {:?}",
+                attempt,
+                decision
+            );
+        }
+
+        controller.take_end_reason();
+        let decision = controller.on_playback_ended().await.unwrap();
+        assert!(
+            !matches!(decision, AutoplayDecision::ResumeStream { .. }),
+            "a stream stuck at the same position must stop retrying, got {:?}",
+            decision
+        );
+    }
+
+    /// Ordinary music is not covered: its streams are not the length-less
+    /// progressive transcode this guards, and a short track legitimately ends
+    /// well before a stale duration would suggest.
+    #[tokio::test]
+    async fn test_truncation_check_does_not_touch_plain_audio_tracks() {
+        let controller = PlayerController::default();
+
+        let mut items = create_test_items(2);
+        items[0].duration = Some(1500.0);
+        controller.play_queue(items, 0).unwrap();
+        controller.seek(60.0).unwrap();
+        controller.take_end_reason();
+
+        let decision = controller.on_playback_ended().await.unwrap();
+        assert!(
+            matches!(decision, AutoplayDecision::AdvanceToNext),
+            "plain queue audio must keep advancing, got {:?}",
+            decision
+        );
+    }
+
+    /// Music and video stream from URLs that declare their own length (a static
+    /// file with byte ranges, an HLS playlist), so a truncation reaches the
+    /// player as an *error* rather than a phantom end. It is the same network
+    /// failure, and the same recovery applies — the previous behaviour turned it
+    /// into `playerStop()` and silence.
+    #[tokio::test]
+    async fn test_recoverable_error_resumes_a_music_track() {
+        let controller = PlayerController::default();
+
+        let mut items = create_test_items(3);
+        for item in &mut items {
+            item.source = MediaSource::Remote {
+                stream_url: format!("http://s/Audio/{}/stream?Static=true", item.id),
+                jellyfin_item_id: item.id.clone(),
+            };
+        }
+        controller.play_queue(items, 1).unwrap();
+        controller.seek(45.0).unwrap();
+
+        let (position, _) = controller
+            .recoverable_error_resume()
+            .expect("a streamed music track must be resumable after a network error");
+        assert_eq!(position, 45.0);
+    }
+
+    /// The resume must reload the failed track IN PLACE. `play_item` replaces the
+    /// whole queue with a single item, so recovering a track that way would throw
+    /// away the rest of the album — turning a network blip into lost state.
+    #[tokio::test]
+    async fn test_resume_keeps_the_rest_of_the_queue() {
+        let controller = PlayerController::default();
+
+        let mut items = create_test_items(3);
+        for item in &mut items {
+            item.source = MediaSource::Remote {
+                stream_url: format!("http://s/Audio/{}/stream?Static=true", item.id),
+                jellyfin_item_id: item.id.clone(),
+            };
+        }
+        controller.play_queue(items, 1).unwrap();
+
+        controller
+            .resume_stream_at(45.0)
+            .await
+            .expect("resume should succeed");
+
+        let queue = controller.queue.lock_safe();
+        assert_eq!(queue.items().len(), 3, "the queue must survive a resume");
+        assert_eq!(queue.current_index(), Some(1), "still on the same track");
+        assert_eq!(queue.current().unwrap().id, "item_1");
+    }
+
+    /// A seekable stream is re-opened by re-preparing the URL it already has and
+    /// seeking — its timeline is intact, and rewriting the URL would restart a
+    /// transcode session for no reason.
+    #[tokio::test]
+    async fn test_resume_seeks_a_seekable_stream_rather_than_rewriting_its_url() {
+        let controller = PlayerController::default();
+
+        let mut items = create_test_items(1);
+        items[0].source = MediaSource::Remote {
+            stream_url: "http://s/Audio/item_0/stream?Static=true".to_string(),
+            jellyfin_item_id: "item_0".to_string(),
+        };
+        controller.play_queue(items, 0).unwrap();
+
+        controller.resume_stream_at(45.0).await.unwrap();
+
+        match &controller.queue.lock_safe().current().unwrap().source {
+            MediaSource::Remote { stream_url, .. } => {
+                assert_eq!(
+                    stream_url, "http://s/Audio/item_0/stream?Static=true",
+                    "a seekable stream's URL must be left alone"
+                );
+            }
+            other => panic!("expected Remote source, got {:?}", other),
+        }
+        assert_eq!(
+            controller.position(),
+            45.0,
+            "and it must land at the position"
+        );
+    }
+
+    /// Downloaded media cannot fail from the network, and re-opening a local file
+    /// would paper over a real read error.
+    #[tokio::test]
+    async fn test_recoverable_error_ignores_local_media() {
+        let controller = PlayerController::default();
+
+        let mut items = create_test_items(1);
+        items[0].source = MediaSource::Local {
+            file_path: "/music/track.flac".into(),
+            jellyfin_item_id: Some("item_0".to_string()),
+        };
+        controller.play_queue(items, 0).unwrap();
+        controller.seek(45.0).unwrap();
+
+        assert!(controller.recoverable_error_resume().is_none());
+    }
+
+    /// A recoverable error during background audio-only playback is the network,
+    /// not the media — the previous behaviour (surface it, frontend stops the
+    /// player) turned a hiccup into silence. Retrying must also back off, or the
+    /// three attempts are spent inside a second and the outage outlives them.
+    #[tokio::test]
+    async fn test_recoverable_error_during_audio_only_resumes_with_backoff() {
+        let controller = PlayerController::default();
+
+        controller
+            .play_queue(vec![audio_only_episode(1500.0)], 0)
+            .unwrap();
+        controller.seek(600.0).unwrap();
+
+        let mut waits = Vec::new();
+        for attempt in 1..=stream_end::MAX_STALLED_RESUME_ATTEMPTS {
+            let (position, delay) = controller
+                .recoverable_error_resume()
+                .unwrap_or_else(|| panic!("attempt {} should still retry", attempt));
+            assert_eq!(position, 600.0);
+            waits.push(delay);
+        }
+        assert_eq!(waits, vec![2, 4, 6], "the wait must grow between attempts");
+        assert!(
+            controller.recoverable_error_resume().is_none(),
+            "a stream that keeps failing at the same spot must surface the error"
+        );
+    }
+
+    /// Plugin/channel `DirectUrl` sources are somebody else's endpoint with no
+    /// Jellyfin item behind them, so the resume has nothing to re-request.
+    #[tokio::test]
+    async fn test_recoverable_error_ignores_direct_url_playback() {
+        let controller = PlayerController::default();
+        controller.play_queue(create_test_items(2), 0).unwrap();
+
+        assert!(controller.recoverable_error_resume().is_none());
+    }
+
+    /// Re-opening the stream must land where it died and keep playing, with the
+    /// handoff base moved to the new stream's zero so returning to the
+    /// foreground still resolves an absolute position.
+    #[tokio::test]
+    async fn test_resume_truncated_stream_reloads_at_position() {
+        let controller = PlayerController::default();
+
+        controller
+            .play_queue(vec![audio_only_episode(1500.0)], 0)
+            .unwrap();
+        controller.set_background_audio_base(0.0);
+
+        controller
+            .resume_stream_at(600.0)
+            .await
+            .expect("resume should succeed");
+
+        let current = controller
+            .queue
+            .lock_safe()
+            .current()
+            .cloned()
+            .expect("the same item should still be loaded");
+        assert_eq!(current.id, "ep2", "resume must not change the item");
+        match &current.source {
+            MediaSource::Remote { stream_url, .. } => {
+                assert!(
+                    stream_url.contains("StartTimeTicks=6000000000"),
+                    "stream must re-open at 600s, got {}",
+                    stream_url
+                );
+                assert!(
+                    stream_url.contains("AudioStreamIndex=2"),
+                    "the selected audio track must survive the resume, got {}",
+                    stream_url
+                );
+            }
+            other => panic!("expected Remote source, got {:?}", other),
+        }
+        assert_eq!(
+            controller.take_background_audio_base(),
+            600.0,
+            "the re-opened stream's zero is the resume position"
+        );
+    }
+
     /// Foreground video playback keeps the countdown-driven advance: the frontend
     /// owns the navigation there, so the backend must NOT load the next episode
     /// itself (that would race the page transition and double-start playback).

src-tauri/src/player/stream_end.rs

@@ -0,0 +1,267 @@
+//! Telling a *finished* stream apart from a *truncated* one.
+//!
+//! TRACES: UR-040 | DR-129 | UT-117
+//!
+//! Background audio-only playback of a video item streams a **progressive mp3
+//! transcode over plain HTTP** (see
+//! `OnlineRepository::build_audio_only_stream_url_for_video`). That response has
+//! no reliable length — a live transcode is chunked — so when the connection
+//! drops mid-episode the data source simply sees end-of-input. ExoPlayer cannot
+//! distinguish that from the real end of the media and reports
+//! `Player.STATE_ENDED`, which the app then treats as "the episode finished".
+//!
+//! The user-visible damage is not the missed advance itself. Playback parks in
+//! ExoPlayer's `STATE_ENDED`, and the next play intent from the lockscreen,
+//! notification or a Bluetooth reconnect goes through media3's
+//! `Util.handlePlayButtonAction`, which seeks an ENDED player to its default
+//! position before playing — so **the episode starts over from 0:00**. On a
+//! flaky connection that reads as "it randomly restarts the episode".
+//!
+//! The player itself has no way to know; the *duration* does. Jellyfin gives us
+//! the item's real runtime, so an end reported well short of it is a truncation,
+//! not a finish — and the right response is to re-open the stream where it died,
+//! which is the "buffer and resume" the user expects.
+
+/// How far short of the item's runtime a stream may end and still count as a
+/// natural finish.
+///
+/// Sized to swallow the two sources of slack in the comparison — the position
+/// poll is up to 250 ms stale, and Jellyfin's reported runtime can disagree with
+/// the transcoded output by a second or two — while staying far below the
+/// minutes-long gap a dropped connection leaves. Erring long is the safe
+/// direction: a false "finished" is the bug we are fixing, whereas a false
+/// "truncated" only re-opens the stream for its last few seconds and then ends
+/// again normally.
+pub const TRUNCATED_STREAM_TOLERANCE_SECS: f64 = 10.0;
+
+/// Consecutive resume attempts allowed at the same position before giving up.
+///
+/// A resume re-opens the same URL, so a server that is genuinely gone would
+/// otherwise end → resume → end forever. Progress past the last attempt resets
+/// the budget (see [`ResumeTracker`]), so this only bounds *stuck* retries.
+pub const MAX_STALLED_RESUME_ATTEMPTS: u32 = 3;
+
+/// Position change that counts as "this is a different playback context" —
+/// either the resume made progress, or a different item is loaded.
+const RESUME_PROGRESS_EPSILON_SECS: f64 = 1.0;
+
+/// Did this end-of-stream happen far enough short of the item's runtime to be a
+/// truncation rather than a finish?
+///
+/// `position` and `duration` must be on the same timeline — for a handoff stream
+/// built with `StartTimeTicks`, that means the *absolute* position (handoff base
+/// + the player's relative position) against the item's full runtime.
+///
+/// An unknown or non-positive `duration` answers `false`: with nothing to
+/// compare against, the reported end is taken at face value (previous behaviour).
+pub fn is_truncated_end(position: f64, duration: Option<f64>, tolerance: f64) -> bool {
+    let Some(duration) = duration else {
+        return false;
+    };
+    if duration <= 0.0 {
+        return false;
+    }
+    position.max(0.0) + tolerance < duration
+}
+
+/// Rewrite an audio-only stream URL to start at `position_seconds`.
+///
+/// Resuming re-opens *the stream we were already playing*, so the URL is edited
+/// in place rather than rebuilt from the repository: every other parameter —
+/// `AudioStreamIndex` (the track the user picked in the video player),
+/// `MediaSourceId`, `api_key` — is carried over untouched, and no network call
+/// is needed to recover from a network failure.
+pub fn with_start_time(url: &str, position_seconds: f64) -> String {
+    let ticks = (position_seconds.max(0.0) * 10_000_000.0) as i64;
+    let param = format!("StartTimeTicks={}", ticks);
+
+    let (base, query) = match url.split_once('?') {
+        Some((base, query)) => (base, query),
+        // No query string at all: the URL was not built by us, but appending the
+        // parameter is still the correct request to make.
+        None => return format!("{}?{}", url, param),
+    };
+
+    let mut replaced = false;
+    let mut parts: Vec<String> = query
+        .split('&')
+        .map(|part| {
+            if part.split('=').next() == Some("StartTimeTicks") {
+                replaced = true;
+                param.clone()
+            } else {
+                part.to_string()
+            }
+        })
+        .collect();
+
+    if !replaced {
+        parts.push(param);
+    }
+
+    format!("{}?{}", base, parts.join("&"))
+}
+
+/// Budget for consecutive resume attempts that make no progress.
+///
+/// Held by the player controller across ends of the *same* stream. Any position
+/// change larger than [`RESUME_PROGRESS_EPSILON_SECS`] — the resume played on,
+/// or a different item was loaded — is a fresh context and refills the budget.
+#[derive(Debug, Default)]
+pub struct ResumeTracker {
+    last_position: Option<f64>,
+    attempts: u32,
+}
+
+impl ResumeTracker {
+    /// Record an attempt at `position`, returning its 1-based number — or `None`
+    /// once the budget is spent. Callers use the number to back off: a stream
+    /// that failed twice at the same spot is waiting on something slower than an
+    /// immediate retry can outrun.
+    pub fn allow_attempt(&mut self, position: f64) -> Option<u32> {
+        let progressed = match self.last_position {
+            Some(last) => (position - last).abs() > RESUME_PROGRESS_EPSILON_SECS,
+            None => true,
+        };
+        if progressed {
+            self.attempts = 0;
+        }
+        self.last_position = Some(position);
+        self.attempts += 1;
+        (self.attempts <= MAX_STALLED_RESUME_ATTEMPTS).then_some(self.attempts)
+    }
+
+    /// Forget the budget — a new item is playing, so nothing is stuck.
+    pub fn reset(&mut self) {
+        self.last_position = None;
+        self.attempts = 0;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_end_near_duration_is_a_natural_finish() {
+        // Episode runtime 25:00, stream ended at 24:56 — that is the end.
+        assert!(!is_truncated_end(
+            1496.0,
+            Some(1500.0),
+            TRUNCATED_STREAM_TOLERANCE_SECS
+        ));
+    }
+
+    #[test]
+    fn test_end_far_short_of_duration_is_truncated() {
+        // Episode runtime 25:00, stream died at 10:00 — the connection dropped.
+        assert!(is_truncated_end(
+            600.0,
+            Some(1500.0),
+            TRUNCATED_STREAM_TOLERANCE_SECS
+        ));
+    }
+
+    #[test]
+    fn test_unknown_duration_is_taken_at_face_value() {
+        // Nothing to compare against: keep the previous end-of-track behaviour
+        // rather than resuming a stream that may really have finished.
+        assert!(!is_truncated_end(
+            600.0,
+            None,
+            TRUNCATED_STREAM_TOLERANCE_SECS
+        ));
+        assert!(!is_truncated_end(
+            600.0,
+            Some(0.0),
+            TRUNCATED_STREAM_TOLERANCE_SECS
+        ));
+    }
+
+    #[test]
+    fn test_tolerance_boundary() {
+        // Exactly one tolerance short still counts as finished, so poll staleness
+        // and runtime rounding never fabricate a truncation.
+        assert!(!is_truncated_end(1490.0, Some(1500.0), 10.0));
+        assert!(is_truncated_end(1489.0, Some(1500.0), 10.0));
+    }
+
+    #[test]
+    fn test_with_start_time_replaces_existing_ticks() {
+        let url = "http://s/Audio/ep2/universal?api_key=k&AudioStreamIndex=2&StartTimeTicks=1200000000&Container=mp3";
+        let out = with_start_time(url, 600.0);
+        assert_eq!(
+            out,
+            "http://s/Audio/ep2/universal?api_key=k&AudioStreamIndex=2&StartTimeTicks=6000000000&Container=mp3"
+        );
+    }
+
+    #[test]
+    fn test_with_start_time_appends_when_absent() {
+        // The next-episode stream is built without StartTimeTicks.
+        let url = "http://s/Audio/ep3/universal?api_key=k&AudioStreamIndex=0";
+        let out = with_start_time(url, 90.0);
+        assert_eq!(
+            out,
+            "http://s/Audio/ep3/universal?api_key=k&AudioStreamIndex=0&StartTimeTicks=900000000"
+        );
+    }
+
+    #[test]
+    fn test_with_start_time_preserves_selected_audio_track() {
+        // The whole point of editing the URL instead of rebuilding it: the track
+        // the user chose in the video player survives the resume.
+        let url = "http://s/Audio/ep2/universal?AudioStreamIndex=3&MediaSourceId=src-1";
+        let out = with_start_time(url, 10.0);
+        assert!(out.contains("AudioStreamIndex=3"));
+        assert!(out.contains("MediaSourceId=src-1"));
+    }
+
+    #[test]
+    fn test_with_start_time_without_query() {
+        assert_eq!(
+            with_start_time("http://s/Audio/ep2/universal", 1.0),
+            "http://s/Audio/ep2/universal?StartTimeTicks=10000000"
+        );
+    }
+
+    #[test]
+    fn test_resume_tracker_bounds_stalled_retries() {
+        let mut tracker = ResumeTracker::default();
+        // Same position over and over: the stream is not recovering.
+        for n in 1..=MAX_STALLED_RESUME_ATTEMPTS {
+            assert_eq!(
+                tracker.allow_attempt(600.0),
+                Some(n),
+                "attempts are numbered so callers can back off"
+            );
+        }
+        assert_eq!(
+            tracker.allow_attempt(600.0),
+            None,
+            "a stream that ends at the same position every time must stop retrying"
+        );
+    }
+
+    #[test]
+    fn test_resume_tracker_refills_after_progress() {
+        let mut tracker = ResumeTracker::default();
+        for _ in 0..MAX_STALLED_RESUME_ATTEMPTS {
+            tracker.allow_attempt(600.0);
+        }
+        assert_eq!(tracker.allow_attempt(600.0), None);
+        // The next drop happened further in — the resumes are working, so the
+        // budget must not be exhausted by earlier trouble.
+        assert_eq!(tracker.allow_attempt(900.0), Some(1));
+    }
+
+    #[test]
+    fn test_resume_tracker_reset() {
+        let mut tracker = ResumeTracker::default();
+        for _ in 0..=MAX_STALLED_RESUME_ATTEMPTS {
+            tracker.allow_attempt(600.0);
+        }
+        tracker.reset();
+        assert_eq!(tracker.allow_attempt(600.0), Some(1));
+    }
+}

src-tauri/src/repository/hybrid.rs

@@ -41,12 +41,34 @@ impl HybridRepository {
         }
     }
 
+    /// The signed-in user this repository acts for.
+    ///
+    /// TRACES: UR-069 | DR-120
+    pub fn user_id(&self) -> &str {
+        self.online.user_id()
+    }
+
     /// Download raw bytes from a URL using the shared authenticated HTTP client.
     /// Delegates to online repository for connection reuse and proper auth.
     pub async fn download_bytes(&self, url: &str) -> Result<Vec<u8>, String> {
         self.online.download_bytes(url).await
     }
 
+    /// Remove catalog entries the server no longer has. Cache-only, so it goes
+    /// straight to the offline repository. Callers must only invoke this after a
+    /// crawl in which every library succeeded — see
+    /// `OfflineRepository::prune_stale_catalog` for why a partial crawl must not
+    /// sweep.
+    ///
+    /// TRACES: UR-065 | DR-110
+    pub async fn prune_stale_catalog(
+        &self,
+        cutoff: &str,
+        item_types: &[String],
+    ) -> Result<usize, RepoError> {
+        self.offline.prune_stale_catalog(cutoff, item_types).await
+    }
+
     /// Query the JRay plugin for actors on screen at time `t`. Online-only
     /// (the plugin lives on the Jellyfin server); empty when JRay isn't present.
     pub async fn get_jray_actors(
@@ -113,6 +135,41 @@ impl HybridRepository {
             .await
     }
 
+    /// Favourites held locally, without touching the server. Backs the instant
+    /// leg of the two-phase favourites read in the command layer.
+    ///
+    /// TRACES: UR-067 | DR-115
+    pub async fn get_favorites_cache_only(
+        &self,
+        scope: SearchScope,
+        options: Option<GetItemsOptions>,
+    ) -> Result<SearchResult, RepoError> {
+        let offline = Arc::clone(&self.offline);
+        self.cache_with_timeout(async move { offline.get_favorites(scope, options).await })
+            .await
+    }
+
+    /// Favourites straight from the server, persisted to the cache on the way
+    /// through — which is also what mirrors their favourite flags into
+    /// `user_data` (DR-114), so the next offline read agrees with the server.
+    ///
+    /// TRACES: UR-067 | DR-115
+    pub async fn get_favorites_server_only(
+        &self,
+        scope: SearchScope,
+        options: Option<GetItemsOptions>,
+    ) -> Result<SearchResult, RepoError> {
+        let result = self.online.get_favorites(scope, options).await?;
+        if !result.items.is_empty() {
+            // Favourites span libraries, so there is no single parent to file
+            // them under; the parent id is only used for stub rows.
+            if let Err(e) = self.offline.save_to_cache("favorites", &result.items).await {
+                debug!("[HybridRepo] Failed to cache favourites: {:?}", e);
+            }
+        }
+        Ok(result)
+    }
+
     /// Fetch a folder's items from the live server and persist them to the
     /// offline cache synchronously (unlike `get_items`, which saves in a
     /// fire-and-forget background task after a 100ms cache race).
@@ -790,6 +847,42 @@ impl MediaRepository for HybridRepository {
         self.parallel_race(cache_future, server_future).await
     }
 
+    /// TRACES: UR-067 | DR-115
+    async fn get_favorites(
+        &self,
+        scope: SearchScope,
+        options: Option<GetItemsOptions>,
+    ) -> Result<SearchResult, RepoError> {
+        let offline = Arc::clone(&self.offline);
+        let online = Arc::clone(&self.online);
+        let opts_clone = options.clone();
+
+        let cache_result = self
+            .cache_with_timeout(async move { offline.get_favorites(scope, opts_clone).await })
+            .await;
+
+        // Downloads-only gate: with "Show all server media" off, an empty local
+        // result means "nothing favourited is on this device" and is
+        // authoritative. Falling through to the server here would re-pad the
+        // page with the full favourited catalog and defeat the filter (DR-080).
+        if !crate::repository::offline::include_catalog_browse() {
+            if let Ok(data) = &cache_result {
+                return Ok(data.clone());
+            }
+        }
+
+        if let Ok(data) = &cache_result {
+            if data.has_content() {
+                return Ok(data.clone());
+            }
+        }
+
+        match online.get_favorites(scope, options).await {
+            Ok(data) => Ok(data),
+            Err(e) => cache_result.or(Err(e)),
+        }
+    }
+
     async fn get_similar_items(
         &self,
         item_id: &str,
@@ -1133,6 +1226,14 @@ mod tests {
             unimplemented!()
         }
 
+        async fn get_favorites(
+            &self,
+            _scope: SearchScope,
+            _options: Option<GetItemsOptions>,
+        ) -> Result<SearchResult, RepoError> {
+            unimplemented!()
+        }
+
         async fn clear_watch_history(&self, _item_id: &str) -> Result<(), RepoError> {
             unimplemented!()
         }
@@ -1395,6 +1496,14 @@ mod tests {
             unimplemented!()
         }
 
+        async fn get_favorites(
+            &self,
+            _scope: SearchScope,
+            _options: Option<GetItemsOptions>,
+        ) -> Result<SearchResult, RepoError> {
+            unimplemented!()
+        }
+
         async fn clear_watch_history(&self, _item_id: &str) -> Result<(), RepoError> {
             unimplemented!()
         }

src-tauri/src/repository/mod.rs

@@ -212,6 +212,20 @@ pub trait MediaRepository: Send + Sync {
     /// Unmark item as favorite
     async fn unmark_favorite(&self, item_id: &str) -> Result<(), RepoError>;
 
+    /// Everything the viewer has favourited, across every library.
+    ///
+    /// Separate from `get_items` because favourites span libraries and
+    /// `get_items` is `ParentId`-shaped. `scope` is the opaque enum the
+    /// frontend sends; this layer expands it to item types (DR-063) so no
+    /// Jellyfin taxonomy is needed on the other side of the IPC boundary.
+    ///
+    /// TRACES: UR-067 | DR-115, JA-033 | UT-100, UT-101
+    async fn get_favorites(
+        &self,
+        scope: SearchScope,
+        options: Option<GetItemsOptions>,
+    ) -> Result<SearchResult, RepoError>;
+
     /// Erase the viewer's watch history for an item: clear its played flag and
     /// its resume position. On a container (series, season) this applies to
     /// everything inside it, so a series is returned to "never watched" and

src-tauri/src/repository/online.rs

@@ -48,6 +48,12 @@ pub struct OnlineRepository {
 }
 
 impl OnlineRepository {
+    /// The signed-in user these requests are made as. Needed by the favourites
+    /// drain, which reads this user's queued rows. TRACES: UR-069 | DR-120
+    pub fn user_id(&self) -> &str {
+        &self.user_id
+    }
+
     pub fn new(
         http_client: Arc<HttpClient>,
         server_url: String,
@@ -560,6 +566,138 @@ struct JellyfinItem {
     media_streams: Option<Vec<JellyfinMediaStream>>,
     media_sources: Option<Vec<JellyfinMediaSource>>,
     people: Option<Vec<crate::repository::types::Person>>,
+    user_data: Option<JellyfinUserData>,
+}
+
+/// Per-user state Jellyfin attaches to an item (favourite, played, resume).
+///
+/// Returned on every `/Users/{uid}/Items*` response; we additionally name
+/// `UserData` in the `Fields=` list so the shape is explicit rather than
+/// dependent on the server's default field set.
+///
+/// TRACES: UR-069 | DR-113, JA-034 | UT-099
+#[derive(Debug, Deserialize, Clone)]
+#[serde(rename_all = "PascalCase")]
+struct JellyfinUserData {
+    playback_position_ticks: Option<i64>,
+    #[serde(rename = "Played")]
+    is_played: Option<bool>,
+    is_favorite: Option<bool>,
+    play_count: Option<i32>,
+    last_played_date: Option<String>,
+}
+
+impl From<JellyfinUserData> for UserData {
+    fn from(jf: JellyfinUserData) -> Self {
+        UserData {
+            playback_position_ticks: jf.playback_position_ticks,
+            playback_position_ms: jf.playback_position_ticks.map(crate::domain::ticks_to_ms),
+            is_played: jf.is_played,
+            is_favorite: jf.is_favorite,
+            play_count: jf.play_count,
+            last_played_date: jf.last_played_date,
+            playback_context_type: None,
+            playback_context_id: None,
+        }
+    }
+}
+
+/// Build the Jellyfin endpoint for a folder listing.
+///
+/// Extracted from `get_items` so the query it produces — in particular the
+/// favourites filter — can be asserted without standing up an HTTP server.
+///
+/// TRACES: UR-007, UR-067 | DR-116 | UT-104
+fn build_get_items_endpoint(
+    user_id: &str,
+    parent_id: &str,
+    options: Option<&GetItemsOptions>,
+) -> String {
+    let mut endpoint = format!("/Users/{}/Items?ParentId={}", user_id, parent_id);
+
+    if let Some(opts) = options {
+        if let Some(limit) = opts.limit {
+            endpoint.push_str(&format!("&Limit={}", limit));
+        }
+        if let Some(start_index) = opts.start_index {
+            endpoint.push_str(&format!("&StartIndex={}", start_index));
+        }
+        if let Some(types) = &opts.include_item_types {
+            endpoint.push_str(&format!("&IncludeItemTypes={}", types.join(",")));
+        }
+        if let Some(sort_by) = &opts.sort_by {
+            endpoint.push_str(&format!("&SortBy={}", sort_by));
+        }
+        if let Some(sort_order) = &opts.sort_order {
+            endpoint.push_str(&format!("&SortOrder={}", sort_order));
+        }
+        if let Some(recursive) = opts.recursive {
+            endpoint.push_str(&format!("&Recursive={}", recursive));
+        }
+        if let Some(genres) = &opts.genres {
+            if !genres.is_empty() {
+                // Genre names may contain spaces/ampersands, so percent-encode each.
+                let encoded: Vec<String> = genres
+                    .iter()
+                    .map(|g| urlencoding::encode(g).into_owned())
+                    .collect();
+                endpoint.push_str(&format!("&Genres={}", encoded.join("|")));
+            }
+        }
+        // TRACES: UR-067 | DR-116 | UT-104
+        if opts.favorites_only == Some(true) {
+            endpoint.push_str("&Filters=IsFavorite");
+        }
+    }
+
+    // Request image fields for list views (People only needed in get_item
+    // detail view). Genres is needed so cached items carry their genres,
+    // which lets the offline store derive genre lists + per-genre counts.
+    endpoint
+        .push_str("&Fields=BackdropImageTags,ParentBackdropImageTags,Genres,PremiereDate,UserData");
+    endpoint
+}
+
+/// Build the Jellyfin endpoint for a favourites listing.
+///
+/// Pulled out of `get_favorites` so the query can be asserted without an HTTP
+/// server. `scope` is expanded here — `SearchScope::All` yields `None`, and the
+/// `IncludeItemTypes` filter is then **omitted entirely** rather than sent as a
+/// union, which would silently drop every type nobody enumerated (see
+/// `SearchScope::item_types`).
+///
+/// TRACES: UR-067 | DR-115, JA-033 | UT-100
+fn build_favorites_endpoint(
+    user_id: &str,
+    scope: SearchScope,
+    options: Option<&GetItemsOptions>,
+) -> String {
+    let mut endpoint = format!("/Users/{}/Items?Filters=IsFavorite&Recursive=true", user_id);
+
+    if let Some(types) = scope.item_types() {
+        endpoint.push_str(&format!("&IncludeItemTypes={}", types.join(",")));
+    }
+
+    // Jellyfin has no "date favourited", so name order is the only stable sort
+    // available; callers may still override it.
+    let sort_by = options
+        .and_then(|o| o.sort_by.as_deref())
+        .unwrap_or("SortName");
+    let sort_order = options
+        .and_then(|o| o.sort_order.as_deref())
+        .unwrap_or("Ascending");
+    endpoint.push_str(&format!("&SortBy={}&SortOrder={}", sort_by, sort_order));
+
+    if let Some(limit) = options.and_then(|o| o.limit) {
+        endpoint.push_str(&format!("&Limit={}", limit));
+    }
+    if let Some(start_index) = options.and_then(|o| o.start_index) {
+        endpoint.push_str(&format!("&StartIndex={}", start_index));
+    }
+
+    endpoint
+        .push_str("&Fields=BackdropImageTags,ParentBackdropImageTags,Genres,PremiereDate,UserData");
+    endpoint
 }
 
 // ImageTags from Jellyfin API - can be a HashMap with various image type keys
@@ -653,7 +791,9 @@ impl JellyfinItem {
             series_name: self.series_name,
             season_id: self.season_id,
             season_name: self.season_name,
-            user_data: None, // User data not included in basic item responses
+            // Favourite/played/resume state as the server sees it. TRACES:
+            // UR-069 | DR-113, JA-034
+            user_data: self.user_data.map(UserData::from),
             media_streams: self.media_streams.map(|streams| {
                 streams
                     .into_iter()
@@ -728,43 +868,7 @@ impl MediaRepository for OnlineRepository {
         parent_id: &str,
         options: Option<GetItemsOptions>,
     ) -> Result<SearchResult, RepoError> {
-        let mut endpoint = format!("/Users/{}/Items?ParentId={}", self.user_id, parent_id);
-
-        if let Some(opts) = options {
-            if let Some(limit) = opts.limit {
-                endpoint.push_str(&format!("&Limit={}", limit));
-            }
-            if let Some(start_index) = opts.start_index {
-                endpoint.push_str(&format!("&StartIndex={}", start_index));
-            }
-            if let Some(types) = opts.include_item_types {
-                endpoint.push_str(&format!("&IncludeItemTypes={}", types.join(",")));
-            }
-            if let Some(sort_by) = opts.sort_by {
-                endpoint.push_str(&format!("&SortBy={}", sort_by));
-            }
-            if let Some(sort_order) = opts.sort_order {
-                endpoint.push_str(&format!("&SortOrder={}", sort_order));
-            }
-            if let Some(recursive) = opts.recursive {
-                endpoint.push_str(&format!("&Recursive={}", recursive));
-            }
-            if let Some(genres) = opts.genres {
-                if !genres.is_empty() {
-                    // Genre names may contain spaces/ampersands, so percent-encode each.
-                    let encoded: Vec<String> = genres
-                        .iter()
-                        .map(|g| urlencoding::encode(g).into_owned())
-                        .collect();
-                    endpoint.push_str(&format!("&Genres={}", encoded.join("|")));
-                }
-            }
-        }
-
-        // Request image fields for list views (People only needed in get_item
-        // detail view). Genres is needed so cached items carry their genres,
-        // which lets the offline store derive genre lists + per-genre counts.
-        endpoint.push_str("&Fields=BackdropImageTags,ParentBackdropImageTags,Genres,PremiereDate");
+        let endpoint = build_get_items_endpoint(&self.user_id, parent_id, options.as_ref());
 
         let response: ItemsResponse = self.get_json(&endpoint).await?;
 
@@ -779,7 +883,7 @@ impl MediaRepository for OnlineRepository {
     }
 
     async fn get_item(&self, item_id: &str) -> Result<MediaItem, RepoError> {
-        let endpoint = format!("/Users/{}/Items/{}?Fields=BackdropImageTags,ParentBackdropImageTags,People,MediaStreams,MediaSources,PremiereDate", self.user_id, item_id);
+        let endpoint = format!("/Users/{}/Items/{}?Fields=BackdropImageTags,ParentBackdropImageTags,People,MediaStreams,MediaSources,PremiereDate,UserData", self.user_id, item_id);
 
         let item: JellyfinItem = self.get_json(&endpoint).await?;
         let media_item = item.to_media_item(self.user_id.clone());
@@ -794,7 +898,7 @@ impl MediaRepository for OnlineRepository {
     ) -> Result<Vec<MediaItem>, RepoError> {
         let limit_str = limit.unwrap_or(16);
         let endpoint = format!(
-            "/Users/{}/Items/Latest?ParentId={}&Limit={}&Fields=BackdropImageTags,ParentBackdropImageTags",
+            "/Users/{}/Items/Latest?ParentId={}&Limit={}&Fields=BackdropImageTags,ParentBackdropImageTags,UserData",
             self.user_id, parent_id, limit_str
         );
 
@@ -812,7 +916,7 @@ impl MediaRepository for OnlineRepository {
     ) -> Result<Vec<MediaItem>, RepoError> {
         let limit_str = limit.unwrap_or(16);
         let mut endpoint = format!(
-            "/Users/{}/Items/Resume?Limit={}&MediaTypes=Video&Fields=BackdropImageTags,ParentBackdropImageTags",
+            "/Users/{}/Items/Resume?Limit={}&MediaTypes=Video&Fields=BackdropImageTags,ParentBackdropImageTags,UserData",
             self.user_id, limit_str
         );
 
@@ -835,7 +939,7 @@ impl MediaRepository for OnlineRepository {
     ) -> Result<Vec<MediaItem>, RepoError> {
         let limit_str = limit.unwrap_or(16);
         let mut endpoint = format!(
-            "/Shows/NextUp?UserId={}&Limit={}&Fields=BackdropImageTags,ParentBackdropImageTags",
+            "/Shows/NextUp?UserId={}&Limit={}&Fields=BackdropImageTags,ParentBackdropImageTags,UserData",
             self.user_id, limit_str
         );
 
@@ -859,7 +963,7 @@ impl MediaRepository for OnlineRepository {
         // Fetch more items to account for grouping reducing the count
         let fetch_limit = limit_val * 3;
         let endpoint = format!(
-            "/Users/{}/Items?SortBy=DatePlayed&SortOrder=Descending&IncludeItemTypes=Audio&Limit={}&Recursive=true&Filters=IsPlayed&Fields=BackdropImageTags,ParentBackdropImageTags",
+            "/Users/{}/Items?SortBy=DatePlayed&SortOrder=Descending&IncludeItemTypes=Audio&Limit={}&Recursive=true&Filters=IsPlayed&Fields=BackdropImageTags,ParentBackdropImageTags,UserData",
             self.user_id, fetch_limit
         );
 
@@ -993,7 +1097,7 @@ impl MediaRepository for OnlineRepository {
         // Filters=IsPlayed keeps only albums the user has actually listened to,
         // and SortBy=DatePlayed ascending surfaces the ones they've neglected.
         let mut endpoint = format!(
-            "/Users/{}/Items?SortBy=DatePlayed&SortOrder=Ascending&IncludeItemTypes=MusicAlbum&Limit={}&Recursive=true&Filters=IsPlayed&Fields=BackdropImageTags,ParentBackdropImageTags",
+            "/Users/{}/Items?SortBy=DatePlayed&SortOrder=Ascending&IncludeItemTypes=MusicAlbum&Limit={}&Recursive=true&Filters=IsPlayed&Fields=BackdropImageTags,ParentBackdropImageTags,UserData",
             self.user_id, limit_val
         );
 
@@ -1012,7 +1116,7 @@ impl MediaRepository for OnlineRepository {
     async fn get_resume_movies(&self, limit: Option<usize>) -> Result<Vec<MediaItem>, RepoError> {
         let limit_str = limit.unwrap_or(16);
         let endpoint = format!(
-            "/Users/{}/Items/Resume?Limit={}&MediaTypes=Video&IncludeItemTypes=Movie&Fields=BackdropImageTags,ParentBackdropImageTags",
+            "/Users/{}/Items/Resume?Limit={}&MediaTypes=Video&IncludeItemTypes=Movie&Fields=BackdropImageTags,ParentBackdropImageTags,UserData",
             self.user_id, limit_str
         );
 
@@ -1113,7 +1217,9 @@ impl MediaRepository for OnlineRepository {
 
         // Request image fields for list views (plus Genres so cached items
         // carry genres for offline genre lists/counts).
-        endpoint.push_str("&Fields=BackdropImageTags,ParentBackdropImageTags,Genres,PremiereDate");
+        endpoint.push_str(
+            "&Fields=BackdropImageTags,ParentBackdropImageTags,Genres,PremiereDate,UserData",
+        );
 
         let response: ItemsResponse = self.get_json(&endpoint).await?;
         Ok(SearchResult {
@@ -1654,6 +1760,25 @@ impl MediaRepository for OnlineRepository {
         self.post_json(&endpoint, &serde_json::json!({})).await
     }
 
+    /// TRACES: UR-067 | DR-115, JA-033 | UT-100
+    async fn get_favorites(
+        &self,
+        scope: SearchScope,
+        options: Option<GetItemsOptions>,
+    ) -> Result<SearchResult, RepoError> {
+        let endpoint = build_favorites_endpoint(&self.user_id, scope, options.as_ref());
+        let response: ItemsResponse = self.get_json(&endpoint).await?;
+
+        Ok(SearchResult {
+            items: response
+                .items
+                .into_iter()
+                .map(|item| item.to_media_item(self.user_id.clone()))
+                .collect(),
+            total_record_count: response.total_record_count,
+        })
+    }
+
     async fn unmark_favorite(&self, item_id: &str) -> Result<(), RepoError> {
         let endpoint = format!("/Users/{}/FavoriteItems/{}", self.user_id, item_id);
         let url = format!("{}{}", self.server_url, endpoint);
@@ -1747,7 +1872,7 @@ impl MediaRepository for OnlineRepository {
         let limit = options.as_ref().and_then(|o| o.limit).unwrap_or(100);
 
         let mut endpoint = format!(
-            "/Users/{}/Items?PersonIds={}&Limit={}&Recursive=true&Fields=BackdropImageTags,ParentBackdropImageTags",
+            "/Users/{}/Items?PersonIds={}&Limit={}&Recursive=true&Fields=BackdropImageTags,ParentBackdropImageTags,UserData",
             self.user_id, person_id, limit
         );
 
@@ -1781,7 +1906,7 @@ impl MediaRepository for OnlineRepository {
 
         // Try the /Similar endpoint which works for most items
         let endpoint = format!(
-            "/Items/{}/Similar?UserId={}&Limit={}&Fields=BackdropImageTags,ParentBackdropImageTags",
+            "/Items/{}/Similar?UserId={}&Limit={}&Fields=BackdropImageTags,ParentBackdropImageTags,UserData",
             item_id, self.user_id, limit_str
         );
 
@@ -2369,6 +2494,140 @@ mod tests {
         );
     }
 
+    /// UT-100 — the favourites endpoint asks the server for favourites, scoped.
+    ///
+    /// TRACES: UR-067 | DR-115, JA-033 | UT-100
+    #[test]
+    fn test_build_favorites_endpoint_scopes_and_filters() {
+        let movies = build_favorites_endpoint("u1", SearchScope::Movies, None);
+        assert!(movies.starts_with("/Users/u1/Items?Filters=IsFavorite&Recursive=true"));
+        assert!(movies.contains("&IncludeItemTypes=Movie"));
+        // Jellyfin has no favourite timestamp, so name order is the default.
+        assert!(movies.contains("&SortBy=SortName&SortOrder=Ascending"));
+        // Hearts must render on the returned cards.
+        assert!(movies.contains("UserData"));
+
+        // Tv covers both the show and any individually favourited episode.
+        let tv = build_favorites_endpoint("u1", SearchScope::Tv, None);
+        assert!(tv.contains("&IncludeItemTypes=Series,Episode"));
+
+        let music = build_favorites_endpoint("u1", SearchScope::Music, None);
+        assert!(music.contains("&IncludeItemTypes=MusicAlbum,MusicArtist,Audio,Playlist"));
+    }
+
+    /// `All` must omit the type filter entirely rather than send a union, which
+    /// would silently drop every type nobody enumerated.
+    ///
+    /// TRACES: UR-067 | DR-115 | UT-100
+    #[test]
+    fn test_build_favorites_endpoint_all_scope_omits_type_filter() {
+        let all = build_favorites_endpoint("u1", SearchScope::All, None);
+        assert!(!all.contains("IncludeItemTypes"));
+    }
+
+    /// Paging and an explicit sort still reach the server.
+    ///
+    /// TRACES: UR-067 | DR-115 | UT-100
+    #[test]
+    fn test_build_favorites_endpoint_honours_paging_and_sort() {
+        let endpoint = build_favorites_endpoint(
+            "u1",
+            SearchScope::All,
+            Some(&GetItemsOptions {
+                limit: Some(20),
+                start_index: Some(40),
+                sort_by: Some("Random".to_string()),
+                sort_order: Some("Descending".to_string()),
+                ..Default::default()
+            }),
+        );
+        assert!(endpoint.contains("&Limit=20"));
+        assert!(endpoint.contains("&StartIndex=40"));
+        assert!(endpoint.contains("&SortBy=Random&SortOrder=Descending"));
+    }
+
+    /// UT-104 — the in-library favourites toggle reaches the server as
+    /// `Filters=IsFavorite`, and is absent unless asked for.
+    ///
+    /// TRACES: UR-067 | DR-116 | UT-104
+    #[test]
+    fn test_get_items_endpoint_applies_favorites_only() {
+        let plain = build_get_items_endpoint("u1", "lib-1", None);
+        assert!(!plain.contains("Filters=IsFavorite"));
+
+        let filtered = build_get_items_endpoint(
+            "u1",
+            "lib-1",
+            Some(&GetItemsOptions {
+                favorites_only: Some(true),
+                include_item_types: Some(vec!["Movie".to_string()]),
+                ..Default::default()
+            }),
+        );
+        assert!(filtered.contains("&Filters=IsFavorite"));
+        // Composes with the filters already there rather than replacing them.
+        assert!(filtered.contains("&IncludeItemTypes=Movie"));
+        assert!(filtered.contains("ParentId=lib-1"));
+
+        // Explicitly false is not a request to filter.
+        let off = build_get_items_endpoint(
+            "u1",
+            "lib-1",
+            Some(&GetItemsOptions {
+                favorites_only: Some(false),
+                ..Default::default()
+            }),
+        );
+        assert!(!off.contains("Filters=IsFavorite"));
+    }
+
+    /// UT-099 — a Jellyfin item's `UserData` reaches `MediaItem.user_data`.
+    ///
+    /// Before DR-113 this mapping was hardcoded to `None`, so nothing outside
+    /// the mini player could know an item was favourited.
+    ///
+    /// TRACES: UR-069 | DR-113, JA-034 | UT-099
+    #[test]
+    fn test_jellyfin_item_maps_user_data_favorite() {
+        let json = r#"{
+            "Id": "movie123",
+            "Name": "Test Movie",
+            "Type": "Movie",
+            "UserData": {
+                "PlaybackPositionTicks": 6000000000,
+                "Played": false,
+                "IsFavorite": true,
+                "PlayCount": 2,
+                "LastPlayedDate": "2026-08-01T12:00:00Z"
+            }
+        }"#;
+
+        let item: JellyfinItem = serde_json::from_str(json).expect("item should deserialize");
+        let media = item.to_media_item("server1".to_string());
+
+        let user_data = media.user_data.expect("user data should be mapped");
+        assert_eq!(user_data.is_favorite, Some(true));
+        assert_eq!(user_data.is_played, Some(false));
+        assert_eq!(user_data.play_count, Some(2));
+        assert_eq!(user_data.playback_position_ticks, Some(6_000_000_000));
+        // Ticks are converted for the frontend, which never divides them itself.
+        assert_eq!(user_data.playback_position_ms, Some(600_000));
+    }
+
+    /// An item without `UserData` still maps — the field is optional, and every
+    /// non-user-scoped endpoint omits it.
+    ///
+    /// TRACES: UR-069 | DR-113 | UT-099
+    #[test]
+    fn test_jellyfin_item_without_user_data_maps_to_none() {
+        let json = r#"{"Id": "x", "Name": "No User Data", "Type": "Movie"}"#;
+
+        let item: JellyfinItem = serde_json::from_str(json).expect("item should deserialize");
+        let media = item.to_media_item("server1".to_string());
+
+        assert!(media.user_data.is_none());
+    }
+
     #[test]
     fn test_jellyfin_item_deserialize_with_artist_items() {
         // Test that ArtistItems with PascalCase fields deserialize correctly

src-tauri/src/repository/types.rs

@@ -292,6 +292,12 @@ pub struct GetItemsOptions {
     pub fields: Option<Vec<String>>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub genres: Option<Vec<String>>,
+    /// Restrict the listing to favourited items. Backs the per-library
+    /// favourites toggle; composes with every other filter here.
+    ///
+    /// TRACES: UR-067 | DR-116 | UT-104
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub favorites_only: Option<bool>,
 }
 
 /// An opaque search scope the frontend selects; Rust owns what it *means*.

src-tauri/src/storage/schema.rs

@@ -25,6 +25,9 @@ pub const MIGRATIONS: &[(&str, &str)] = &[
     ("018_items_is_folder", MIGRATION_018),
     ("019_genres_cache", MIGRATION_019),
     ("020_items_season_index", MIGRATION_020),
+    ("021_rebuild_items_fts", MIGRATION_021),
+    ("022_people_fts", MIGRATION_022),
+    ("023_downloads_expiry", MIGRATION_023),
 ];
 
 /// Initial schema migration
@@ -728,3 +731,91 @@ CREATE INDEX IF NOT EXISTS idx_genres_scope ON genres(server_id, library_id);
 const MIGRATION_020: &str = r#"
 CREATE INDEX IF NOT EXISTS idx_items_season ON items(season_id);
 "#;
+
+/// Discard and rebuild the FTS index from the `items` table.
+///
+/// Until DR-110, `save_to_cache` used `INSERT OR REPLACE INTO items`. REPLACE
+/// deletes the conflicting row and inserts a new one, but SQLite only fires
+/// `AFTER DELETE` triggers on that implicit delete when `recursive_triggers` is
+/// enabled — it is not (storage/mod.rs sets only `foreign_keys` and
+/// `journal_mode`), so `items_ad` never ran and the old index row was orphaned.
+/// Worse, `items.id` is a `TEXT PRIMARY KEY`, so the replacement row also took a
+/// *fresh rowid* and `items_ai` appended a second entry. Every catalog pass
+/// therefore left another duplicate behind, and existing installs carry one
+/// stale entry per item per sync since the database was created.
+///
+/// This was invisible in results — the `JOIN items_fts fts ON fts.rowid =
+/// i.rowid` drops rowids that no longer exist — but it degrades `MATCH`
+/// permanently, and it becomes a *correctness* problem the moment rowids are
+/// freed and reused: a new item landing on a freed rowid inherits the orphan's
+/// index entry and matches queries for the deleted item's title. The DR-110
+/// deletion sweep frees rowids, so this rebuild must run before it.
+///
+/// `'rebuild'` is the FTS5 command for exactly this: it truncates the index and
+/// repopulates it from the external content table.
+///
+/// TRACES: UR-065 | DR-110
+const MIGRATION_021: &str = r#"
+INSERT INTO items_fts(items_fts) VALUES('rebuild');
+"#;
+
+/// Full-text index over `people`, mirroring `items_fts`.
+///
+/// People live in their own table (migration 009) rather than in `items`, and
+/// had no FTS index at all — so the People group UR-060 requires could only ever
+/// be filled by the server leg of search. With the local index now answering
+/// first, an actor's name has to be findable offline too.
+///
+/// TRACES: UR-065, UR-060 | DR-111
+const MIGRATION_022: &str = r#"
+CREATE VIRTUAL TABLE IF NOT EXISTS people_fts USING fts5(
+    name,
+    overview,
+    content='people',
+    content_rowid='rowid'
+);
+
+CREATE TRIGGER IF NOT EXISTS people_ai AFTER INSERT ON people BEGIN
+    INSERT INTO people_fts(rowid, name, overview)
+    VALUES (new.rowid, new.name, new.overview);
+END;
+
+CREATE TRIGGER IF NOT EXISTS people_ad AFTER DELETE ON people BEGIN
+    INSERT INTO people_fts(people_fts, rowid, name, overview)
+    VALUES('delete', old.rowid, old.name, old.overview);
+END;
+
+CREATE TRIGGER IF NOT EXISTS people_au AFTER UPDATE ON people BEGIN
+    INSERT INTO people_fts(people_fts, rowid, name, overview)
+    VALUES('delete', old.rowid, old.name, old.overview);
+    INSERT INTO people_fts(rowid, name, overview)
+    VALUES (new.rowid, new.name, new.overview);
+END;
+
+-- Backfill for rows cached before this index existed.
+INSERT INTO people_fts(people_fts) VALUES('rebuild');
+"#;
+
+/// Give temporary downloads a life limit.
+///
+/// A cache entry is not a different kind of object from a download — it is a
+/// download with a shorter life. Modelling it as one `downloads` row with an
+/// expiry (rather than a parallel cache store) means there is a single storage
+/// accounting, a single eviction path, and no way for a cache and a download
+/// library to disagree about what is on disk.
+///
+/// `expires_at` is NULL for permanent rows, which is every row that exists
+/// today: `download_source` defaults to `'user'`, and a user's own download
+/// never expires. Only `'auto'` rows get a timestamp, and they are reclaimed by
+/// whichever comes first — the expiry passing, or LRU eviction under space
+/// pressure (DR-126).
+///
+/// TRACES: UR-071 | DR-127
+const MIGRATION_023: &str = r#"
+ALTER TABLE downloads ADD COLUMN expires_at TEXT;
+
+-- Reclaim scans filter on expiry among 'auto' rows; index both so the sweep
+-- stays cheap as the cache tier grows.
+CREATE INDEX IF NOT EXISTS idx_downloads_expiry
+    ON downloads(download_source, expires_at);
+"#;