feat(android): MutationQueue PLAY_OFFLINE kind

Extends the offline mutation queue with the play_offline kind so the
upcoming PlayEventsReporter can durably capture plays that complete
without a successful live play_started, or whose live ended/skipped
close failed. Replay re-fires POST /api/events with type=play_offline.

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

android/app/src/main/java/com/fabledsword/minstrel/cache/mutations/MutationQueue.kt

@@ -19,6 +19,7 @@ object MutationKind {
     const val QUARANTINE_UNFLAG: String = "quarantine_unflag"
     const val PLAYLIST_APPEND: String = "playlist_append"
     const val QUARANTINE_FLAG: String = "quarantine_flag"
+    const val PLAY_OFFLINE: String = "play_offline"
 }
 
 /**
@@ -125,6 +126,13 @@ class MutationQueue @Inject constructor(
             ),
         ),
     )
+
+    suspend fun enqueuePlayOffline(payload: PlayOfflinePayload): Long = dao.insert(
+        CachedMutationEntity(
+            kind = MutationKind.PLAY_OFFLINE,
+            payload = json.encodeToString(PlayOfflinePayload.serializer(), payload),
+        ),
+    )
 }
 
 /**
@@ -159,3 +167,19 @@ data class QuarantineFlagPayload(
     val reason: String,
     val notes: String,
 )
+
+/**
+ * Persisted payload for `MutationKind.PLAY_OFFLINE`. The replayer
+ * re-fires `POST /api/events` with type=play_offline; the server
+ * records start+end from `atIso` (the original play-start time)
+ * + `durationPlayedMs`, applies the canonical skip rule, and
+ * advances #415 rotation when source is a system playlist.
+ */
+@Serializable
+data class PlayOfflinePayload(
+    val trackId: String,
+    val clientId: String,
+    val atIso: String,
+    val durationPlayedMs: Long,
+    val source: String? = null,
+)

android/app/src/main/java/com/fabledsword/minstrel/cache/mutations/MutationReplayer.kt

@@ -2,10 +2,12 @@ package com.fabledsword.minstrel.cache.mutations
 
 import com.fabledsword.minstrel.api.endpoints.AppendTracksRequest
 import com.fabledsword.minstrel.api.endpoints.DiscoverApi
+import com.fabledsword.minstrel.api.endpoints.EventsApi
 import com.fabledsword.minstrel.api.endpoints.FlagRequest
 import com.fabledsword.minstrel.api.endpoints.LikesApi
 import com.fabledsword.minstrel.api.endpoints.PlaylistsApi
 import com.fabledsword.minstrel.api.endpoints.QuarantineApi
+import com.fabledsword.minstrel.models.wire.PlayOfflineRequest
 import com.fabledsword.minstrel.auth.AuthStore
 import com.fabledsword.minstrel.cache.db.dao.CachedMutationDao
 import com.fabledsword.minstrel.cache.db.entities.CachedMutationEntity
@@ -61,6 +63,7 @@ class MutationReplayer @Inject constructor(
     private val discoverApi: DiscoverApi = retrofit.create()
     private val quarantineApi: QuarantineApi = retrofit.create()
     private val playlistsApi: PlaylistsApi = retrofit.create()
+    private val eventsApi: EventsApi = retrofit.create()
 
     private val mutex = Mutex()
 
@@ -109,6 +112,7 @@ class MutationReplayer @Inject constructor(
         MutationKind.QUARANTINE_UNFLAG -> dispatchQuarantineUnflag(row.payload)
         MutationKind.PLAYLIST_APPEND -> dispatchPlaylistAppend(row.payload)
         MutationKind.QUARANTINE_FLAG -> dispatchQuarantineFlag(row.payload)
+        MutationKind.PLAY_OFFLINE -> dispatchPlayOffline(row.payload)
         else -> {
             // Unknown kind — drop the row by claiming success so a
             // stale schema entry can't wedge the queue forever.
@@ -217,4 +221,25 @@ class MutationReplayer @Inject constructor(
             false
         }
     }
+
+    private suspend fun dispatchPlayOffline(payload: String): Boolean {
+        val decoded = json.decodeFromString(PlayOfflinePayload.serializer(), payload)
+        return try {
+            eventsApi.playOffline(
+                PlayOfflineRequest(
+                    trackId = decoded.trackId,
+                    clientId = decoded.clientId,
+                    at = decoded.atIso,
+                    durationPlayedMs = decoded.durationPlayedMs,
+                    source = decoded.source,
+                ),
+            )
+            true
+        } catch (
+            @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable,
+        ) {
+            // Intentional swallow: row stays queued; next drain pass retries.
+            false
+        }
+    }
 }