stop the poll loop and the notification on inactive waiting rows

The cross-source reconcile went too far: a source switched away from can have
waiting rows that promote will never hand over, since it resolves only while
active and downloads are only ever started for the active source. Those rows
kept the loop alive forever, so it scanned the whole table and ran every second
for the life of the process, and the notification sat there claiming to be
downloading songs that were not moving.

Both now count a row as in flight only when the engine owns it or it belongs to
the active source. Reconcile still covers every source, so a download running
for a source you switched away from keeps updating.

The downloaded-filter DAO test gains a second artist that must be excluded (its
first version passed whichever artist clause you wrote), and the preference test
covers the legacy three-field value and a downloaded round trip.

app/src/test/kotlin/com/subtracks/data/db/LibraryDaoTest.kt

@@ -302,13 +302,20 @@ class LibraryDaoTest {
                 listOf(
                     album(sourceId, "al-1", "Downloaded", year = 2000, starred = null),
                     album(sourceId, "al-2", "Empty", year = 2001, starred = null),
+                    album(sourceId, "al-3", "Elsewhere", year = 2002, starred = null).copy(artistId = "ar-2"),
+                ),
+            )
+            dao.upsertArtists(
+                listOf(
+                    artist(sourceId, "ar-1", "Artist", albumCount = 2),
+                    artist(sourceId, "ar-2", "Other", albumCount = 1),
                 ),
             )
-            dao.upsertArtists(listOf(artist(sourceId, "ar-1", "Artist", albumCount = 2)))
             dao.upsertSongs(
                 listOf(
                     song(sourceId, "s1", "One", starred = null),
                     song(sourceId, "s2", "Two", starred = null).copy(albumId = "al-2"),
+                    song(sourceId, "s3", "Three", starred = null).copy(albumId = "al-3", artistId = "ar-2"),
                 ),
             )
             dao.upsertPlaylists(
@@ -337,7 +344,7 @@ class LibraryDaoTest {
                 listOf("pl-1"),
                 dao.playlistsByName(sourceId, "", downloadedFilter = 1).page().map { it.id },
             )
-            assertEquals(2, dao.albumsByName(sourceId, 0, "").page().size)
+            assertEquals(3, dao.albumsByName(sourceId, 0, "").page().size)
         }
 
     private suspend fun plan(sql: String): String =

app/src/test/kotlin/com/subtracks/data/prefs/UserPreferencesTest.kt

@@ -1,6 +1,8 @@
 package com.subtracks.data.prefs
 
 import androidx.datastore.preferences.core.PreferenceDataStoreFactory
+import androidx.datastore.preferences.core.edit
+import androidx.datastore.preferences.core.stringPreferencesKey
 import androidx.test.ext.junit.runners.AndroidJUnit4
 import com.subtracks.data.net.NetworkMode
 import com.subtracks.data.source.DEFAULT_FETCH_CONCURRENCY
@@ -25,11 +27,14 @@ class UserPreferencesTest {
             assertEquals(ListQuery("Name"), prefs.listQuery(LibraryListTab.Albums).first())
             assertEquals(ListQuery("Name"), prefs.listQuery(LibraryListTab.Artists).first())
 
-            prefs.setListQuery(LibraryListTab.Albums, ListQuery("Added", descending = true, starred = StarredFilter.Starred))
+            prefs.setListQuery(
+                LibraryListTab.Albums,
+                ListQuery("Added", descending = true, starred = StarredFilter.Starred, downloaded = true),
+            )
             prefs.setListQuery(LibraryListTab.Artists, ListQuery("AlbumCount", descending = true))
 
             assertEquals(
-                ListQuery("Added", descending = true, starred = StarredFilter.Starred),
+                ListQuery("Added", descending = true, starred = StarredFilter.Starred, downloaded = true),
                 prefs.listQuery(LibraryListTab.Albums).first(),
             )
             assertEquals(
@@ -40,6 +45,23 @@ class UserPreferencesTest {
             file.delete()
         }
 
+    @Test
+    fun aListQueryStoredBeforeTheDownloadedFilterStillDecodes() =
+        runTest {
+            val file = File.createTempFile("user-prefs", ".preferences_pb").apply { delete() }
+            val store = PreferenceDataStoreFactory.create(scope = backgroundScope) { file }
+            val prefs = UserPreferences(store)
+
+            store.edit { it[stringPreferencesKey("list_query_albums")] = "Added|1|1" }
+
+            assertEquals(
+                ListQuery("Added", descending = true, starred = StarredFilter.Starred),
+                prefs.listQuery(LibraryListTab.Albums).first(),
+            )
+
+            file.delete()
+        }
+
     @Test
     fun streamQualityRoundTripsPerNetworkMode() =
         runTest {

app/src/test/kotlin/com/subtracks/data/repo/DownloadRepositoryTest.kt

@@ -852,7 +852,40 @@ class DownloadRepositoryTest {
         }
 
     @Test
-    fun activeDownloadsCoverEverySourceNotJustTheActiveOne() =
+    fun activeDownloadsCoverEngineWorkInAnySourceButNotStalledWaitingRows() =
+        runTest {
+            seedLibrary()
+            runBlocking {
+                db.sourcesDao().upsertSource(Source(2, "other", "http://other/", isActive = false, createdAt = 0))
+                val other =
+                    Song(
+                        sourceId = 2,
+                        id = "o1",
+                        albumId = "al2",
+                        artistId = "ar2",
+                        title = "Other",
+                        album = "Other",
+                        artist = "Other",
+                        duration = 100,
+                        track = 1,
+                        disc = 1,
+                        starred = null,
+                        genre = null,
+                    )
+                db.libraryDao().upsertSongs(listOf(other, other.copy(id = "o2", track = 2)))
+                db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Running, engineId = 1))
+                db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed))
+                db.downloadDao().upsert(SongDownload(2, "o1", DownloadStatus.Running, engineId = 2))
+                db.downloadDao().upsert(SongDownload(2, "o2", DownloadStatus.Queued))
+            }
+
+            val rows = runBlocking { repository.activeDownloads().first() }
+
+            assertEquals(setOf(1L to "s1", 2L to "o1"), rows.map { it.sourceId to it.songId }.toSet())
+        }
+
+    @Test
+    fun aDownloadInASourceThatIsNotActiveIsStillReconciled() =
         runTest {
             seedLibrary()
             runBlocking {
@@ -875,14 +908,16 @@ class DownloadRepositoryTest {
                         ),
                     ),
                 )
-                db.downloadDao().upsert(SongDownload(1, "s1", DownloadStatus.Running, engineId = 1))
-                db.downloadDao().upsert(SongDownload(1, "s2", DownloadStatus.Completed))
-                db.downloadDao().upsert(SongDownload(2, "o1", DownloadStatus.Queued))
+                db.downloadDao().upsert(SongDownload(2, "o1", DownloadStatus.Queued, engineId = 42))
             }
+            engine.running(42, bytes = 1, total = 2)
+            repository.start()
 
-            val rows = runBlocking { repository.activeDownloads().first() }
+            await { row(2, "o1")?.status == DownloadStatus.Running }
 
-            assertEquals(setOf(1L to "s1", 2L to "o1"), rows.map { it.sourceId to it.songId }.toSet())
+            writeFile(2, "o1", "audio")
+            engine.complete(42, total = 2)
+            await { row(2, "o1")?.status == DownloadStatus.Completed }
         }
 
     private fun requestedSongIds(): List<String> = engine.requests.map { it.second.path.substringAfterLast('/') }