fix(crypto): warm apps adopt rotations via a couple-doc watcher; streak RMW → transaction

Landmine audit after C-ROTATE-001 (user: "any landmines that need fixing?"),
hunting the class that has now bitten twice — stale reads feeding one-shot
decisions. Two found, both fixed.

1. A WARM app never adopted a partner's rotation at all. Adoption ran at process
   start (MainActivity, once per uid) and on Home's initial load — and nowhere
   else. The 🔑 push routes to Security, which doesn't adopt; the couple listener
   in HomeViewModel watches the USER doc (pairing state), not the couple doc. So
   an app sitting open on any screen when the partner rotates renders 🔒 for
   every new-key message until process death — and a foregrounded process can
   stay warm for days. Fix: the couple doc is now observable
   (FirestoreCoupleDataSource.observeCouple → CoupleRepository.observeCoupleForUser,
   composed observeUser → coupleId → flatMapLatest so pairing changes reroute
   the stream), and CoupleKeyRotationAdopter.watch(uid) adopts on every
   emission. MainActivity's hook is now that long-lived watch (collectLatest on
   the auth uid: sign-out cancels it cleanly). Snapshot listeners deliver the
   partner's write from the server within seconds — no cache trap, no push
   required, warm or cold. Adoption stays idempotent, a failed adoption is
   recorded and the watch keeps collecting, and a stream error is recorded and
   ends the watch quietly (degrades to the one-shot hooks, never crashes).

2. updateStreak was the same disease verbatim: cache-first get() → compute next
   streak → merge write. A stale snapshot double-counts or clobbers the streak
   when both partners answer near-simultaneously. It now runs as a Firestore
   transaction — server-read values, retried on contention, concurrent answers
   converge. Same StreakCalculator semantics, same skip-if-unchanged.

Audited and deliberately left alone: updateDisplayName/updateSex read coupleId
via a cache-first snapshot, but a stale/missing coupleId only means the field is
stored plaintext until migrateProfileFields heals it on key-unlock — self-
healing already exists, so no change. requestRestore is stale-cache-immune
(delete-then-set on a fixed doc id). The backup manifest already does CAS.

CoupleKeyRotationAdopterTest pins the watcher contract (adopt per emission, skip
nulls, keep collecting past failures, stream errors recorded not thrown);
mutation-checked — dropping the recording fails exactly those two tests. Full
suite green, assembleDebug clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app/src/main/java/app/closer/MainActivity.kt

@@ -43,6 +43,7 @@ import app.closer.core.notifications.TokenRegistrar
 import app.closer.domain.model.AuthState
 import app.closer.notifications.PartnerNotificationPayload
 import app.closer.notifications.PartnerNotificationType
+import kotlinx.coroutines.flow.collectLatest
 import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.flow.filterIsInstance
 import kotlinx.coroutines.flow.first
@@ -292,11 +293,11 @@ class MainActivity : AppCompatActivity() {
     private fun adoptCoupleKeyRotation() {
         lifecycleScope.launch {
             authRepository.authState
-                .filterIsInstance<AuthState.Authenticated>()
+                .map { (it as? AuthState.Authenticated)?.userId }
-                .map { it.userId }
                 .distinctUntilChanged()
-                .collect { uid ->
+                .collectLatest { uid ->
-                    runCatching { coupleKeyRotationAdopter.adoptIfNeeded(uid) }
+                    // collectLatest: sign-out (null) or a different uid cancels the previous watch.
+                    if (uid != null) runCatching { coupleKeyRotationAdopter.watch(uid) }
                 }
         }
     }

app/src/main/java/app/closer/data/remote/FirestoreCoupleDataSource.kt

@@ -8,6 +8,9 @@ import com.google.firebase.firestore.DocumentSnapshot
 import com.google.firebase.firestore.FirebaseFirestore
 import com.google.firebase.firestore.SetOptions
 import com.google.firebase.functions.FirebaseFunctions
+import kotlinx.coroutines.channels.awaitClose
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.callbackFlow
 import kotlinx.coroutines.tasks.await
 import javax.inject.Inject
 import javax.inject.Singleton
@@ -112,8 +115,30 @@ class FirestoreCoupleDataSource @Inject constructor(
         return if (snap.exists()) snap.toCouple() else null
     }
 
+    /**
+     * The couple doc as a live stream. Snapshot listeners receive the partner's writes from the
+     * server within seconds, which makes this the right feed for reacting to `keyGeneration` bumps —
+     * unlike the one-shot reads, there is no cache-staleness trap to fall into (C-ROTATE-001).
+     */
+    fun observeCouple(coupleId: String): Flow<Couple?> = callbackFlow {
+        val listener = coupleRef(coupleId).addSnapshotListener { snap, error ->
+            if (error != null) {
+                close(error)
+                return@addSnapshotListener
+            }
+            trySend(snap?.takeIf { it.exists() }?.toCouple())
+        }
+        awaitClose { listener.remove() }
+    }
+
     suspend fun updateStreak(coupleId: String) {
-        val snap = coupleRef(coupleId).get().await()
+        // Read-modify-write on a counter, so it runs as a TRANSACTION: the old cache-first
+        // get()-compute-set() could derive `next` from a stale snapshot and double-count or clobber
+        // the streak when both partners answered near-simultaneously — the same stale-read-feeding-a-
+        // write class as C-ROTATE-001. A transaction reads current values from the server and retries
+        // on contention, so concurrent answers converge instead of racing.
+        db.runTransaction { txn ->
+            val snap = txn.get(coupleRef(coupleId))
             val lastAnsweredAt = snap.millisOrNull("lastAnsweredAt") ?: 0L
             val now = System.currentTimeMillis()
             val current = (snap.getLong("streakCount") ?: 0L).toInt()
@@ -125,11 +150,14 @@ class FirestoreCoupleDataSource @Inject constructor(
                 currentStreak = current
             )
             // Same-day answer → nothing to change; skip the redundant write.
-        if (!next.changed) return
+            if (next.changed) {
-        coupleRef(coupleId).set(
+                txn.set(
+                    coupleRef(coupleId),
                     mapOf("streakCount" to next.newStreak, "lastAnsweredAt" to now),
                     SetOptions.merge()
-        ).await()
+                )
+            }
+        }.await()
     }
 
     suspend fun leaveCouple() {

app/src/main/java/app/closer/data/repository/CoupleRepositoryImpl.kt

@@ -1,6 +1,10 @@
 package app.closer.data.repository
 
 import app.closer.core.crash.CrashReporter
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.flatMapLatest
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.map
 import app.closer.crypto.CoupleEncryptionManager
 import app.closer.data.remote.FirestoreCoupleDataSource
 import app.closer.data.remote.FirestoreUserDataSource
@@ -37,6 +41,15 @@ class CoupleRepositoryImpl @Inject constructor(
         error("Direct couple creation from the client is no longer supported; use InviteRepository.acceptInvite.")
     }
 
+    @kotlinx.coroutines.ExperimentalCoroutinesApi
+    override fun observeCoupleForUser(userId: String): kotlinx.coroutines.flow.Flow<Couple?> =
+        userDataSource.observeUser(userId)
+            .map { it?.coupleId }
+            .distinctUntilChanged()
+            .flatMapLatest { coupleId ->
+                if (coupleId.isNullOrBlank()) flowOf(null) else coupleDataSource.observeCouple(coupleId)
+            }
+
     override suspend fun updateStreak(coupleId: String): Result<Unit> = runCatching {
         coupleDataSource.updateStreak(coupleId)
     }

app/src/main/java/app/closer/domain/repository/CoupleRepository.kt

@@ -4,6 +4,12 @@ import app.closer.domain.model.Couple
 
 interface CoupleRepository {
     suspend fun getCoupleForUser(userId: String): Couple?
+
+    /**
+     * The user's couple as a live stream (null while unpaired). Server-fed via snapshot listener —
+     * the safe feed for decisions that react to couple-doc changes (e.g. keyGeneration adoption).
+     */
+    fun observeCoupleForUser(userId: String): kotlinx.coroutines.flow.Flow<Couple?>
     suspend fun createCouple(inviterUserId: String, acceptorUserId: String, inviteCode: String, recoveryPhrase: String): Result<String>
     suspend fun updateStreak(coupleId: String): Result<Unit>
     suspend fun leaveCouple(userId: String): Result<Unit>

app/src/main/java/app/closer/domain/usecase/CoupleKeyRotationAdopter.kt

@@ -3,6 +3,7 @@ package app.closer.domain.usecase
 import app.closer.core.crash.CrashReporter
 import app.closer.crypto.CoupleEncryptionManager
 import app.closer.domain.repository.CoupleRepository
+import kotlinx.coroutines.flow.catch
 import javax.inject.Inject
 import javax.inject.Singleton
 
@@ -26,6 +27,23 @@ class CoupleKeyRotationAdopter @Inject constructor(
     private val encryptionManager: CoupleEncryptionManager,
     private val crashReporter: CrashReporter
 ) {
+    /**
+     * Long-lived entry point: watches the couple doc and adopts on every emission. This is what
+     * covers the WARM app — the partner's rotation lands via the snapshot listener within seconds,
+     * whereas the one-shot hooks (process start, Home load) only help after the fact. Caught in
+     * review: with one-shot hooks alone, a foregrounded app never adopted at all; the 🔑 push routes
+     * to Security, which doesn't adopt, and a process can stay warm for days showing 🔒.
+     *
+     * Adoption is idempotent (generation compare), so repeated emissions are cheap no-ops. A stream
+     * failure is recorded and ends the watch quietly — the process degrades to the one-shot hooks,
+     * never crashes; the next launch re-establishes it.
+     */
+    suspend fun watch(uid: String) {
+        coupleRepository.observeCoupleForUser(uid)
+            .catch { crashReporter.recordException(it) }
+            .collect { couple -> couple?.let { adoptForCouple(it) } }
+    }
+
     /** Startup entry point: resolves the couple itself. Returns the outcome for callers that care. */
     suspend fun adoptIfNeeded(uid: String): CoupleEncryptionManager.AdoptionResult {
         val couple = runCatching { coupleRepository.getCoupleForUser(uid) }

app/src/test/java/app/closer/domain/usecase/CoupleKeyRotationAdopterTest.kt

@@ -0,0 +1,97 @@
+package app.closer.domain.usecase
+
+import app.closer.core.crash.CrashReporter
+import app.closer.crypto.CoupleEncryptionManager
+import app.closer.crypto.CoupleEncryptionManager.AdoptionResult
+import app.closer.domain.model.Couple
+import app.closer.domain.repository.CoupleRepository
+import io.mockk.coEvery
+import io.mockk.coVerify
+import io.mockk.every
+import io.mockk.mockk
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.test.runTest
+import org.junit.Assert.assertEquals
+import org.junit.Test
+
+/**
+ * The watcher is what covers the WARM app: with only the one-shot hooks (process start, Home load),
+ * a foregrounded app never adopted a partner's rotation at all — the 🔑 push routes to Security,
+ * which doesn't adopt, and a process can stay warm for days rendering 🔒 for every new-key message.
+ *
+ * Its contract: adopt on every couple emission, skip nulls, keep collecting past a failed adoption,
+ * and never let a stream error escape (a dead watcher degrades to the one-shot hooks — it must not
+ * take MainActivity's collector down with it).
+ */
+class CoupleKeyRotationAdopterTest {
+
+    private val coupleRepository: CoupleRepository = mockk()
+    private val encryptionManager: CoupleEncryptionManager = mockk()
+    private val crashReporter: CrashReporter = mockk(relaxUnitFun = true)
+
+    private val adopter = CoupleKeyRotationAdopter(coupleRepository, encryptionManager, crashReporter)
+
+    private val gen1 = Couple(id = "c1", keyGeneration = 1L)
+    private val gen2 = Couple(id = "c1", keyGeneration = 2L)
+
+    @Test
+    fun `every emission is offered for adoption — the warm-app rotation lands`() = runTest {
+        every { coupleRepository.observeCoupleForUser("u1") } returns flowOf(gen1, gen2)
+        coEvery { encryptionManager.adoptRotationIfNeeded(gen1) } returns AdoptionResult.UpToDate
+        coEvery { encryptionManager.adoptRotationIfNeeded(gen2) } returns AdoptionResult.Adopted
+
+        adopter.watch("u1")
+
+        coVerify(exactly = 1) { encryptionManager.adoptRotationIfNeeded(gen1) }
+        coVerify(exactly = 1) { encryptionManager.adoptRotationIfNeeded(gen2) }
+    }
+
+    @Test
+    fun `unpaired (null) emissions are skipped`() = runTest {
+        every { coupleRepository.observeCoupleForUser("u1") } returns flowOf(null, gen1)
+        coEvery { encryptionManager.adoptRotationIfNeeded(gen1) } returns AdoptionResult.UpToDate
+
+        adopter.watch("u1")
+
+        coVerify(exactly = 1) { encryptionManager.adoptRotationIfNeeded(any()) }
+    }
+
+    @Test
+    fun `a failed adoption is recorded and the watch keeps collecting`() = runTest {
+        val boom = RuntimeException("unwrap failed")
+        every { coupleRepository.observeCoupleForUser("u1") } returns flowOf(gen1, gen2)
+        coEvery { encryptionManager.adoptRotationIfNeeded(gen1) } returns AdoptionResult.Failed(boom)
+        coEvery { encryptionManager.adoptRotationIfNeeded(gen2) } returns AdoptionResult.Adopted
+
+        adopter.watch("u1")
+
+        coVerify { crashReporter.recordException(boom) }
+        // The failure on gen1 must not stop gen2 from being adopted.
+        coVerify(exactly = 1) { encryptionManager.adoptRotationIfNeeded(gen2) }
+    }
+
+    @Test
+    fun `a stream error is recorded, not thrown — the collector upstairs survives`() = runTest {
+        val listenerError = RuntimeException("PERMISSION_DENIED after sign-out race")
+        every { coupleRepository.observeCoupleForUser("u1") } returns flow {
+            emit(gen1)
+            throw listenerError
+        }
+        coEvery { encryptionManager.adoptRotationIfNeeded(gen1) } returns AdoptionResult.UpToDate
+
+        adopter.watch("u1") // must return normally
+
+        coVerify { crashReporter.recordException(listenerError) }
+    }
+
+    @Test
+    fun `one-shot entry point still resolves the couple and adopts`() = runTest {
+        coEvery { coupleRepository.getCoupleForUser("u1") } returns gen2
+        coEvery { encryptionManager.adoptRotationIfNeeded(gen2) } returns AdoptionResult.Adopted
+
+        val result = adopter.adoptIfNeeded("u1")
+
+        assertEquals(AdoptionResult.Adopted, result)
+    }
+}