fix: harden offline/playback recording (contract-audit follow-ups)

Audit of every Android↔server connection point (2026-06-11) cleared the
silent-contract class that caused the events `type` bug, but surfaced a
cluster of offline/playback-robustness defects. Fixes:

1. Playback double-count on background. PlayEventsReporter no longer
   enqueues a partial play_offline + leaves the live row open on every
   screen-lock. closeCurrent() now routes by whether the server has an
   open row: close-by-id (durable PLAY_ENDED on failure) when it does,
   offline only when no row exists, and a no-op while a play_started is
   in flight (the server auto-closes that orphan). onStop only durably
   closes a *paused* play — a still-playing one is left to the live path
   under the foreground service. Adds the PLAY_ENDED mutation kind.

2. Replayer poison rows. MutationReplayer now classifies each replay as
   SENT / DROP / RETRY: permanent 4xx (and corrupt payloads) are dropped
   instead of retried forever; 408/429/5xx/transport still retry.

3. Offline-play / close-by-id idempotency (server). RecordOfflinePlay
   dedups on (user, track, started_at); RecordPlayEnded skips a second
   skip_events insert when re-closing an already-ended row. Makes the
   at-least-once replay safe against lost-response duplicates.

4. Like-toggle collapse. Replayer drops like-toggles superseded by a
   later toggle for the same entity, so partial-failure + differential
   retry can't invert the final like state.

5. Connectivity-return trigger. MutationReplayer + SyncController now
   also drain/sync when NetworkStatusController recovers to Healthy, so
   an offline→online transition mid-session doesn't wait for a cold
   start. SyncController.syncSafe gains a single-in-flight mutex.

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

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

@@ -28,6 +28,13 @@ object MutationKind {
     const val PLAY_OFFLINE: String = "play_offline"
     const val REQUEST_CANCEL: String = "request_cancel"
     const val PLAYBACK_ERROR_REPORT: String = "playback_error_report"
+
+    // Durable close-by-id for a play whose server row is already open
+    // (play_started succeeded). Distinct from PLAY_OFFLINE, which records
+    // a *whole* play when no server row exists. Using close-by-id on
+    // background avoids the duplicate + orphan row the old offline-on-stop
+    // path produced (see 2026-06-11 contract audit).
+    const val PLAY_ENDED: String = "play_ended"
 }
 
 /**
@@ -167,6 +174,17 @@ class MutationQueue @Inject constructor(
         ),
     )
 
+    /**
+     * Durable close-by-id for an already-open server play row. Background
+     * path only (no user hint) — same as [enqueuePlayOffline].
+     */
+    suspend fun enqueuePlayEnded(payload: PlayEndedPayload): Long = dao.insert(
+        CachedMutationEntity(
+            kind = MutationKind.PLAY_ENDED,
+            payload = json.encodeToString(PlayEndedPayload.serializer(), payload),
+        ),
+    )
+
     private suspend fun insertUserDriven(kind: String, payload: String): Long {
         val id = dao.insert(CachedMutationEntity(kind = kind, payload = payload))
         _userEnqueueHints.tryEmit(QUEUED_MESSAGE)
@@ -223,6 +241,19 @@ data class PlayOfflinePayload(
     val source: String? = null,
 )
 
+/**
+ * Persisted payload for `MutationKind.PLAY_ENDED`. The replayer re-fires
+ * `POST /api/events` with type=play_ended, closing the already-open server
+ * row [playEventId] at [durationPlayedMs]. Server-side close-by-id is
+ * idempotent (re-ending updates the same row and skips a duplicate
+ * skip_events insert), so a lost-response retry can't duplicate history.
+ */
+@Serializable
+data class PlayEndedPayload(
+    val playEventId: String,
+    val durationPlayedMs: Long,
+)
+
 /**
  * Persisted payload for `MutationKind.REQUEST_CANCEL` — the
  * `POST /api/requests/{id}/cancel` call lost during a connectivity

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

@@ -1,3 +1,5 @@
+@file:Suppress("TooManyFunctions") // one dispatcher per mutation kind + replay helpers
+
 package com.fabledsword.minstrel.cache.mutations
 
 import com.fabledsword.minstrel.api.endpoints.AppendTracksRequest
@@ -10,6 +12,9 @@ import com.fabledsword.minstrel.api.endpoints.PlaybackErrorsApi
 import com.fabledsword.minstrel.api.endpoints.PlaylistsApi
 import com.fabledsword.minstrel.api.endpoints.QuarantineApi
 import com.fabledsword.minstrel.api.endpoints.RequestsApi
+import com.fabledsword.minstrel.connectivity.NetworkStatusController
+import com.fabledsword.minstrel.connectivity.ServerHealth
+import com.fabledsword.minstrel.models.wire.PlayEndedRequest
 import com.fabledsword.minstrel.models.wire.PlayOfflineRequest
 import com.fabledsword.minstrel.auth.AuthStore
 import com.fabledsword.minstrel.cache.db.dao.CachedMutationDao
@@ -19,12 +24,16 @@ import com.fabledsword.minstrel.likes.data.LikesRepository
 import com.fabledsword.minstrel.models.wire.CreateRequestBody
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.filterNotNull
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.datetime.Clock
+import kotlinx.serialization.SerializationException
 import kotlinx.serialization.json.Json
+import retrofit2.HttpException
 import retrofit2.Retrofit
 import retrofit2.create
 import javax.inject.Inject
@@ -34,25 +43,24 @@ import javax.inject.Singleton
  * Drains the offline MutationQueue. Reads pending rows in FIFO order,
  * re-fires each call against the raw Retrofit API (NOT the high-level
  * Repository, which would re-enqueue on failure → infinite loop), and
- * deletes the row on 2xx success.
+ * resolves each to an [Outcome]:
+ *  - SENT  — server accepted (2xx) → delete the row.
+ *  - DROP  — permanently un-sendable (4xx, unknown kind/variant, corrupt
+ *            payload) → delete the row so it can't wedge the queue forever.
+ *  - RETRY — transient (transport, 5xx, 408, 429) → leave queued, bump
+ *            attempts, retry on the next trigger.
  *
  * Triggered by:
- *   - app start (MinstrelApplication @Inject forces the singleton's
- *     init block to subscribe)
- *   - every time AuthStore.sessionCookie transitions to a non-null
- *     value (sign-in, app launch with persisted cookie)
+ *   - app start (MinstrelApplication @Inject forces the singleton's init)
+ *   - AuthStore.sessionCookie transitioning to non-null (sign-in / cold
+ *     start with a persisted cookie)
+ *   - NetworkStatusController recovering to Healthy (server reachable again
+ *     mid-session — covers offline→online without needing a cold start)
  *
- * Single in-flight via the `mutex` — back-to-back triggers collapse
- * into one drain pass. A proper connectivity-listener / WorkManager
- * job that re-tries on Wi-Fi-came-back is a follow-up; for MVP the
- * sign-in + app-open triggers cover the common offline → online
- * transition (operator unlocks phone, opens app).
- *
- * Failed rows stay in the queue with their `attempts` and
- * `lastAttemptAt` updated. They get another chance on the next
- * trigger; no exponential backoff or max-attempts yet — the rows are
- * small (low single-digit count for typical use) so retry-forever is
- * cheap.
+ * Single in-flight via the [mutex]; back-to-back triggers collapse into one
+ * drain pass. No exponential backoff / max-attempts: the rows are few for
+ * typical use, and permanent failures are now dropped rather than retried
+ * forever.
  */
 @Singleton
 class MutationReplayer @Inject constructor(
@@ -60,6 +68,7 @@ class MutationReplayer @Inject constructor(
     private val authStore: AuthStore,
     private val json: Json,
     @ApplicationScope private val scope: CoroutineScope,
+    networkStatus: NetworkStatusController,
     retrofit: Retrofit,
 ) {
     private val likesApi: LikesApi = retrofit.create()
@@ -72,6 +81,8 @@ class MutationReplayer @Inject constructor(
 
     private val mutex = Mutex()
 
+    private enum class Outcome { SENT, DROP, RETRY }
+
     init {
         scope.launch {
             authStore.sessionCookie
@@ -79,6 +90,16 @@ class MutationReplayer @Inject constructor(
                 .distinctUntilChanged()
                 .collect { drainSafe() }
         }
+        // Server-reachable-again: replay queued writes the moment the
+        // health signal recovers, so an offline→online transition mid-
+        // session doesn't wait for the next cold start / re-auth.
+        scope.launch {
+            networkStatus.state
+                .map { it == ServerHealth.Healthy }
+                .distinctUntilChanged()
+                .filter { it }
+                .collect { drainSafe() }
+        }
     }
 
     /** Public entry-point; coalesces concurrent triggers via a Mutex. */
@@ -93,193 +114,182 @@ class MutationReplayer @Inject constructor(
 
     private suspend fun drain() {
         val rows = dao.getAll()
+        // Collapse superseded like-toggles: only the latest desired state per
+        // (entity) is replayed; older toggles for the same entity are dropped
+        // unsent. Without this, partial-failure + differential retry could
+        // replay an older toggle last and invert the final like state.
+        val superseded = supersededLikeToggleIds(rows)
         for (row in rows) {
-            val sent = runCatching { dispatch(row) }.getOrDefault(false)
-            if (sent) {
+            if (row.id in superseded) {
                 dao.delete(row.id)
-            } else {
-                dao.recordAttempt(row.id, Clock.System.now())
+                continue
+            }
+            when (outcomeFor(row)) {
+                Outcome.SENT, Outcome.DROP -> dao.delete(row.id)
+                Outcome.RETRY -> dao.recordAttempt(row.id, Clock.System.now())
             }
         }
     }
 
+    /** Row ids of like-toggles superseded by a later toggle for the same entity. */
+    private fun supersededLikeToggleIds(rows: List<CachedMutationEntity>): Set<Long> {
+        val latestByEntity = HashMap<String, Long>()
+        val superseded = HashSet<Long>()
+        for (row in rows) {
+            if (row.kind != MutationKind.LIKE_TOGGLE) continue
+            val decoded = runCatching {
+                json.decodeFromString(LikeTogglePayload.serializer(), row.payload)
+            }.getOrNull() ?: continue
+            val key = "${decoded.entityType}:${decoded.entityId}"
+            // `rows` is ascending by id, so a prior entry is always older.
+            latestByEntity.put(key, row.id)?.let(superseded::add)
+        }
+        return superseded
+    }
+
+    private suspend fun outcomeFor(row: CachedMutationEntity): Outcome = try {
+        dispatch(row)
+    } catch (e: HttpException) {
+        if (isPermanent(e.code())) Outcome.DROP else Outcome.RETRY
+    } catch (@Suppress("SwallowedException") e: SerializationException) {
+        // Corrupt / schema-incompatible payload — permanent, drop it.
+        Outcome.DROP
+    } catch (@Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable) {
+        // Transport / unknown — leave queued for the next pass.
+        Outcome.RETRY
+    }
+
+    /** 4xx are permanent except the two "retry me" statuses. */
+    private fun isPermanent(code: Int): Boolean =
+        code in HTTP_CLIENT_ERR_MIN..HTTP_CLIENT_ERR_MAX &&
+            code != HTTP_TIMEOUT && code != HTTP_TOO_MANY
+
     /**
-     * Returns true if the server accepted the call, false on any
-     * failure (transport, server error, decode). The replayer treats
-     * false as "leave the row for next pass" — it doesn't try to
-     * distinguish permanent rejections from transient ones; a row
-     * that consistently fails will accumulate attempt count and
-     * lastAttemptAt for diagnostics.
+     * Returns the outcome of replaying one row. SENT on a 2xx; DROP for an
+     * unknown kind/variant. Network / HTTP failures THROW and are classified
+     * by [outcomeFor] — dispatchers no longer swallow.
      */
-    private suspend fun dispatch(row: CachedMutationEntity): Boolean = when (row.kind) {
+    private suspend fun dispatch(row: CachedMutationEntity): Outcome = when (row.kind) {
         MutationKind.LIKE_TOGGLE -> dispatchLikeToggle(row.payload)
         MutationKind.REQUEST_CREATE -> dispatchRequestCreate(row.payload)
         MutationKind.QUARANTINE_UNFLAG -> dispatchQuarantineUnflag(row.payload)
         MutationKind.PLAYLIST_APPEND -> dispatchPlaylistAppend(row.payload)
         MutationKind.QUARANTINE_FLAG -> dispatchQuarantineFlag(row.payload)
         MutationKind.PLAY_OFFLINE -> dispatchPlayOffline(row.payload)
+        MutationKind.PLAY_ENDED -> dispatchPlayEnded(row.payload)
         MutationKind.REQUEST_CANCEL -> dispatchRequestCancel(row.payload)
         MutationKind.PLAYBACK_ERROR_REPORT -> dispatchPlaybackErrorReport(row.payload)
-        else -> {
-            // Unknown kind — drop the row by claiming success so a
-            // stale schema entry can't wedge the queue forever.
-            true
-        }
+        // Unknown kind — drop so a stale schema entry can't wedge the queue.
+        else -> Outcome.DROP
     }
 
-    private suspend fun dispatchLikeToggle(payload: String): Boolean {
+    private suspend fun dispatchLikeToggle(payload: String): Outcome {
         val decoded = json.decodeFromString(LikeTogglePayload.serializer(), payload)
         val kindPath = when (decoded.entityType) {
             LikesRepository.ENTITY_ARTIST -> "artists"
             LikesRepository.ENTITY_ALBUM -> "albums"
             LikesRepository.ENTITY_TRACK -> "tracks"
-            else -> return true  // unknown variant — drop.
+            else -> return Outcome.DROP
         }
-        return try {
-            if (decoded.desiredState) {
-                likesApi.like(kindPath, decoded.entityId)
-            } else {
-                likesApi.unlike(kindPath, decoded.entityId)
-            }
-            true
-        } catch (
-            @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable,
-        ) {
-            // Intentional swallow: the row stays in the queue with
-            // attempts + lastAttemptAt incremented; next drain pass
-            // retries. Per-attempt diagnostic logging lands when we
-            // wire up Timber at the replayer level.
-            false
+        if (decoded.desiredState) {
+            likesApi.like(kindPath, decoded.entityId)
+        } else {
+            likesApi.unlike(kindPath, decoded.entityId)
         }
+        return Outcome.SENT
     }
 
-    private suspend fun dispatchRequestCreate(payload: String): Boolean {
+    private suspend fun dispatchRequestCreate(payload: String): Outcome {
         val decoded = json.decodeFromString(RequestCreatePayload.serializer(), payload)
-        val body = CreateRequestBody(
-            kind = decoded.kind,
-            artistMbid = decoded.artistMbid,
-            artistName = decoded.artistName,
-            albumMbid = decoded.albumMbid,
-            albumTitle = decoded.albumTitle,
-            trackMbid = decoded.trackMbid,
-            trackTitle = decoded.trackTitle,
+        discoverApi.createRequest(
+            CreateRequestBody(
+                kind = decoded.kind,
+                artistMbid = decoded.artistMbid,
+                artistName = decoded.artistName,
+                albumMbid = decoded.albumMbid,
+                albumTitle = decoded.albumTitle,
+                trackMbid = decoded.trackMbid,
+                trackTitle = decoded.trackTitle,
+            ),
         )
-        return try {
-            discoverApi.createRequest(body)
-            true
-        } catch (
-            @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable,
-        ) {
-            // Intentional swallow: the row stays in the queue with
-            // attempts + lastAttemptAt incremented; next drain pass
-            // retries. Per-attempt diagnostic logging lands when we
-            // wire up Timber at the replayer level.
-            false
-        }
+        return Outcome.SENT
     }
 
-    private suspend fun dispatchQuarantineUnflag(payload: String): Boolean {
+    private suspend fun dispatchQuarantineUnflag(payload: String): Outcome {
         val decoded = json.decodeFromString(QuarantineUnflagPayload.serializer(), payload)
-        return try {
-            quarantineApi.unflag(decoded.trackId)
-            true
-        } catch (
-            @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable,
-        ) {
-            // Intentional swallow: the row stays in the queue with
-            // attempts + lastAttemptAt incremented; next drain pass
-            // retries. Per-attempt diagnostic logging lands when we
-            // wire up Timber at the replayer level.
-            false
-        }
+        quarantineApi.unflag(decoded.trackId)
+        return Outcome.SENT
     }
 
-    private suspend fun dispatchPlaylistAppend(payload: String): Boolean {
+    private suspend fun dispatchPlaylistAppend(payload: String): Outcome {
         val decoded = json.decodeFromString(PlaylistAppendPayload.serializer(), payload)
-        return try {
-            playlistsApi.appendTracks(
-                playlistId = decoded.playlistId,
-                body = AppendTracksRequest(trackIds = decoded.trackIds),
-            )
-            true
-        } catch (
-            @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable,
-        ) {
-            // Intentional swallow: row stays queued; next drain pass retries.
-            false
-        }
+        playlistsApi.appendTracks(
+            playlistId = decoded.playlistId,
+            body = AppendTracksRequest(trackIds = decoded.trackIds),
+        )
+        return Outcome.SENT
     }
 
-    private suspend fun dispatchQuarantineFlag(payload: String): Boolean {
+    private suspend fun dispatchQuarantineFlag(payload: String): Outcome {
         val decoded = json.decodeFromString(QuarantineFlagPayload.serializer(), payload)
-        return try {
-            quarantineApi.flag(
-                FlagRequest(
-                    trackId = decoded.trackId,
-                    reason = decoded.reason,
-                    notes = decoded.notes,
-                ),
-            )
-            true
-        } catch (
-            @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable,
-        ) {
-            // Intentional swallow: row stays queued; next drain pass retries.
-            false
-        }
+        quarantineApi.flag(
+            FlagRequest(
+                trackId = decoded.trackId,
+                reason = decoded.reason,
+                notes = decoded.notes,
+            ),
+        )
+        return Outcome.SENT
     }
 
-    private suspend fun dispatchPlayOffline(payload: String): Boolean {
+    private suspend fun dispatchPlayOffline(payload: String): Outcome {
         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
-        }
+        eventsApi.playOffline(
+            PlayOfflineRequest(
+                trackId = decoded.trackId,
+                clientId = decoded.clientId,
+                at = decoded.atIso,
+                durationPlayedMs = decoded.durationPlayedMs,
+                source = decoded.source,
+            ),
+        )
+        return Outcome.SENT
     }
 
-    private suspend fun dispatchRequestCancel(payload: String): Boolean {
+    private suspend fun dispatchPlayEnded(payload: String): Outcome {
+        val decoded = json.decodeFromString(PlayEndedPayload.serializer(), payload)
+        eventsApi.playEnded(
+            PlayEndedRequest(
+                playEventId = decoded.playEventId,
+                durationPlayedMs = decoded.durationPlayedMs,
+            ),
+        )
+        return Outcome.SENT
+    }
+
+    private suspend fun dispatchRequestCancel(payload: String): Outcome {
         val decoded = json.decodeFromString(RequestCancelPayload.serializer(), payload)
-        return try {
-            requestsApi.cancel(decoded.requestId)
-            true
-        } catch (
-            @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable,
-        ) {
-            // Intentional swallow: row stays queued; next drain pass retries.
-            false
-        }
+        requestsApi.cancel(decoded.requestId)
+        return Outcome.SENT
     }
 
-    private suspend fun dispatchPlaybackErrorReport(payload: String): Boolean {
+    private suspend fun dispatchPlaybackErrorReport(payload: String): Outcome {
         val decoded = json.decodeFromString(PlaybackErrorReportPayload.serializer(), payload)
-        return try {
-            playbackErrorsApi.report(
-                PlaybackErrorReportRequest(
-                    trackId = decoded.trackId,
-                    kind = decoded.kind,
-                    detail = decoded.detail,
-                    clientId = decoded.clientId,
-                ),
-            )
-            true
-        } catch (
-            @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable,
-        ) {
-            // Intentional swallow: row stays queued; next drain pass retries.
-            false
-        }
+        playbackErrorsApi.report(
+            PlaybackErrorReportRequest(
+                trackId = decoded.trackId,
+                kind = decoded.kind,
+                detail = decoded.detail,
+                clientId = decoded.clientId,
+            ),
+        )
+        return Outcome.SENT
+    }
+
+    private companion object {
+        const val HTTP_CLIENT_ERR_MIN = 400
+        const val HTTP_CLIENT_ERR_MAX = 499
+        const val HTTP_TIMEOUT = 408
+        const val HTTP_TOO_MANY = 429
     }
 }

android/app/src/main/java/com/fabledsword/minstrel/cache/sync/SyncController.kt

@@ -2,6 +2,8 @@ package com.fabledsword.minstrel.cache.sync
 
 import com.fabledsword.minstrel.api.endpoints.SyncApi
 import com.fabledsword.minstrel.auth.AuthStore
+import com.fabledsword.minstrel.connectivity.NetworkStatusController
+import com.fabledsword.minstrel.connectivity.ServerHealth
 import com.fabledsword.minstrel.cache.db.dao.CachedAlbumDao
 import com.fabledsword.minstrel.cache.db.dao.CachedArtistDao
 import com.fabledsword.minstrel.cache.db.dao.CachedTrackDao
@@ -16,8 +18,11 @@ import com.fabledsword.minstrel.models.wire.SyncArtistWire
 import com.fabledsword.minstrel.models.wire.SyncTrackWire
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.filterNotNull
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
 import kotlinx.datetime.Clock
 import retrofit2.Retrofit
 import retrofit2.create
@@ -50,10 +55,15 @@ class SyncController @Inject constructor(
     private val trackDao: CachedTrackDao,
     private val authStore: AuthStore,
     @ApplicationScope private val scope: CoroutineScope,
+    networkStatus: NetworkStatusController,
     retrofit: Retrofit,
 ) {
     private val api: SyncApi = retrofit.create()
 
+    // Single in-flight guard: the cookie + health-recovery triggers (and a
+    // pull-to-refresh) can fire concurrently; coalesce them into one pass.
+    private val mutex = Mutex()
+
     init {
         scope.launch {
             authStore.sessionCookie
@@ -61,11 +71,25 @@ class SyncController @Inject constructor(
                 .distinctUntilChanged()
                 .collect { syncSafe() }
         }
+        // Re-sync when the server becomes reachable again mid-session, so a
+        // delta missed while offline lands without waiting for a cold start.
+        scope.launch {
+            networkStatus.state
+                .map { it == ServerHealth.Healthy }
+                .distinctUntilChanged()
+                .filter { it }
+                .collect { syncSafe() }
+        }
     }
 
     /** Public entry-point for "Sync now" affordances. Swallows errors. */
     suspend fun syncSafe() {
-        runCatching { sync() }
+        if (!mutex.tryLock()) return
+        try {
+            runCatching { sync() }
+        } finally {
+            mutex.unlock()
+        }
     }
 
     private suspend fun sync() {

android/app/src/main/java/com/fabledsword/minstrel/player/PlayEventsReporter.kt

@@ -6,6 +6,7 @@ import androidx.lifecycle.ProcessLifecycleOwner
 import com.fabledsword.minstrel.api.endpoints.EventsApi
 import com.fabledsword.minstrel.auth.AuthStore
 import com.fabledsword.minstrel.cache.mutations.MutationQueue
+import com.fabledsword.minstrel.cache.mutations.PlayEndedPayload
 import com.fabledsword.minstrel.cache.mutations.PlayOfflinePayload
 import com.fabledsword.minstrel.di.ApplicationScope
 import com.fabledsword.minstrel.models.wire.PlayEndedRequest
@@ -64,11 +65,24 @@ class PlayEventsReporter @Inject constructor(
     private var curLastPositionMs: Long = 0
     private var curDurationMs: Long = 0
 
+    // Whether the currently-tracked play is actively playing (vs paused).
+    // onStop reads this: a backgrounded play that's still playing is left
+    // to the live path (the foreground service keeps us alive), while a
+    // paused one is durably closed so a kill can't orphan its open row.
+    private var curPlaying: Boolean = false
+
     // Server play_event_id when the live play_started succeeded;
     // null if start failed/offline — then the close is captured into
     // the offline mutation queue instead of a live ended/skipped.
     private var openPlayEventId: String? = null
 
+    // True from beginTrack until startLive's play_started resolves. When a
+    // close races an in-flight start (id still null), the server row is
+    // being created but we've lost its id — we must NOT also enqueue an
+    // offline play (that would duplicate); the server auto-closes the
+    // orphan on the next play_started instead.
+    private var startInFlight: Boolean = false
+
     private var prevTrackId: String? = null
 
     init {
@@ -95,7 +109,7 @@ class PlayEventsReporter @Inject constructor(
     ) {
         // Track-change → close the prior tracked play.
         if (trackId != prevTrackId && curTrackId != null) {
-            closeCurrent(viaOffline = false)
+            closeCurrent()
         }
 
         // Entered playing for a new track → begin tracking it.
@@ -110,6 +124,7 @@ class PlayEventsReporter @Inject constructor(
         if (curTrackId != null && trackId == curTrackId) {
             curLastPositionMs = positionMs
             if (durationMs > 0) curDurationMs = durationMs
+            curPlaying = playing
         }
 
         prevTrackId = trackId
@@ -121,12 +136,18 @@ class PlayEventsReporter @Inject constructor(
         curSource = source
         curLastPositionMs = 0
         curDurationMs = durationMs
+        curPlaying = true
         openPlayEventId = null
+        startInFlight = true
         startLive(trackId = trackId, source = source)
     }
 
     private fun startLive(trackId: String, source: String?) {
-        val cid = resolveClientId() ?: return
+        val cid = resolveClientId()
+        if (cid == null) {
+            startInFlight = false
+            return
+        }
         scope.launch {
             try {
                 val response = api.playStarted(
@@ -144,56 +165,72 @@ class PlayEventsReporter @Inject constructor(
             ) {
                 // Offline / flaky — openPlayEventId stays null; close
                 // path will enqueue the completed play for replay.
+            } finally {
+                startInFlight = false
             }
         }
     }
 
-    private fun closeCurrent(viaOffline: Boolean) {
+    /**
+     * Closes the tracked play. Routes by whether the server already has an
+     * open row, NOT by caller context — this is what keeps background /
+     * track-change / failure paths from each minting a second row:
+     *  - open server row (id) → close it by id; on failure enqueue a durable
+     *    close-by-id (never an offline play).
+     *  - start in flight (no id yet) → leave it; the server auto-closes the
+     *    orphan on the next play_started. Enqueuing offline here would
+     *    duplicate the in-flight row.
+     *  - no server row at all → record the whole play offline.
+     *
+     * Always uses the actual position played; the server's skip rule
+     * (50%/30s defaults) classifies was_skipped from it.
+     */
+    private fun closeCurrent() {
         val trackId = curTrackId
         val startedAt = curStartedAt
         if (trackId == null || startedAt == null) {
             resetCurrent()
             return
         }
-        // Always send the actual position played to; the server's
-        // skip rule (50%/30s defaults, configurable) classifies
-        // was_skipped from this. If the track ran to completion the
-        // last observed position is ~= duration, which gives ratio ~1
-        // and the server records a non-skip. The COMPLETION_TOLERANCE
-        // rounding the prior shape did is no longer needed — the
-        // server's threshold is the source of truth.
         val durationMs = curLastPositionMs
         val source = curSource
         val id = openPlayEventId
-
-        if (!viaOffline && id != null) {
-            scope.launch {
-                try {
-                    api.playEnded(
-                        PlayEndedRequest(
-                            playEventId = id,
-                            durationPlayedMs = durationMs,
-                        ),
-                    )
-                } catch (
-                    @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable,
-                ) {
-                    enqueueOffline(trackId, startedAt, source, durationMs)
-                }
-            }
-        } else {
-            enqueueOffline(trackId, startedAt, source, durationMs)
+        val inFlight = startInFlight
+        when {
+            id != null -> closeById(id, durationMs)
+            inFlight -> Unit
+            else -> enqueueOffline(trackId, startedAt, source, durationMs)
         }
         resetCurrent()
     }
 
+    private fun closeById(id: String, durationMs: Long) {
+        scope.launch {
+            try {
+                api.playEnded(
+                    PlayEndedRequest(playEventId = id, durationPlayedMs = durationMs),
+                )
+            } catch (
+                @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable,
+            ) {
+                // Durable close-by-id (idempotent server-side) — survives a
+                // process kill without orphaning the open row or duplicating.
+                mutationQueue.enqueuePlayEnded(
+                    PlayEndedPayload(playEventId = id, durationPlayedMs = durationMs),
+                )
+            }
+        }
+    }
+
     private fun resetCurrent() {
         curTrackId = null
         curStartedAt = null
         curSource = null
         curLastPositionMs = 0
         curDurationMs = 0
+        curPlaying = false
         openPlayEventId = null
+        startInFlight = false
     }
 
     private fun enqueueOffline(
@@ -232,8 +269,17 @@ class PlayEventsReporter @Inject constructor(
     // ── Lifecycle: durable-close on app background / detach ────────
 
     override fun onStop(owner: LifecycleOwner) {
-        if (curTrackId != null) {
-            closeCurrent(viaOffline = true)
+        // App backgrounded. If the tracked play is still PLAYING, leave it
+        // alone: the foreground media service keeps the process — and this
+        // app-scope uiState collector — alive, so the live state machine
+        // closes the play accurately on the next real track change. Closing
+        // here (the old always-offline behavior) raced the collector's
+        // re-begin and minted a duplicate, skip-misclassified row on every
+        // screen-lock. Only durably close a PAUSED play, where no further
+        // progress will happen and a process kill would otherwise orphan the
+        // open server row. See docs/superpowers/2026-06-11 contract audit.
+        if (curTrackId != null && !curPlaying) {
+            closeCurrent()
         }
     }
 }

internal/playevents/writer.go

@@ -180,6 +180,12 @@ func (w *Writer) RecordPlayEnded(
 		if err != nil {
 			return err
 		}
+		// Idempotency for the client's durable close-by-id replay: if the
+		// row is already closed, re-applying the close is harmless (the
+		// UPDATE is last-write-wins) but a second skip_events insert must
+		// NOT fire — a lost-response retry would otherwise double the
+		// skip signal. See the 2026-06-11 contract audit.
+		alreadyEnded := ev.EndedAt.Valid
 		ratio := 0.0
 		if track.DurationMs > 0 {
 			ratio = float64(durationPlayedMs) / float64(track.DurationMs)
@@ -196,7 +202,7 @@ func (w *Writer) RecordPlayEnded(
 		}); err != nil {
 			return err
 		}
-		if isSkip {
+		if isSkip && !alreadyEnded {
 			if _, err := q.InsertSkipEvent(ctx, dbq.InsertSkipEventParams{
 				UserID:     ev.UserID,
 				TrackID:    ev.TrackID,
@@ -356,6 +362,24 @@ func (w *Writer) RecordOfflinePlay(
 		if err != nil {
 			return err
 		}
+		// Idempotency: the offline queue is at-least-once, so a response lost
+		// after commit makes the client re-fire the same play. The replayed
+		// payload carries the original `at`, so an exact (user, track,
+		// started_at) match means we already recorded it — no-op instead of a
+		// duplicate history row. See the 2026-06-11 contract audit.
+		var dupID pgtype.UUID
+		err = tx.QueryRow(ctx,
+			`SELECT id FROM play_events
+			 WHERE user_id = $1 AND track_id = $2 AND started_at = $3
+			 LIMIT 1`,
+			userID, trackID, pgtype.Timestamptz{Time: at, Valid: true},
+		).Scan(&dupID)
+		if err == nil {
+			return nil
+		}
+		if !errors.Is(err, pgx.ErrNoRows) {
+			return err
+		}
 		if err := w.autoClosePriorOpen(ctx, q, userID, at); err != nil {
 			return err
 		}

internal/playevents/writer_test.go

@@ -219,6 +219,55 @@ func TestRecordPlayEnded_DurationOver50Percent_NotSkip(t *testing.T) {
 	}
 }
 
+// A lost-response retry of an offline play (same user/track/started_at)
+// must not create a second history row. Guards the at-least-once replay
+// path added in the 2026-06-11 audit.
+func TestRecordOfflinePlay_DedupsByUserTrackStartedAt(t *testing.T) {
+	f := newFixture(t, 200_000)
+	ctx := context.Background()
+	at := time.Now().UTC()
+	if err := f.w.RecordOfflinePlay(ctx, f.user, f.track, "c", "", at, 120_000); err != nil {
+		t.Fatalf("first offline: %v", err)
+	}
+	if err := f.w.RecordOfflinePlay(ctx, f.user, f.track, "c", "", at, 120_000); err != nil {
+		t.Fatalf("replay offline: %v", err)
+	}
+	var count int
+	if err := f.pool.QueryRow(ctx,
+		"SELECT COUNT(*) FROM play_events WHERE user_id=$1 AND track_id=$2 AND started_at=$3",
+		f.user, f.track, pgtype.Timestamptz{Time: at, Valid: true}).Scan(&count); err != nil {
+		t.Fatalf("count: %v", err)
+	}
+	if count != 1 {
+		t.Errorf("play_events count = %d, want 1 (deduped)", count)
+	}
+}
+
+// Re-closing an already-ended row (the client's durable close-by-id replay)
+// updates the row but must NOT insert a second skip_events row.
+func TestRecordPlayEnded_IdempotentClose_NoDuplicateSkipEvent(t *testing.T) {
+	f := newFixture(t, 200_000)
+	ctx := context.Background()
+	t1 := time.Now().UTC()
+	res, _ := f.w.RecordPlayStarted(ctx, f.user, f.track, "c", t1)
+	t2 := t1.Add(10 * time.Second) // 5% / 10s -> skip
+	if err := f.w.RecordPlayEnded(ctx, res.PlayEventID, 10_000, t2); err != nil {
+		t.Fatalf("first end: %v", err)
+	}
+	if err := f.w.RecordPlayEnded(ctx, res.PlayEventID, 10_000, t2); err != nil {
+		t.Fatalf("replay end: %v", err)
+	}
+	var count int
+	if err := f.pool.QueryRow(ctx,
+		"SELECT COUNT(*) FROM skip_events WHERE user_id=$1 AND track_id=$2",
+		f.user, f.track).Scan(&count); err != nil {
+		t.Fatalf("count: %v", err)
+	}
+	if count != 1 {
+		t.Errorf("skip_events count = %d, want 1 (idempotent close)", count)
+	}
+}
+
 func TestRecordPlaySkipped_AlwaysSkipFlagAndSkipEventRow(t *testing.T) {
 	f := newFixture(t, 200_000)
 	t1 := time.Now().UTC()