fix(tor): self-heal a guard sample that is rotten at runtime, not just on disk

Tor wedged across app restarts with ~87% of relay connections failing, and
neither existing recovery fired. Captured from the device:

  guards.json  default: 60 guards, 59 disabled, 1 unlisted -> 1 usable
  Arti log     AllGuardsDown { n_accepted: 0, n_rejected: 60 }  x21,490
  sockets      36,007 failures vs 5,276 successful opens

Everything above Tor degraded with it. Concord was the visible casualty: its
control-plane sync drained exactly 12 wraps on every launch and never folded, so
the Messages tab showed zero Concord rows for two restarts running.

Two recoveries exist and this state slipped between both:

- `ArtiGuardState.hasNoUsableGuards` required `usable == 0`. Its own KDoc claimed
  "a single usable guard is enough to recover" — the device disproved it. One
  survivor that is merely *unreachable* is useless for recovery but sufficient to
  veto it, so the wipe never ran.
- `TorManager`'s watchdog only arms while status sits at Connecting. Tor had
  bootstrapped: the SOCKS proxy was bound, `hasEverBootstrapped` was true, ~13% of
  connections still worked, so status reached Active and the watchdog never fired.
  It is built for "Tor never came up"; this is "Tor came up and its guards rotted".

Nothing observed the steady-state failure rate, so all three gates evaluated the
same way on every launch and `guards.json` carried the wedge forward forever.

1. Proportional disk rule. A sample of at least MIN_SAMPLE_TO_JUDGE_RATIO whose
   usable guards fall under 1/USABLE_RATIO_DIVISOR of the total is wedged. Below
   that size only the strict `usable == 0` rule applies, so a young sample Arti is
   still filling is never wiped out from under a legitimate first bootstrap.

2. Runtime detector. `TorService` counts Arti's own AllGuardsDown log lines
   (GUARDS_DOWN_THRESHOLD within GUARDS_DOWN_WINDOW_MS) and exposes
   `TorBackend.guardsDownSignal`; `TorManager` routes it through the same
   rate-limited self-heal to `resetWithCleanState()`. This is the general fix: it
   believes Arti when it says every guard was rejected, so it fires even while
   `guards.json` still looks healthy and status is Active — precisely the blind
   spot between the two older heuristics. The count lives in the log callback
   because that is the only place Arti surfaces it; the reset stays in TorManager,
   which owns the cadence.

Verified on the wedged device. Next launch, unprompted:

  W TorService: No usable Arti guards left on disk — wiping state to rebuild the guard sample

  guard sample   60 total / 1 usable   ->  20 total / 20 usable / 0 disabled
  AllGuardsDown  21,569                ->  0
  sockets        36,007 fail / 5,276   ->  444 fail / 232 open
  Concord wraps  12, 12, 12 (pinned)   ->  12 -> 37 -> 258
  Concord rows   0                     ->  4 and climbing

Tests cover the 59-of-60 field case, a small-sample false-positive guard, and a
healthy-majority sample. The pre-existing 22-guard "poisoned but not wedged"
fixture still passes, so the ratio does not regress the earlier variant.

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

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/ArtiGuardState.kt

@@ -48,12 +48,34 @@ object ArtiGuardState {
     fun parse(json: String): JsonNode = mapper.readTree(json)
 
     /**
-     * True when at least one non-empty guard selection has *zero* usable guards —
-     * every guard permanently `disabled` or dropped from the consensus
-     * (`unlisted_since`). This is the AllGuardsDown wedge: Arti can neither build
-     * circuits nor replenish the sample (it's full of unusable entries), so it
-     * stays broken across restarts until the on-disk state is wiped. A single
-     * usable guard is enough to recover, so this only trips on a total wipeout.
+     * Smallest sample worth judging by ratio. Below this a low usable count is just a young
+     * sample that Arti is still filling, and wiping it would loop the bootstrap.
+     */
+    const val MIN_SAMPLE_TO_JUDGE_RATIO = 10
+
+    /** A sample is wedged when fewer than 1 in [USABLE_RATIO_DIVISOR] of its guards are usable. */
+    const val USABLE_RATIO_DIVISOR = 10
+
+    /**
+     * True when a non-empty guard selection has no usable guards left, or so few that Arti cannot
+     * realistically recover from them — every guard permanently `disabled` or dropped from the
+     * consensus (`unlisted_since`).
+     *
+     * This is the AllGuardsDown wedge: Arti can neither build circuits nor replenish the sample, so
+     * it stays broken across restarts until the on-disk state is wiped.
+     *
+     * ### Why this is not `usable == 0`
+     *
+     * It used to be, on the assumption that "a single usable guard is enough to recover". Observed in
+     * the field: a sample of 60 with **59 disabled and 1 usable**, while Arti rejected all sixty at
+     * runtime (`AllGuardsDown { n_accepted: 0, n_rejected: 60 }`) across repeated app restarts. The
+     * lone survivor was just as unreachable as the rest — useless for recovery, but enough to veto
+     * it, because `usable == 0` never became true. ~87% of relay connections failed indefinitely.
+     *
+     * So the test is proportional: a sample of at least [MIN_SAMPLE_TO_JUDGE_RATIO] whose usable
+     * guards are under 1/[USABLE_RATIO_DIVISOR] of the total is wedged. A small, young sample is
+     * still only judged by the strict `usable == 0` rule, so a legitimate first bootstrap that has
+     * sampled one or two guards is never wiped out from under itself.
      */
     fun hasNoUsableGuards(root: JsonNode): Boolean {
         var wedged = false
@@ -61,7 +83,11 @@ object ArtiGuardState {
             val guards = selection.get("guards") ?: return@forEach
             if (guards.isArray && guards.size() > 0) {
                 val usable = guards.count { !it.isDisabled() && !it.isUnlisted() }
-                if (usable == 0) wedged = true
+                if (usable == 0) {
+                    wedged = true
+                } else if (guards.size() >= MIN_SAMPLE_TO_JUDGE_RATIO && usable * USABLE_RATIO_DIVISOR < guards.size()) {
+                    wedged = true
+                }
             }
         }
         return wedged

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorBackend.kt

@@ -20,6 +20,7 @@
  */
 package com.vitorpamplona.amethyst.ui.tor
 
+import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.StateFlow
 
 /**
@@ -46,4 +47,16 @@ interface TorBackend {
      * this is a `suspend` call.
      */
     suspend fun hasBootstrappedBefore(): Boolean
+
+    /**
+     * Emits when Arti has reported `AllGuardsDown` persistently enough that the guard sample is
+     * considered rotten at runtime, regardless of what `guards.json` claims.
+     *
+     * The on-disk heuristic ([ArtiGuardState.hasNoUsableGuards]) only sees guards Arti has
+     * permanently retired. A guard that is merely unreachable stays "usable" on disk forever, so a
+     * sample can be entirely dead in practice while still looking healthy to that check — which is
+     * exactly the state that survived repeated restarts in the field. This is the runtime half:
+     * Arti itself says every guard was rejected, so believe it.
+     */
+    val guardsDownSignal: Flow<Unit>
 }

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorManager.kt

@@ -250,6 +250,24 @@ class TorManager(
                 if (it is TorServiceStatus.Active) hasEverBootstrapped = true
             }.launchIn(scope)
 
+        // Rotten guard sample while Tor is otherwise UP. The watchdog above only fires on a status
+        // stuck at Connecting, and the on-disk check only sees guards Arti has permanently retired —
+        // so a sample whose guards are all merely *unreachable* falls between them: status reaches
+        // Active, `guards.json` still lists usable entries, and every circuit fails anyway. Observed
+        // in the field surviving repeated restarts with ~87% of relay connections failing. Arti's own
+        // AllGuardsDown log is the only reliable signal, so route it through the same rate-limited
+        // wipe. Always a clean-state reset: the whole point is that the persisted sample is the
+        // problem.
+        service.guardsDownSignal
+            .onEach {
+                val now = nowMs()
+                if (now - lastSelfHealAtMs < SELF_HEAL_COOLDOWN_MS) return@onEach
+                lastSelfHealAtMs = now
+                Log.w("TorManager") { "Arti guard sample rotten at runtime — self-healing (drop client + wipe state)" }
+                service.resetWithCleanState()
+                resetEpoch.update { it + 1 }
+            }.launchIn(scope)
+
         selfHealSignal
             .onEach {
                 val now = nowMs()

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorService.kt

@@ -25,8 +25,12 @@ import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import com.vitorpamplona.quartz.utils.Log
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
@@ -34,6 +38,15 @@ import kotlinx.coroutines.withContext
 import java.io.File
 import java.util.concurrent.atomic.AtomicBoolean
 
+/** Arti's log marker for "no usable guard could be found". */
+private const val ALL_GUARDS_DOWN_MARKER = "AllGuardsDown"
+
+/** How many [ALL_GUARDS_DOWN_MARKER] lines inside [GUARDS_DOWN_WINDOW_MS] mean the sample is rotten. */
+private const val GUARDS_DOWN_THRESHOLD = 40
+
+/** Window for [GUARDS_DOWN_THRESHOLD]. Wide enough that ordinary transient churn never trips it. */
+private const val GUARDS_DOWN_WINDOW_MS = 60_000L
+
 private const val DEFAULT_SOCKS_PORT = 17392
 private const val MAX_PORT_RETRIES = 10
 
@@ -85,6 +98,47 @@ class TorService(
     private val _status = MutableStateFlow<TorServiceStatus>(TorServiceStatus.Off)
     override val status: StateFlow<TorServiceStatus> = _status.asStateFlow()
 
+    /**
+     * Runtime detector for a rotten guard sample. Arti logs [ALL_GUARDS_DOWN_MARKER] every time it
+     * fails to find a usable guard; [GUARDS_DOWN_THRESHOLD] of those inside [GUARDS_DOWN_WINDOW_MS]
+     * means every guard in the sample is unreachable *right now* — which the on-disk check cannot
+     * see, because an unreachable guard is not a `disabled` one (see
+     * [ArtiGuardState.hasNoUsableGuards]).
+     *
+     * Counted here because the log callback is the only place Arti surfaces it; acted on by
+     * [TorManager], which owns the reset cadence and its rate limiting. Both counters are only
+     * touched from Arti's log thread.
+     */
+    private var guardsDownCount = 0
+
+    private var guardsDownWindowStartMs = 0L
+
+    private val _guardsDownSignal =
+        MutableSharedFlow<Unit>(extraBufferCapacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
+
+    override val guardsDownSignal: Flow<Unit> = _guardsDownSignal.asSharedFlow()
+
+    /**
+     * Feeds one Arti log line to the [guardsDownSignal] detector. Cheap by design: this runs for
+     * every log line, and a wedged Arti emits tens of thousands of them.
+     */
+    private fun trackGuardFailures(text: String) {
+        if (!text.contains(ALL_GUARDS_DOWN_MARKER)) return
+
+        val now = System.currentTimeMillis()
+        if (now - guardsDownWindowStartMs > GUARDS_DOWN_WINDOW_MS) {
+            guardsDownWindowStartMs = now
+            guardsDownCount = 0
+        }
+        guardsDownCount++
+        if (guardsDownCount >= GUARDS_DOWN_THRESHOLD) {
+            guardsDownCount = 0
+            guardsDownWindowStartMs = now
+            Log.w("TorService") { "Arti reported $ALL_GUARDS_DOWN_MARKER $GUARDS_DOWN_THRESHOLD times in ${GUARDS_DOWN_WINDOW_MS}ms — guard sample is rotten at runtime" }
+            _guardsDownSignal.tryEmit(Unit)
+        }
+    }
+
     private fun artiDataDir() = File(context.filesDir, "arti")
 
     /** Diagnostic: total bytes of the consensus/descriptor cache, to correlate with bootstrap time. */
@@ -181,6 +235,7 @@ class TorService(
                     // (gated on proxyRunning) intermittently dropped the transition. start()
                     // now sets Active deterministically once the proxy is bound.
                     ArtiNative.setLogCallback { text ->
+                        trackGuardFailures(text)
                         Log.d("TorService") {
                             val newLine = text.indexOf('\n')
                             if (newLine > 1) {

amethyst/src/test/java/com/vitorpamplona/amethyst/ui/tor/ArtiGuardStateTest.kt

@@ -100,6 +100,75 @@ class ArtiGuardStateTest {
         assertTrue(ArtiGuardState.hasConfirmedGuard(root))
     }
 
+    /**
+     * The field case this ratio rule exists for: a 60-guard sample with 59 permanently disabled and
+     * exactly ONE still nominally usable, while Arti rejected all sixty at runtime
+     * (`AllGuardsDown { n_accepted: 0, n_rejected: 60 }`) across repeated app restarts. Under the old
+     * `usable == 0` rule the lone survivor vetoed the wipe forever, so Tor stayed ~87% broken and
+     * Concord never folded.
+     */
+    @Test
+    fun `one usable guard out of sixty is wedged, not healthy`() {
+        val guards =
+            buildString {
+                append("""{ "default": { "guards": [""")
+                repeat(59) {
+                    append("""{ "confirmed_at": "2026-07-01T00:00:00Z", "disabled": { "type": "TooManyIndeterminateFailures" }, "unlisted_since": null },""")
+                }
+                append("""{ "confirmed_at": null, "disabled": null, "unlisted_since": null }""")
+                append("] } }")
+            }
+        val root = ArtiGuardState.parse(guards)
+
+        assertTrue(
+            "1 usable of 60 is unrecoverable in practice → must wipe",
+            ArtiGuardState.hasNoUsableGuards(root),
+        )
+    }
+
+    /**
+     * The opposite guard-rail: a young sample must never be wiped just for being small, or a genuine
+     * first bootstrap would loop. Only the strict `usable == 0` rule applies below
+     * [ArtiGuardState.MIN_SAMPLE_TO_JUDGE_RATIO].
+     */
+    @Test
+    fun `small fresh sample is never wedged by the ratio rule`() {
+        val root =
+            ArtiGuardState.parse(
+                """
+                { "default": { "guards": [
+                    { "confirmed_at": null, "disabled": null, "unlisted_since": null },
+                    { "confirmed_at": null, "disabled": { "type": "TooManyIndeterminateFailures" } },
+                    { "confirmed_at": null, "disabled": { "type": "TooManyIndeterminateFailures" } }
+                ] } }
+                """.trimIndent(),
+            )
+
+        assertFalse(
+            "1 usable of 3 is a young sample Arti is still filling → leave it alone",
+            ArtiGuardState.hasNoUsableGuards(root),
+        )
+    }
+
+    /** A healthy majority must not trip the ratio rule. */
+    @Test
+    fun `mostly usable large sample is not wedged`() {
+        val guards =
+            buildString {
+                append("""{ "default": { "guards": [""")
+                repeat(20) {
+                    append("""{ "confirmed_at": null, "disabled": null, "unlisted_since": null },""")
+                }
+                append("""{ "confirmed_at": null, "disabled": { "type": "TooManyIndeterminateFailures" } }""")
+                append("] } }")
+            }
+
+        assertFalse(
+            "20 usable of 21 is healthy",
+            ArtiGuardState.hasNoUsableGuards(ArtiGuardState.parse(guards)),
+        )
+    }
+
     @Test
     fun `empty selection is neither wedged nor confirmed`() {
         val root = ArtiGuardState.parse("""{ "default": { "guards": [] }, "restricted": { "guards": [] } }""")

amethyst/src/test/java/com/vitorpamplona/amethyst/ui/tor/TorManagerTest.kt

@@ -22,6 +22,8 @@ package com.vitorpamplona.amethyst.ui.tor
 
 import com.vitorpamplona.amethyst.commons.tor.TorType
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
@@ -506,6 +508,11 @@ private class FakeTorBackend : TorBackend {
     /** Simulates a persisted confirmed guard on disk (prior successful bootstrap). */
     var bootstrappedBefore = false
 
+    /** Lets a test fire Arti's "every guard was rejected" signal (see [TorBackend.guardsDownSignal]). */
+    val guardsDown = MutableSharedFlow<Unit>(extraBufferCapacity = 1)
+
+    override val guardsDownSignal: Flow<Unit> = guardsDown
+
     override suspend fun hasBootstrappedBefore(): Boolean = bootstrappedBefore
 
     override suspend fun start() {