feat(android): close the audit's remaining parity gaps

Seven items the earlier repair waves deliberately left alone, so that the
blocker fixes could land with a working safety net first.

CERTIFICATE LIFECYCLE (the audit's only "high"). iOS ships a
CertificateRotationScheduler; Android had no counterpart, so a device
certificate simply expired and needed a manual re-enroll. The decision is a
pure total function over five answers, with RE_ENROLL_REQUIRED checked first so
it outranks any backoff. Renewal is single-flight — a second trigger coalesces,
and a cancelled leader releases the slot so cancellation cannot wedge rotation
for the process lifetime. Failure leaves the prior identity fully live (the
existing atomic ping-pong flip already guaranteed that) and is published rather
than swallowed.

Two real defects surfaced while building it:

  renew() decoded its 201 with EnrollResponseDto, whose deviceId is required —
  but /device/:id/renew returns only {cert, caChain, notAfter}; only
  /device/enroll echoes deviceId. Every silent renewal would have thrown
  MalformedResponse. Fixed with a RenewResponseDto plus the deviceId from the
  path segment we addressed.

  And because no renew 201 carries renewAfter, a client that persisted it would
  rotate exactly once and then report not-due until the cert died. That is why
  renewAfter is re-derived from the live leaf as notBefore + 2/3·lifetime — the
  control plane's own formula. This is a latent iOS defect that Android now
  avoids rather than inherits.

/device/:id/recover was confirmed real (control-plane/src/api/renew.ts:443,
contract pinned by that suite's CP6e) and is now wired. It goes over a separate
NON-mTLS client that throws rather than falling back, because an expired client
certificate cannot authenticate its own recovery — the deadlock a previous
session already hit and recorded.

Asked explicitly about step-up: rotation is unaffected. stepUp appears nowhere
in control-plane/src; it lives only on the relay's WebSocket upgrade. On a
step-up host the SESSION is denied, not the rotation, so the certificate still
stays alive and that principal gap is orthogonal.

FOLLOW-UP QUEUE (W2). The queue frame decoded but the three routes managing it
were missing, so Android could see "N queued" and not queue anything. One
outcome union covers both guarded writes and distinguishes full / too-large /
disabled / session-gone / rate-limited, because they need different copy. 429 is
surfaced and never auto-retried — POST and DELETE share one 20/min bucket.
Queued text is raw shell input and travels verbatim, proven with control
characters and CJK.

UNREAD WATERMARK now persists, so the unread state survives process death —
which is exactly when it matters, since the product premise is walking away and
coming back. It stores a plain map rather than an UnreadLedger: the reducer
lives in :session-core, which :host-registry may not depend on, so the logic is
not restated anywhere.

COPY-OUT. Text selection was a recorded feature gap: the stock ActionMode's own
Copy handler dereferences the absent TerminalSession, so making it reachable
puts an NPE on the button — worse than no selection. The refusal was
re-verified from the bytecode, and the same disassembly showed the way out:
TerminalBuffer.getSelectedText touches only mLines/TerminalRow, no session and
no view. A first-party selection layer now builds on that, with a pure
row-major model so a backwards drag normalises correctly. Terminal content
reaches the clipboard only on an explicit user copy; OSC 52 stays declined.

HOST REMOVAL. PushRegistrar.unregisterHost had zero call sites, so a removed
host kept receiving pushes. Removal is now confirm-gated and cleans everything
keyed by host id — a half-removed host is worse than none.

/config/ui is finally honoured. It was implemented and never called, so
allowAutoMode was ignored and the plan gate offered "approve + auto" even where
the operator had disabled it. It fails CLOSED: an unfetchable config treats auto
as disabled, because the permissive default is the dangerous one.

Verified: ./gradlew test :app:assembleDebug :app:assembleDebugAndroidTest
:macrobenchmark:assembleBenchmark koverVerify -> BUILD SUCCESSFUL, 1150 JVM
tests, 0 failures (903 -> 1150). Instrumented suite re-run on emulator-5554
against live servers: 67 tests, 0 failures, 0 skipped — no device regression.

android/host-registry/src/androidTest/kotlin/wang/yaojia/webterm/hostregistry/DataStoreSessionWatermarkStoreTest.kt

@@ -0,0 +1,188 @@
+package wang.yaojia.webterm.hostregistry
+
+import android.content.Context
+import androidx.datastore.core.DataStore
+import androidx.datastore.preferences.core.PreferenceDataStoreFactory
+import androidx.datastore.preferences.core.Preferences
+import androidx.datastore.preferences.core.edit
+import androidx.datastore.preferences.core.stringPreferencesKey
+import androidx.datastore.preferences.preferencesDataStoreFile
+import androidx.test.core.app.ApplicationProvider
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.job
+import kotlinx.coroutines.runBlocking
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertNull
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import org.junit.runner.RunWith
+import java.util.UUID
+
+/**
+ * Instrumented (device) contract tests for [DataStoreSessionWatermarkStore] — the storage half
+ * of the module, which needs a real Context/file (plan §7).
+ *
+ * The load-bearing one is [survives_a_new_store_instance_over_the_same_file]: that is the
+ * process-death simulation this whole class exists for. The rest mirror the JVM transform tests
+ * at the storage boundary (garbage at rest is dropped, the cap holds, an emptied map removes the
+ * key instead of leaving `{}` behind).
+ */
+@RunWith(AndroidJUnit4::class)
+class DataStoreSessionWatermarkStoreTest {
+
+    private val idA = "3f2504e0-4f89-41d3-9a0c-0305e82c3301"
+    private val idB = "3f2504e0-4f89-41d3-9a0c-0305e82c3302"
+
+    /**
+     * A fresh, uniquely-named Preferences file per test (no cross-test bleed). [scope] is
+     * injectable because DataStore throws if two instances are live on ONE file: releasing the
+     * file means cancelling the owning scope, which is how process death is simulated below.
+     */
+    private fun newDataStore(
+        name: String = "watermarks-test-${UUID.randomUUID()}",
+        scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob()),
+    ): DataStore<Preferences> {
+        val ctx = ApplicationProvider.getApplicationContext<Context>()
+        return PreferenceDataStoreFactory.create(
+            scope = scope,
+            produceFile = { ctx.preferencesDataStoreFile(name) },
+        )
+    }
+
+    @Test
+    fun an_empty_store_loads_an_empty_map() = runBlocking {
+        assertTrue(DataStoreSessionWatermarkStore(newDataStore()).loadAll().isEmpty())
+    }
+
+    @Test
+    fun replaceAll_then_loadAll_returns_the_same_watermarks() = runBlocking {
+        val store = DataStoreSessionWatermarkStore(newDataStore())
+
+        store.replaceAll(mapOf(idA to 10L, idB to 20L))
+
+        assertEquals(mapOf(idA to 10L, idB to 20L), store.loadAll())
+    }
+
+    @Test
+    fun survives_a_new_store_instance_over_the_same_file() = runBlocking {
+        val fileName = "watermarks-persist-${UUID.randomUUID()}"
+        val firstScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
+        // replaceAll suspends until edit() has committed to disk.
+        DataStoreSessionWatermarkStore(newDataStore(fileName, firstScope))
+            .replaceAll(mapOf(idA to 1_700_000_000_000L))
+
+        // Cancelling the owning scope releases the file — DataStore refuses two live instances
+        // on one file, and letting go of it is exactly what process death does.
+        firstScope.coroutineContext.job.cancelAndJoin()
+
+        // A brand-new store over the same file = what a cold start after process death sees.
+        val reopened = DataStoreSessionWatermarkStore(newDataStore(fileName))
+
+        assertEquals(mapOf(idA to 1_700_000_000_000L), reopened.loadAll())
+    }
+
+    @Test
+    fun replaceAll_is_a_wholesale_swap_not_a_merge() = runBlocking {
+        val store = DataStoreSessionWatermarkStore(newDataStore())
+        store.replaceAll(mapOf(idA to 10L))
+
+        store.replaceAll(mapOf(idB to 20L))
+
+        assertEquals(mapOf(idB to 20L), store.loadAll())
+    }
+
+    @Test
+    fun remove_drops_one_session_and_keeps_the_rest() = runBlocking {
+        val store = DataStoreSessionWatermarkStore(newDataStore())
+        store.replaceAll(mapOf(idA to 10L, idB to 20L))
+
+        val remaining = store.remove(idA)
+
+        assertEquals(mapOf(idB to 20L), remaining)
+        assertEquals(remaining, store.loadAll())
+    }
+
+    @Test
+    fun removing_an_unknown_session_is_a_no_op() = runBlocking {
+        val store = DataStoreSessionWatermarkStore(newDataStore())
+        store.replaceAll(mapOf(idA to 10L))
+
+        assertEquals(mapOf(idA to 10L), store.remove(idB))
+    }
+
+    @Test
+    fun emptying_the_map_removes_the_stored_key_instead_of_writing_an_empty_blob() = runBlocking {
+        val dataStore = newDataStore()
+        val store = DataStoreSessionWatermarkStore(dataStore)
+        store.replaceAll(mapOf(idA to 10L))
+
+        store.remove(idA)
+
+        assertNull(dataStore.data.first()[stringPreferencesKey(WATERMARKS_KEY_NAME)])
+    }
+
+    @Test
+    fun garbage_stored_at_rest_reads_back_as_no_watermarks() = runBlocking {
+        val dataStore = newDataStore()
+        dataStore.edit { it[stringPreferencesKey(WATERMARKS_KEY_NAME)] = "{not json" }
+
+        assertTrue(DataStoreSessionWatermarkStore(dataStore).loadAll().isEmpty())
+    }
+
+    @Test
+    fun a_non_v4_id_stored_at_rest_is_dropped_on_read() = runBlocking {
+        val dataStore = newDataStore()
+        // A v1 UUID: parses as a UUID (the DataStoreLastSessionStore precedent bug) but is not
+        // a valid wire session id, so Validation.isValidSessionId must reject it.
+        val v1 = "3f2504e0-4f89-11d3-9a0c-0305e82c3301"
+        dataStore.edit { it[stringPreferencesKey(WATERMARKS_KEY_NAME)] = """{"$idA":10,"$v1":20}""" }
+
+        assertEquals(mapOf(idA to 10L), DataStoreSessionWatermarkStore(dataStore).loadAll())
+    }
+
+    @Test
+    fun the_stored_map_is_capped_so_it_cannot_grow_without_bound() = runBlocking {
+        val store = DataStoreSessionWatermarkStore(newDataStore())
+        val oversized = (0 until MAX_WATERMARKS + 5).associate { sessionId(it) to (it + 1).toLong() }
+
+        val stored = store.replaceAll(oversized)
+
+        assertEquals(MAX_WATERMARKS, stored.size)
+        assertEquals(MAX_WATERMARKS, store.loadAll().size)
+        assertFalse(store.loadAll().containsKey(sessionId(0))) // oldest-seen dropped first
+    }
+
+    @Test
+    fun the_watermark_key_is_disjoint_from_the_other_stores_over_one_file() = runBlocking {
+        // ONE Preferences DataStore is shared app-wide (:app StorageModule) — this is the
+        // regression test for "never collide with a sibling store's key".
+        val dataStore = newDataStore()
+        val hosts = DataStoreHostStore(dataStore)
+        val lastSession = DataStoreLastSessionStore(dataStore)
+        val watermarks = DataStoreSessionWatermarkStore(dataStore)
+        val host = Host.create(id = "host-1", name = "mac", baseUrl = "http://10.0.2.2:3000")!!
+
+        hosts.upsert(host)
+        lastSession.setLastSessionId(idB, hostId = "host-1")
+        watermarks.replaceAll(mapOf(idA to 10L))
+
+        assertEquals(listOf(host), hosts.loadAll())
+        assertEquals(idB, lastSession.lastSessionId("host-1"))
+        assertEquals(mapOf(idA to 10L), watermarks.loadAll())
+    }
+
+    /** A distinct valid v4 id per [index] (the last 4 hex digits carry the index). */
+    private fun sessionId(index: Int): String =
+        "3f2504e0-4f89-41d3-9a0c-0305e82c%04x".format(index)
+
+    private companion object {
+        /** Must match `DataStoreSessionWatermarkStore.WATERMARKS_KEY` (private there, asserted here). */
+        const val WATERMARKS_KEY_NAME = "unreadWatermarks"
+    }
+}

android/host-registry/src/main/kotlin/wang/yaojia/webterm/hostregistry/DataStoreSessionWatermarkStore.kt

@@ -0,0 +1,65 @@
+package wang.yaojia.webterm.hostregistry
+
+import androidx.datastore.core.DataStore
+import androidx.datastore.preferences.core.Preferences
+import androidx.datastore.preferences.core.edit
+import androidx.datastore.preferences.core.stringPreferencesKey
+import kotlinx.coroutines.flow.first
+
+/**
+ * Preferences-DataStore-backed [SessionWatermarkStore]. The whole watermark map is ONE JSON
+ * string under [WATERMARKS_KEY] ([SessionWatermarkCodec]) — the same shape [DataStoreHostStore]
+ * uses for `"hosts"`, and for the same reason: the cap and the sanitize pass have to apply to
+ * the map as a whole, which a per-session key prefix (the [DataStoreLastSessionStore] shape)
+ * cannot do without scanning every key on every write and still leaking orphaned keys.
+ *
+ * The key name mirrors the iOS `UserDefaults` key `"unreadWatermarks"`. It is **disjoint** from
+ * every other key over this file (`"hosts"`, `"lastSessionId.<hostId>"`, `"authCookiesSealed"`),
+ * because ONE Preferences DataStore instance is shared app-wide (`:app` `StorageModule`) — a
+ * second instance over the same file would corrupt it, so never open one.
+ *
+ * Reads decode then [forStorage] (validate + cap): the blob is untrusted at rest. Writes are an
+ * atomic read-modify-write via [DataStore.edit], and an empty result REMOVES the key rather than
+ * writing `{}`, so the file shrinks back to nothing.
+ *
+ * The [DataStore] is injected (constructed from a Context in `:app` DI) so this class has no
+ * Android-framework surface of its own; the on-device behaviour — including survival across a
+ * fresh store instance, i.e. process death — is verified instrumented (androidTest, plan §7).
+ * Not encrypted, deliberately: see the posture note on [SessionWatermarkStore].
+ */
+public class DataStoreSessionWatermarkStore(
+    private val dataStore: DataStore<Preferences>,
+) : SessionWatermarkStore {
+
+    override suspend fun loadAll(): Map<String, Long> =
+        SessionWatermarkCodec.decode(dataStore.data.first()[WATERMARKS_KEY]).forStorage()
+
+    override suspend fun replaceAll(watermarks: Map<String, Long>): Map<String, Long> =
+        writeTransform { watermarks.forStorage() }
+
+    override suspend fun remove(sessionId: String): Map<String, Long> =
+        writeTransform { it.removingWatermark(sessionId) }
+
+    private suspend fun writeTransform(
+        transform: (Map<String, Long>) -> Map<String, Long>,
+    ): Map<String, Long> {
+        lateinit var updated: Map<String, Long>
+        dataStore.edit { prefs ->
+            // Decoding through forStorage() means every write also garbage-collects invalid
+            // rows and re-applies the cap, even when the transform itself only removes.
+            val current = SessionWatermarkCodec.decode(prefs[WATERMARKS_KEY]).forStorage()
+            updated = transform(current)
+            if (updated.isEmpty()) {
+                prefs.remove(WATERMARKS_KEY)
+            } else {
+                prefs[WATERMARKS_KEY] = SessionWatermarkCodec.encode(updated)
+            }
+        }
+        return updated
+    }
+
+    private companion object {
+        /** The one stored value: `SessionWatermarkCodec.encode(watermarks)`. */
+        val WATERMARKS_KEY = stringPreferencesKey("unreadWatermarks")
+    }
+}

android/host-registry/src/main/kotlin/wang/yaojia/webterm/hostregistry/InMemorySessionWatermarkStore.kt

@@ -0,0 +1,40 @@
+package wang.yaojia.webterm.hostregistry
+
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+
+/**
+ * In-memory [SessionWatermarkStore]. Lives in `main` (not `test`) on purpose — it stands in for
+ * the DataStore store in this module's contract tests AND in the AW4 ViewModel tests, exactly
+ * like [InMemoryHostStore] / [InMemoryAuthCookieStore].
+ *
+ * It applies the SAME [forStorage] rules (validation + cap) as [DataStoreSessionWatermarkStore];
+ * a laxer double would let a ViewModel test pass against behaviour no device exhibits.
+ *
+ * A [Mutex] serialises each read-modify-write; the state is a value snapshot replaced wholesale
+ * on every change (no in-place mutation of a shared reference).
+ */
+public class InMemorySessionWatermarkStore(
+    initial: Map<String, Long> = emptyMap(),
+) : SessionWatermarkStore {
+    private val mutex = Mutex()
+
+    // forStorage() always builds a fresh map, so a mutable map handed to the ctor cannot alias state.
+    private var stored: Map<String, Long> = initial.forStorage()
+
+    override suspend fun loadAll(): Map<String, Long> = mutex.withLock { stored }
+
+    override suspend fun replaceAll(watermarks: Map<String, Long>): Map<String, Long> =
+        write { watermarks.forStorage() }
+
+    override suspend fun remove(sessionId: String): Map<String, Long> =
+        write { it.removingWatermark(sessionId) }
+
+    private suspend fun write(
+        transform: (Map<String, Long>) -> Map<String, Long>,
+    ): Map<String, Long> = mutex.withLock {
+        val updated = transform(stored)
+        stored = updated
+        updated
+    }
+}

android/host-registry/src/main/kotlin/wang/yaojia/webterm/hostregistry/SessionWatermarkCodec.kt

@@ -0,0 +1,33 @@
+package wang.yaojia.webterm.hostregistry
+
+import kotlinx.serialization.decodeFromString
+import kotlinx.serialization.encodeToString
+import kotlinx.serialization.json.Json
+
+/**
+ * Pure JSON codec for the persisted watermark map (shared by [DataStoreSessionWatermarkStore],
+ * JVM-testable without a Context). Mirrors [HostCodec]: encode is total, decode is defensive —
+ * a corrupt/undecodable blob reads back as "no watermarks" so a cold start can never crash on
+ * a bad row.
+ *
+ * Deliberately dumb about *meaning*: it does not validate session ids or instants. That is
+ * [sanitizedWatermarks]' job, so exactly one place owns the trust decision (and the in-memory
+ * double gets the identical treatment without touching JSON).
+ */
+internal object SessionWatermarkCodec {
+    private val json = Json {
+        ignoreUnknownKeys = true
+        isLenient = true
+    }
+
+    fun encode(watermarks: Map<String, Long>): String = json.encodeToString(watermarks)
+
+    fun decode(raw: String?): Map<String, Long> {
+        if (raw.isNullOrBlank()) return emptyMap()
+        return try {
+            json.decodeFromString<Map<String, Long>>(raw)
+        } catch (_: Exception) {
+            emptyMap() // corrupted blob at rest → start clean, never crash
+        }
+    }
+}

android/host-registry/src/main/kotlin/wang/yaojia/webterm/hostregistry/SessionWatermarkStore.kt

@@ -0,0 +1,151 @@
+package wang.yaojia.webterm.hostregistry
+
+import wang.yaojia.webterm.wire.Validation
+
+/**
+ * Durable storage for the unread **watermarks** — `sessionId → last-seen instant (ms since
+ * epoch)` — that decide the session-list unread dot. Ports the iOS
+ * `UnreadWatermarkStore` / `UserDefaultsUnreadWatermarkStore` pair (plan §2 lists "unread
+ * watermarks" under *Persistence (non-secret)*).
+ *
+ * Implementations: [DataStoreSessionWatermarkStore] (real, Preferences-DataStore-backed) and
+ * [InMemorySessionWatermarkStore] (in-`main` double, exactly like [InMemoryHostStore] /
+ * [InMemoryAuthCookieStore]).
+ *
+ * ## Why this exists at all
+ * Without it the dot resets whenever the process dies — which is precisely when it matters:
+ * the product premise is walking away and coming back, and a dot that only survives while
+ * the app is warm answers a question the user never asked.
+ *
+ * ## Division of labour with `UnreadLedger` (do not duplicate it)
+ * The *reducer* — monotonic `record`, the `isUnread` comparison, the tie-broken cap — lives
+ * in `:session-core`'s `UnreadLedger`, which is deliberately persistence-agnostic. This
+ * module cannot see it (`:host-registry` depends only on `:wire-protocol`; nothing points
+ * sideways — see `android/README.md`), and must not restate it. So the contract here is
+ * plain-map I/O, like the iOS store's `[UUID: Int]`: `:app` owns the ledger, hands its
+ * `watermarks` map to [replaceAll], and rebuilds a ledger from [loadAll] on cold start.
+ *
+ * ## At-rest posture — no Tink here, deliberately
+ * The neighbouring [AuthCookieStore] blob IS Tink-AEAD-sealed, so a reader will reasonably
+ * ask why this one is not. Because it is not a credential and not private data: a watermark
+ * is a UUID the device already stores in the clear (the host list, the last-session id) plus
+ * a coarse timestamp of when the user last *looked* at a session. Sealing it would buy
+ * nothing an attacker with app-private read access does not already have, while adding a
+ * fail-closed dependency on `AndroidKeyStore` — i.e. a keystore hiccup would silently reset
+ * unread state. Plan §2's split is explicit: secrets get the Tink/AndroidKeyStore tier,
+ * non-secret UI state gets plain Preferences DataStore. This is the second tier, same as
+ * [HostStore] and [LastSessionStore]. (`android:allowBackup="false"` still applies to the
+ * whole file.)
+ *
+ * ## Untrusted at rest
+ * Persisted ids are re-validated on every read with the frozen
+ * [Validation.isValidSessionId] — the SAME v4-specific guard the wire codec applies to
+ * server-issued ids, **not** a format-only `UUID.fromString` (that mistake was already made
+ * once in [DataStoreLastSessionStore] and had to be corrected; do not repeat it).
+ *
+ * Immutable style throughout, like [HostStore]: mutations RETURN the new map instead of
+ * mutating shared state, an unknown id is an explicit no-op, and encounter order is
+ * preserved (only the cap may reorder — see [cappedWatermarks]).
+ */
+public interface SessionWatermarkStore {
+    /**
+     * Every stored watermark, already sanitized and capped ([forStorage]). Empty when
+     * nothing is stored or the blob no longer decodes — never throws.
+     */
+    public suspend fun loadAll(): Map<String, Long>
+
+    /**
+     * Swap the whole map for [watermarks] — the write `:app` drives after a `markSeen`
+     * ("this is the ledger now"). A wholesale swap, NOT a merge: the in-memory `UnreadLedger`
+     * is the source of truth, so a merge here could resurrect an entry the ledger dropped.
+     * Returns exactly what was stored (i.e. [watermarks].[forStorage]), which may be a
+     * subset of the argument.
+     */
+    public suspend fun replaceAll(watermarks: Map<String, Long>): Map<String, Long>
+
+    /**
+     * Forget [sessionId]'s watermark — the eager GC hook for a session that ended (`exit`
+     * frame / a kill), matching how [LastSessionStore] is cleared on `.exited`. An unknown
+     * or malformed id is an explicit no-op returning the unchanged map, never a throw.
+     */
+    public suspend fun remove(sessionId: String): Map<String, Long>
+}
+
+// ── Pure immutable transforms shared by every SessionWatermarkStore ───────────────────────
+// (DRY, mirrors HostStore.kt / AuthCookieStore.kt). Never mutate the receiver — always
+// return a fresh map. `internal` so both stores and the same-module JVM tests can use them
+// without widening the public surface.
+
+/**
+ * Hard ceiling on stored watermarks. **This is what bounds growth** (see the GC note below)
+ * and mirrors `UnreadLedger.MAX_ENTRIES` in `:session-core`, which cannot be imported here
+ * (no sideways module edge). Keep the two numbers equal: the ledger caps what `:app` holds,
+ * this caps what reaches the disk, and a store that trusted the caller would be no bound at
+ * all.
+ *
+ * ### GC rule (the decision, stated explicitly)
+ * Sessions are ephemeral, so a watermark can outlive its session forever. Growth is bounded
+ * by **three** rules, in order of preference:
+ *  1. **Eager** — [SessionWatermarkStore.remove] on a session that exits or is killed. This
+ *     is the only rule that frees an entry the moment it is provably dead.
+ *  2. **Structural** — this count cap. Over [MAX_WATERMARKS], the OLDEST-seen entries are
+ *     dropped ([cappedWatermarks]), so the file can never exceed ~512 · ~50 B ≈ 25 KB no
+ *     matter how many sessions a device sees or how many exits it misses (force-stop, crash,
+ *     a host killed from another device). This is the backstop that actually guarantees the
+ *     bound.
+ *  3. **Hygiene** — [sanitizedWatermarks] drops rows that no longer validate, so a format
+ *     change or a corrupted row cannot accumulate either.
+ *
+ * Deliberately NOT a rule: pruning to the ids returned by the latest `GET /live-sessions`.
+ * That looks tempting and is wrong — the fetch covers ONE host, so it would wipe every other
+ * host's watermarks on a host switch, and an offline/erroring host returns nothing at all
+ * (iOS records the same decision in `SessionListViewModel`).
+ *
+ * Also deliberately NOT a rule: a staleness/TTL sweep. Dropping an old watermark re-lights
+ * the dot for a session that is still alive and still unchanged — a *false* unread, which is
+ * worse than 25 KB.
+ */
+internal const val MAX_WATERMARKS: Int = 512
+
+/**
+ * Drop everything untrustworthy: ids that are not valid wire session ids
+ * ([Validation.isValidSessionId]) and non-positive instants (`<= 0` means "never seen", which
+ * is already the default for a missing entry — storing it is pure waste). Valid ids are
+ * canonicalised to lower case and case-variant spellings of the same id collapse to their
+ * newest instant (they are the same session; `Validation` is case-insensitive, `UUID.toString`
+ * is lower case, and merging by max can never lower a watermark). Encounter order is kept.
+ */
+internal fun Map<String, Long>.sanitizedWatermarks(): Map<String, Long> {
+    val out = LinkedHashMap<String, Long>(size)
+    for ((id, atMs) in this) {
+        if (atMs <= 0L || !Validation.isValidSessionId(id)) continue
+        val key = id.lowercase()
+        val existing = out[key]
+        out[key] = if (existing == null) atMs else maxOf(existing, atMs)
+    }
+    return out
+}
+
+/**
+ * Keep at most the [MAX_WATERMARKS] NEWEST watermarks, tie-broken by session id so the
+ * outcome is deterministic across runs (same rule as `UnreadLedger.capped`). Under the cap
+ * the receiver's own order is returned untouched; over it the result is newest-first, since a
+ * map that had to be truncated has no meaningful insertion order left to preserve.
+ */
+internal fun Map<String, Long>.cappedWatermarks(): Map<String, Long> {
+    if (size <= MAX_WATERMARKS) return this
+    return entries
+        .sortedWith(compareByDescending<Map.Entry<String, Long>> { it.value }.thenBy { it.key })
+        .take(MAX_WATERMARKS)
+        .associate { it.key to it.value }
+}
+
+/** A new map without [sessionId]'s watermark (case-insensitive). Unknown id → unchanged contents. */
+internal fun Map<String, Long>.removingWatermark(sessionId: String): Map<String, Long> {
+    val target = sessionId.lowercase()
+    return filterKeys { it.lowercase() != target }
+}
+
+/** The single write/read gate: sanitize, then cap. Everything crossing storage goes through it. */
+internal fun Map<String, Long>.forStorage(): Map<String, Long> =
+    sanitizedWatermarks().cappedWatermarks()

android/host-registry/src/test/kotlin/wang/yaojia/webterm/hostregistry/InMemorySessionWatermarkStoreTest.kt

@@ -0,0 +1,77 @@
+package wang.yaojia.webterm.hostregistry
+
+import kotlinx.coroutines.test.runTest
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+
+/**
+ * Contract tests for the in-`main` [InMemorySessionWatermarkStore] double (the same role
+ * [InMemoryHostStore] / [InMemoryAuthCookieStore] play: it stands in for the DataStore
+ * store in the AW4 ViewModel tests). The double must enforce the SAME at-rest rules as
+ * the real store, otherwise a ViewModel test can pass against behaviour the device never
+ * exhibits.
+ */
+class InMemorySessionWatermarkStoreTest {
+
+    private val idA = "3f2504e0-4f89-41d3-9a0c-0305e82c3301"
+    private val idB = "3f2504e0-4f89-41d3-9a0c-0305e82c3302"
+
+    @Test
+    fun `an empty store loads an empty map`() = runTest {
+        assertTrue(InMemorySessionWatermarkStore().loadAll().isEmpty())
+    }
+
+    @Test
+    fun `replaceAll then loadAll returns the same watermarks`() = runTest {
+        val store = InMemorySessionWatermarkStore()
+
+        store.replaceAll(mapOf(idA to 10L, idB to 20L))
+
+        assertEquals(mapOf(idA to 10L, idB to 20L), store.loadAll())
+    }
+
+    @Test
+    fun `replaceAll returns exactly what was stored`() = runTest {
+        val store = InMemorySessionWatermarkStore()
+
+        val stored = store.replaceAll(mapOf(idA to 10L, "garbage" to 20L))
+
+        assertEquals(mapOf(idA to 10L), stored)
+        assertEquals(stored, store.loadAll())
+    }
+
+    @Test
+    fun `replaceAll is a wholesale swap, not a merge`() = runTest {
+        val store = InMemorySessionWatermarkStore(mapOf(idA to 10L))
+
+        store.replaceAll(mapOf(idB to 20L))
+
+        assertEquals(mapOf(idB to 20L), store.loadAll())
+    }
+
+    @Test
+    fun `an initial map is sanitized too`() = runTest {
+        val store = InMemorySessionWatermarkStore(mapOf(idA to 10L, "garbage" to 20L))
+
+        assertEquals(mapOf(idA to 10L), store.loadAll())
+    }
+
+    @Test
+    fun `remove drops one session and returns the new map`() = runTest {
+        val store = InMemorySessionWatermarkStore(mapOf(idA to 10L, idB to 20L))
+
+        val remaining = store.remove(idA)
+
+        assertEquals(mapOf(idB to 20L), remaining)
+        assertEquals(remaining, store.loadAll())
+    }
+
+    @Test
+    fun `removing an unknown session is a no-op`() = runTest {
+        val store = InMemorySessionWatermarkStore(mapOf(idA to 10L))
+
+        assertEquals(mapOf(idA to 10L), store.remove(idB))
+        assertEquals(mapOf(idA to 10L), store.loadAll())
+    }
+}

android/host-registry/src/test/kotlin/wang/yaojia/webterm/hostregistry/SessionWatermarkCodecTest.kt

@@ -0,0 +1,60 @@
+package wang.yaojia.webterm.hostregistry
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+
+/**
+ * At-rest codec tests (mirrors `HostCodecTest`): encode is total, decode is defensive —
+ * the blob is untrusted at rest, so anything undecodable reads back as "no watermarks"
+ * rather than crashing the session list on cold start.
+ */
+class SessionWatermarkCodecTest {
+
+    private val idA = "3f2504e0-4f89-41d3-9a0c-0305e82c3301"
+    private val idB = "3f2504e0-4f89-41d3-9a0c-0305e82c3302"
+
+    @Test
+    fun `round-trips a watermark map`() {
+        val map = mapOf(idA to 1_700_000_000_000L, idB to 42L)
+
+        assertEquals(map, SessionWatermarkCodec.decode(SessionWatermarkCodec.encode(map)))
+    }
+
+    @Test
+    fun `an empty map round-trips to an empty map`() {
+        assertTrue(SessionWatermarkCodec.decode(SessionWatermarkCodec.encode(emptyMap())).isEmpty())
+    }
+
+    @Test
+    fun `null reads back as empty`() {
+        assertTrue(SessionWatermarkCodec.decode(null).isEmpty())
+    }
+
+    @Test
+    fun `blank reads back as empty`() {
+        assertTrue(SessionWatermarkCodec.decode("   ").isEmpty())
+    }
+
+    @Test
+    fun `a corrupt blob reads back as empty instead of throwing`() {
+        assertTrue(SessionWatermarkCodec.decode("{not json").isEmpty())
+    }
+
+    @Test
+    fun `a wrong-shaped blob reads back as empty`() {
+        // A JSON array where a map was written (e.g. a hand-edited or superseded format).
+        assertTrue(SessionWatermarkCodec.decode("""["$idA"]""").isEmpty())
+    }
+
+    @Test
+    fun `a non-numeric instant reads back as empty`() {
+        assertTrue(SessionWatermarkCodec.decode("""{"$idA":"yesterday"}""").isEmpty())
+    }
+
+    @Test
+    fun `decode does not itself validate ids - the store transforms do`() {
+        // Keeping the codec dumb keeps ONE place (sanitizedWatermarks) responsible for trust.
+        assertEquals(mapOf("garbage" to 5L), SessionWatermarkCodec.decode("""{"garbage":5}"""))
+    }
+}

android/host-registry/src/test/kotlin/wang/yaojia/webterm/hostregistry/SessionWatermarkTransformsTest.kt

@@ -0,0 +1,150 @@
+package wang.yaojia.webterm.hostregistry
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+
+/**
+ * Pure-transform tests for the watermark map shared by [InMemorySessionWatermarkStore] and
+ * [DataStoreSessionWatermarkStore] (mirrors `HostStoreTransformsTest` /
+ * `AuthCookieStoreTransformsTest`). These are the JVM-testable half of the storage split
+ * (plan §3: "the logic half of :host-registry" is in the Kover gate).
+ *
+ * Covers the three at-rest defences and the GC bound:
+ *  - every stored session id is re-validated with the frozen `Validation.isValidSessionId`;
+ *  - non-positive instants are dropped (a watermark of 0 means "never seen" anyway);
+ *  - case-variant spellings of one id collapse via max (they are the same session);
+ *  - the map is capped at [MAX_WATERMARKS], newest kept — what bounds growth.
+ */
+class SessionWatermarkTransformsTest {
+
+    private val idA = "3f2504e0-4f89-41d3-9a0c-0305e82c3301"
+    private val idB = "3f2504e0-4f89-41d3-9a0c-0305e82c3302"
+
+    // ── sanitizedWatermarks() ─────────────────────────────────────────────────
+
+    @Test
+    fun `a valid map survives sanitizing unchanged`() {
+        val map = mapOf(idA to 10L, idB to 20L)
+
+        assertEquals(map, map.sanitizedWatermarks())
+    }
+
+    @Test
+    fun `a non-v4 session id is dropped`() {
+        // UUID v1 (version nibble 1) parses as a UUID but is not a valid wire session id.
+        val v1 = "3f2504e0-4f89-11d3-9a0c-0305e82c3301"
+        val map = mapOf(idA to 10L, v1 to 20L, "abc123" to 30L, "" to 40L)
+
+        assertEquals(mapOf(idA to 10L), map.sanitizedWatermarks())
+    }
+
+    @Test
+    fun `non-positive instants are dropped`() {
+        val map = mapOf(idA to 0L, idB to -5L)
+
+        assertTrue(map.sanitizedWatermarks().isEmpty())
+    }
+
+    @Test
+    fun `case-variant spellings of one id collapse to the newest`() {
+        val map = mapOf(idA to 10L, idA.uppercase() to 99L)
+
+        assertEquals(mapOf(idA to 99L), map.sanitizedWatermarks())
+    }
+
+    @Test
+    fun `sanitizing preserves encounter order`() {
+        val map = linkedMapOf(idB to 20L, idA to 10L)
+
+        assertEquals(listOf(idB, idA), map.sanitizedWatermarks().keys.toList())
+    }
+
+    @Test
+    fun `sanitizing never mutates the receiver`() {
+        val map = linkedMapOf(idA to 10L, "abc123" to 20L)
+
+        map.sanitizedWatermarks()
+
+        assertEquals(linkedMapOf(idA to 10L, "abc123" to 20L), map)
+    }
+
+    // ── cappedWatermarks() ───────────────────────────────────────────────────
+
+    @Test
+    fun `a map at the cap is returned unchanged`() {
+        val map = watermarks(MAX_WATERMARKS)
+
+        assertEquals(map, map.cappedWatermarks())
+    }
+
+    @Test
+    fun `over the cap the newest entries are kept`() {
+        val map = watermarks(MAX_WATERMARKS + 10)
+
+        val capped = map.cappedWatermarks()
+
+        assertEquals(MAX_WATERMARKS, capped.size)
+        // Instants ascend with the index, so the 10 oldest (lowest) must be the dropped ones.
+        assertFalse(capped.containsKey(sessionId(0)))
+        assertTrue(capped.containsKey(sessionId(MAX_WATERMARKS + 9)))
+    }
+
+    @Test
+    fun `equal instants break the tie deterministically by session id`() {
+        val map = (0 until MAX_WATERMARKS + 2).associate { sessionId(it) to 7L }
+
+        val capped = map.cappedWatermarks()
+
+        assertEquals(MAX_WATERMARKS, capped.size)
+        assertEquals(map.cappedWatermarks().keys, capped.keys) // stable across runs
+        // Lowest ids win the tie-break, so the two highest ids are the dropped ones.
+        assertFalse(capped.containsKey(sessionId(MAX_WATERMARKS + 1)))
+    }
+
+    // ── removingWatermark() ──────────────────────────────────────────────────
+
+    @Test
+    fun `removing drops just that session`() {
+        val map = mapOf(idA to 10L, idB to 20L)
+
+        assertEquals(mapOf(idB to 20L), map.removingWatermark(idA))
+    }
+
+    @Test
+    fun `removing an unknown session is a no-op`() {
+        val map = mapOf(idA to 10L)
+
+        assertEquals(map, map.removingWatermark(idB))
+    }
+
+    @Test
+    fun `removing is case-insensitive like the id itself`() {
+        val map = mapOf(idA to 10L)
+
+        assertTrue(map.removingWatermark(idA.uppercase()).isEmpty())
+    }
+
+    // ── forStorage() = sanitize + cap ────────────────────────────────────────
+
+    @Test
+    fun `forStorage sanitizes and caps in one pass`() {
+        val map = watermarks(MAX_WATERMARKS + 1) + mapOf("garbage" to 999_999L)
+
+        val stored = map.forStorage()
+
+        assertEquals(MAX_WATERMARKS, stored.size)
+        assertFalse(stored.containsKey("garbage"))
+    }
+
+    // ── helpers ─────────────────────────────────────────────────────────────
+
+    /** A distinct valid v4 id per [index] (the last 4 hex digits carry the index). */
+    private fun sessionId(index: Int): String =
+        "3f2504e0-4f89-41d3-9a0c-0305e82c%04x".format(index)
+
+    /** [count] valid watermarks whose instants ascend with the index (1-based, so all positive). */
+    private fun watermarks(count: Int): Map<String, Long> =
+        (0 until count).associate { sessionId(it) to (it + 1).toLong() }
+}