app: learn probe durations per device; stop claiming VPN after teardown

The ETA table was fixed at compile time, but real durations are a property
of this phone and the network it stands in - ICMPv6 answers in
milliseconds where IPv6 works and waits out its timeout where it does
not. Estimates now prefer an EMA (70/30) of what this device actually
measured, seeded by the old constants on first run.

VPN detection had two lies in it, both found on hardware: a disconnected
tunnel lingers in allNetworks while tearing down, so 'VPN active' is now
judged from the ACTIVE network only; and any bind failure counted as
'per-network blocked', so a network dying mid-run flipped a healthy run
to INCONCLUSIVE - now only a genuine EPERM refusal counts. Banner and
finding split three ways (VPN + blocked / blocked only / VPN only) so the
app never names a VPN the user just turned off.

App version 0.2.3.

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

docs/build-status.md

@@ -1346,3 +1346,24 @@ The spec-vs-implementation gap audit closed its top items; protocol_version 1.0.
 
 Client-side counterparts still to build: sending 0x03 trains + parsing 0x05 reports
 (`train.udp_updown`), and passing `dscp` on downtrain actions.
+
+## ⚠ Version lineage broken: fmr runs v0.11.2, the repo's tags stop at v0.9.x (2026-08-02)
+
+Discovered while preparing to self-update fmr to the freshly released server-v0.9.2:
+**fmr runs v0.11.2** (binary installed 2026-08-02 08:51), but this repo's remote has tags only
+up to `server-v0.9.1`, master fast-forwarded cleanly from this machine, there is no v0.10/v0.11
+release in Gitea, no source checkout or Go toolchain on fmr, and no deploy script in this repo
+that stamps versions. Conclusion: v0.11.2 was cross-built from a clone whose commits were never
+pushed — presumably another dev machine.
+
+Consequences until resolved:
+- **Do NOT run `--self-update` on fmr.** Gitea's `/releases/latest` is the *newest-created*
+  release, which is now `server-v0.9.2` — semantically older than the deployed binary; the
+  updater compares strings, not SemVer, and would happily "update" v0.11.2 down to it. No
+  automatic risk exists (fmr has no update timer installed, only the cert timer), but a manual
+  run would downgrade.
+- The next real release must be tagged **above v0.11.2** (e.g. `server-v0.11.3` or `v0.12.0`)
+  *after* the missing commits are pushed, so "latest" becomes truly latest again.
+- The unpushed v0.10–v0.11 work needs to be found and pushed from whichever machine built it,
+  or the deployed binary's provenance re-established some other way, before the release channel
+  can be trusted again.

echolot-app/app/build.gradle.kts

@@ -15,7 +15,7 @@ plugins {
 //
 // major*1_000_000 + minor*10_000 + patch*10 leaves room for 9 patch-level rebuilds (the trailing
 // digit) without disturbing the mapping, and stays inside the 2_100_000_000 ceiling until major 2100.
-val appVersionName = "0.2.2"
+val appVersionName = "0.2.3"
 
 fun versionCodeOf(semver: String): Int {
     val (major, minor, patch) = semver.substringBefore('-').split(".").map(String::toInt)

echolot-app/app/src/main/kotlin/app/echo_lot/app/MainActivity.kt

@@ -422,16 +422,28 @@ private fun Results(doc: MeasurementDocument) {
             .mapNotNull { id -> doc.networks.firstOrNull { it.id == id } }
             .joinToString(", ") { it.iface?.takeIf { s -> s.isNotBlank() } ?: it.transport.name.lowercase() }
             .ifBlank { "the networks beneath it" }
+        // Same three-way split as the finding: saying "VPN" when the user just disconnected
+        // theirs (the wall lingers during teardown) reads as the app being wrong, not the OS.
+        val (headline, body) = when {
+            constraints.vpnActive && constraints.perNetworkBlocked ->
+                "Measured through a VPN" to
+                    ("Android does not let apps send on the networks beneath an active VPN, so " +
+                        "$blocked could not be measured — these results describe the tunnel. " +
+                        "Disconnect the VPN and run again to measure the networks themselves.")
+            constraints.perNetworkBlocked ->
+                "Some networks could not be measured" to
+                    ("Android refused sends on $blocked — the restriction a VPN leaves in place " +
+                        "while it tears down. These networks went unmeasured; wait a few " +
+                        "seconds and run again.")
+            else ->
+                "A VPN holds the default route" to
+                    ("Default-route results describe the tunnel; per-network measurements " +
+                        "reached the underlying networks.")
+        }
         Card(colors = CardDefaults.cardColors(containerColor = Color(0xFF3A2E12))) {
             Column(Modifier.fillMaxWidth().padding(12.dp)) {
-                Text("Measured through a VPN", color = Color(0xFFFFD08A),
-                    fontWeight = FontWeight.SemiBold)
-                Text(
-                    "Android does not let apps send on the networks beneath an active VPN, so " +
-                        "$blocked could not be measured — these results describe the tunnel. " +
-                        "Disconnect the VPN and run again to measure the networks themselves.",
-                    fontSize = 12.sp, color = Color(0xFFFFD08A),
-                )
+                Text(headline, color = Color(0xFFFFD08A), fontWeight = FontWeight.SemiBold)
+                Text(body, fontSize = 12.sp, color = Color(0xFFFFD08A))
             }
         }
     }

echolot-app/app/src/main/kotlin/app/echo_lot/app/RunViewModel.kt

@@ -396,7 +396,7 @@ class RunViewModel(app: Application) : AndroidViewModel(app) {
         // What will this run be prevented from measuring? Decided up front, from one throwaway
         // bind per network, so the document can say so instead of leaving it to be inferred from
         // per-test `attempted: false` breadcrumbs (measurement-schema.md §3 `constraints`).
-        runConstraints = app.echo_lot.probe.ConstraintDetector.detect(entries)
+        runConstraints = app.echo_lot.probe.ConstraintDetector.detect(ctx, entries)
 
         val probes: List<Probe> = listOf(
             LinkSnapshotProbe(entries),
@@ -425,10 +425,13 @@ class RunViewModel(app: Application) : AndroidViewModel(app) {
         )
 
         // Plan the run first: the Shizuku battery is counted alongside the app-tier probes so
-        // the bar reflects the whole run. Estimates are per-probe (see Probe.estimatedMs).
-        val shizukuEstimateMs = 8_000L
+        // the bar reflects the whole run. Estimates prefer what THIS device measured on recent
+        // runs (Settings EMA); Probe.estimatedMs is only the cold-start seed — a fixed table
+        // cannot know whether ICMPv6 answers in milliseconds here or waits out its timeout.
+        fun estimateOf(p: Probe) = settings.learnedDurationMs(p.type) ?: p.estimatedMs
+        val shizukuEstimateMs = settings.learnedDurationMs(SHIZUKU_DURATION_KEY) ?: 8_000L
         val totalSteps = probes.size + 1
-        var remainingMs = probes.sumOf { it.estimatedMs } + shizukuEstimateMs
+        var remainingMs = probes.sumOf { estimateOf(it) } + shizukuEstimateMs
         state = state.copy(stepsDone = 0, stepsTotal = totalSteps,
             etaSeconds = ((remainingMs + 999) / 1000).toInt())
 
@@ -448,22 +451,33 @@ class RunViewModel(app: Application) : AndroidViewModel(app) {
                 }
             )
             tests.add(result); collected.add(result)
-            remainingMs -= p.estimatedMs
+            settings.recordDurationMs(p.type, (result.endedMonoNs - result.startedMonoNs) / 1_000_000)
+            remainingMs -= estimateOf(p)
         }
 
-        // Shizuku shell tier — self-degrades to UNSUPPORTED when Shizuku isn't running.
+        // Shizuku shell tier — self-degrades to UNSUPPORTED when Shizuku isn't running. One
+        // battery, three tests: the raw captures plus the parsed ra_source/arp_watch views.
         step("link.ip_monitor (shizuku)", done = probes.size, total = totalSteps, etaMs = shizukuEstimateMs)
-        val shizukuTest = try {
+        val shizukuT0 = System.nanoTime()
+        val shizukuTests = try {
             ShizukuProbe().run(ctx, ids::uuid, ids::monoNs)
         } catch (t: Throwable) {
-            Test(
+            listOf(Test(
                 id = ids.uuid(), type = TestType.LINK_IP_MONITOR, tier = Tier.SHIZUKU,
                 startedMonoNs = ids.monoNs(), endedMonoNs = ids.monoNs(),
                 status = TestStatus.FAILED, error = TestError("uncaught", t.message ?: t.javaClass.simpleName),
-            )
+            ))
+        }
+        // One key for the whole step: the three tests come out of one shell battery, and the
+        // bar plans them as one step.
+        settings.recordDurationMs(SHIZUKU_DURATION_KEY, (System.nanoTime() - shizukuT0) / 1_000_000)
+        tests.addAll(shizukuTests); collected.addAll(shizukuTests)
+        // "Shizuku tier ran" is the battery's verdict — the derived tests can be PARTIAL on a
+        // perfectly healthy shell tier (e.g. an RA-less v4-only link).
+        runShizukuOk = shizukuTests.any {
+            it.type == TestType.LINK_IP_MONITOR &&
+                (it.status == TestStatus.OK || it.status == TestStatus.PARTIAL)
         }
-        tests.add(shizukuTest); collected.add(shizukuTest)
-        runShizukuOk = shizukuTest.status == TestStatus.OK || shizukuTest.status == TestStatus.PARTIAL
 
         return buildDocument(tests)
     }
@@ -566,6 +580,30 @@ class RunViewModel(app: Application) : AndroidViewModel(app) {
             val blocked = runConstraints.unmeasuredNetworks
                 .joinToString(", ") { id -> ifaceOf(networks, id) }
                 .ifBlank { "the underlying networks" }
+            // Three distinct situations share this finding code, and naming the wrong one costs
+            // trust: claiming "a VPN is active" right after the user disconnected theirs is how
+            // this text was first proven wrong on hardware.
+            val (title, description) = when {
+                runConstraints.vpnActive && runConstraints.perNetworkBlocked ->
+                    "A VPN is active — $blocked could not be measured" to
+                        ("Android refuses to let apps send on the networks beneath an active " +
+                            "VPN (that is how it prevents traffic leaking around the tunnel), " +
+                            "so every per-network test here measured the tunnel or nothing. " +
+                            "Nothing in this run says anything about $blocked. To measure them, " +
+                            "disconnect the VPN and run again.")
+                runConstraints.perNetworkBlocked ->
+                    "The OS refused sends on $blocked" to
+                        ("Android denied this app permission to send on $blocked (EPERM on " +
+                            "bind). That is the restriction a VPN imposes on the networks " +
+                            "beneath it — a tunnel disconnected moments ago can still leave it " +
+                            "in place while it tears down. Nothing in this run says anything " +
+                            "about $blocked; wait a few seconds and run again.")
+                else ->
+                    "A VPN holds the default route" to
+                        ("Everything using the default route in this run describes the tunnel, " +
+                            "not the network it rides on. Per-network measurements were " +
+                            "permitted and did measure the underlying networks.")
+            }
             out.add(
                 Finding(
                     id = ids.uuid(),
@@ -573,12 +611,8 @@ class RunViewModel(app: Application) : AndroidViewModel(app) {
                     category = FindingRegistry.MEASUREMENT_VPN_CONSTRAINED.category,
                     severity = FindingRegistry.MEASUREMENT_VPN_CONSTRAINED.severity,
                     confidence = Confidence.HIGH,
-                    title = "A VPN is active — $blocked could not be measured",
-                    description = "Android refuses to let apps send on the networks beneath an " +
-                        "active VPN (that is how it prevents traffic leaking around the tunnel), " +
-                        "so every per-network test here measured the tunnel or nothing. Nothing " +
-                        "in this run says anything about $blocked. To measure them, disconnect " +
-                        "the VPN and run again.",
+                    title = title,
+                    description = description,
                     evidenceRefs = linkEvidence,
                 )
             )
@@ -899,4 +933,9 @@ class RunViewModel(app: Application) : AndroidViewModel(app) {
             etaSeconds = if (etaMs >= 0) ((etaMs + 999) / 1000).toInt() else state.etaSeconds,
         )
     }
+
+    private companion object {
+        /** Duration-learning key for the Shizuku step, which is three tests but one battery. */
+        const val SHIZUKU_DURATION_KEY = "shizuku.battery"
+    }
 }

echolot-app/app/src/main/kotlin/app/echo_lot/app/Settings.kt

@@ -50,6 +50,35 @@ class Settings(context: Context) {
         maxTotalBytes = maxTotalMb.toLong() * 1024 * 1024,
     )
 
+    // ---- run-duration learning ---------------------------------------------------------
+
+    /**
+     * Learned duration of one test type on THIS device, or null before the first run.
+     *
+     * The static Probe.estimatedMs values are only cold-start seeds: real durations depend on
+     * the phone and the network it stands in (ICMPv6 answers in milliseconds where IPv6 works
+     * and waits out full timeouts where it does not), so a fixed table is wrong for almost
+     * everyone almost always. What was measured last time is the only estimate that tracks
+     * reality.
+     */
+    fun learnedDurationMs(type: String): Long? =
+        prefs.getLong("$DURATION_PREFIX$type", -1L).takeIf { it > 0 }
+
+    /**
+     * Feeds one measured duration into the estimate — EMA, 70 % old / 30 % new. Heavy enough
+     * on history that a single odd run (a captive portal stalling DNS) does not whipsaw the
+     * bar, light enough that a real change (enrolling with a server un-skips three probes)
+     * converges within a few runs. Recorded whatever the test's status: a probe that skips in
+     * 2 ms will keep skipping in 2 ms until circumstances change, and then the EMA follows.
+     */
+    fun recordDurationMs(type: String, ms: Long) {
+        if (ms < 0) return
+        val key = "$DURATION_PREFIX$type"
+        val old = prefs.getLong(key, -1L)
+        val next = if (old <= 0) ms else (old * 7 + ms * 3) / 10
+        prefs.edit().putLong(key, next).apply()
+    }
+
     // ---- upload ----------------------------------------------------------------------
 
     /**
@@ -215,5 +244,6 @@ class Settings(context: Context) {
         const val PENDING_STATE = "pending_auth_state"
         const val ACCOUNT_NAME = "account_name"
         const val ACCOUNT_ID = "account_id"
+        const val DURATION_PREFIX = "duration_ms."
     }
 }

echolot-app/core-engine/src/main/kotlin/app/echo_lot/engine/DownstreamMeasurement.kt

@@ -168,6 +168,10 @@ class DownstreamMeasurement(private val ids: IdSource) {
         trainCount: Int = 100,
         trainSizeBytes: Int = 300,
         trainIntervalUs: Int = 3_000,
+        // DSCP to mark the train with (0-63), or -1 to leave packets unmarked. Pairing a marked
+        // downtrain with the server-observed DSCP of an upstream train is the two-direction
+        // sec.dscp_ecn_survival measurement.
+        trainDscp: Int = -1,
     ): Pair<List<Test>, List<Finding>> {
         val tests = ArrayList<Test>()
         val findings = ArrayList<Finding>()
@@ -175,7 +179,8 @@ class DownstreamMeasurement(private val ids: IdSource) {
         val df = bigSend(credential, sessionId, control, probe, sessionRef, sizes, df = true)
         val frag = bigSend(credential, sessionId, control, probe, sessionRef, sizes, df = false)
         val train = downTrain(
-            credential, sessionId, control, probe, sessionRef, trainCount, trainSizeBytes, trainIntervalUs,
+            credential, sessionId, control, probe, sessionRef, trainCount, trainSizeBytes,
+            trainIntervalUs, trainDscp,
         )
 
         tests.add(df.test); tests.add(frag.test); tests.add(train.test)
@@ -338,15 +343,19 @@ class DownstreamMeasurement(private val ids: IdSource) {
 
     private fun downTrain(
         credential: String, sessionId: String, control: ControlClient, probe: ProbeSession,
-        sessionRef: String, count: Int, sizeBytes: Int, intervalUs: Int,
+        sessionRef: String, count: Int, sizeBytes: Int, intervalUs: Int, dscp: Int = -1,
     ): TrainResult {
         val testId = ids.uuid()
         val started = ids.monoNs()
 
+        // dscp is only sent when requested: an older server rejects unknown-value problems
+        // louder than absent keys, and unmarked is the correct default for a plain loss train.
+        val dscpField = if (dscp in 0..63) ""","dscp":$dscp""" else ""
         val reply = runCatching {
             control.action(
                 credential, sessionId,
-                """{"action":"downtrain","count":$count,"size_bytes":$sizeBytes,"interval_us":$intervalUs}""",
+                """{"action":"downtrain","count":$count,"size_bytes":$sizeBytes,""" +
+                    """"interval_us":$intervalUs$dscpField}""",
             )
         }
         if (reply.isFailure) {
@@ -394,6 +403,12 @@ class DownstreamMeasurement(private val ids: IdSource) {
                 interArrivalMsAvg = interArrival.average().takeIf { interArrival.isNotEmpty() }?.let(::round1),
                 interArrivalMsMax = interArrival.maxOrNull()?.let(::round1),
                 sendIntervalUs = intervalUs,
+                dscpRequested = dscp.takeIf { it in 0..63 },
+                // The server says whether it could actually mark (dscp_applied); recorded so a
+                // survival comparison never blames the path for a marking the sender skipped.
+                dscpApplied = reply.getOrNull()?.let {
+                    Regex("\"dscp_applied\"\\s*:\\s*(true|false)").find(it)?.groupValues?.get(1)?.toBoolean()
+                },
             ),
         ) as JsonObject
 
@@ -490,4 +505,6 @@ data class DownTrainMetrics(
     @SerialName("inter_arrival_ms_avg") val interArrivalMsAvg: Double? = null,
     @SerialName("inter_arrival_ms_max") val interArrivalMsMax: Double? = null,
     @SerialName("send_interval_us") val sendIntervalUs: Int,
+    @SerialName("dscp_requested") val dscpRequested: Int? = null,
+    @SerialName("dscp_applied") val dscpApplied: Boolean? = null,
 )

echolot-app/core-engine/src/main/kotlin/app/echo_lot/engine/ServerMeasurement.kt

@@ -88,6 +88,14 @@ class ServerMeasurement(
             tests.add(test)
             allFindings.addAll(findings)
 
+            // The upstream train needs no grant and no capability beyond udp-probe itself; a
+            // server that predates trains simply never answers the report request, which the
+            // measurement reports as exactly that ambiguity rather than as network loss.
+            val (utTest, utFindings) = UpstreamTrainMeasurement(ids)
+                .run(ps, sessionRef = "sess-1")
+            tests.add(utTest)
+            allFindings.addAll(utFindings)
+
             // Downstream needs a session the server has already seen traffic from — the echo
             // train just provided that — and a server that advertises the grants. Skipped
             // quietly against an older server rather than reported as a failure of the network.

echolot-app/core-engine/src/main/kotlin/app/echo_lot/engine/UpstreamTrainMeasurement.kt

@@ -0,0 +1,159 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package app.echo_lot.engine
+
+import app.echo_lot.measurement.*
+import app.echo_lot.protocol.ProbeSession
+import kotlinx.serialization.SerialName
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.encodeToJsonElement
+
+/**
+ * train.udp_updown — the client sends a paced train (types 0x03), then asks the server what
+ * arrived (0x04 → 0x05) and lines both views up per sequence number.
+ *
+ * This is the measurement a round trip cannot make: an echo run only says "lost somewhere", the
+ * train's two ledgers say lost on the way OUT, specifically, because the server's report names
+ * exactly which sequence numbers reached it. The downstream direction has its own test
+ * (train.udp_downstream) under a grant; this one needs none, since the client generates all the
+ * traffic itself.
+ *
+ * The evidence is the schema's columnar TrainEvidence: one index per sent packet, with the
+ * server-side columns null where a packet never arrived. Server timestamps are on the server's
+ * own clock — only differences within that clock mean anything unless time.server_offset maps
+ * them (two-clock rule).
+ */
+class UpstreamTrainMeasurement(private val ids: IdSource) {
+
+    private val json = Json { encodeDefaults = true; explicitNulls = true }
+
+    fun run(
+        probe: ProbeSession,
+        sessionRef: String,
+        count: Int = 200,
+        sizeBytes: Int = 200,
+        interPacketMs: Long = 5,
+    ): Pair<Test, List<Finding>> {
+        val testId = ids.uuid()
+        val started = ids.monoNs()
+        // The id only needs to be unique within this session; a clash across sessions is
+        // meaningless because trains are buffered per session on the server.
+        val trainId = (System.nanoTime() and 0x7FFFFFFF).toInt()
+
+        val sent = probe.sendTrain(trainId, count, sizeBytes, interPacketMs)
+        // Let the tail arrive before asking for the ledger; packets still in flight when the
+        // report is cut would read as upstream loss.
+        Thread.sleep(300)
+        val report = probe.trainReport(trainId)
+
+        if (report == null) {
+            return Test(
+                id = testId, type = TestType.TRAIN_UDP_UPDOWN, sessionRef = sessionRef,
+                tier = Tier.APP, startedMonoNs = started, endedMonoNs = ids.monoNs(),
+                status = TestStatus.FAILED,
+                // Honest ambiguity: an old server drops 0x04 silently, and a lost report looks
+                // identical from here. Neither says anything about the train itself.
+                error = TestError(
+                    "no_report",
+                    "no train report arrived — the report was lost, or the server predates trains",
+                ),
+            ) to emptyList()
+        }
+
+        val bySeq = report.rows.associateBy { it.seq }
+        fun col255(v: Int): Int? = v.takeIf { it != 255 } // 255 = "not observed" on the wire
+
+        val evidence = TrainEvidence(
+            epochMonoNs = started,
+            seq = sent.map { it.seq },
+            tTxNs = sent.map { it.tTxNs },
+            tSrvRxNs = sent.map { bySeq[it.seq]?.tRxNs },
+            tRxNs = sent.map { null }, // upstream only: nothing comes back per packet
+            sizeBytes = sent.map { it.sizeBytes },
+            ttlSeenByServer = sent.map { bySeq[it.seq]?.let { r -> col255(r.ttl) } },
+            dscpSeenByServer = sent.map { bySeq[it.seq]?.let { r -> col255(r.dscp) } },
+            ecnSeenByServer = sent.map { bySeq[it.seq]?.let { r -> col255(r.ecn) } },
+            evidenceTruncated = report.truncated,
+        ).toEvidence()
+
+        // Loss against the server's total count, not its row list: rows past the server's buffer
+        // cap are counted but not kept, and treating them as lost would invent loss exactly on
+        // the biggest trains.
+        val lossPct = if (sent.isEmpty()) 0.0 else {
+            (sent.size - report.received).coerceAtLeast(0) * 100.0 / sent.size
+        }
+        val metrics = json.encodeToJsonElement(
+            UpstreamTrainMetrics(
+                sent = sent.size,
+                receivedByServer = report.received,
+                lossPct = round1(lossPct),
+                reportPartsExpected = report.partsExpected,
+                reportPartsReceived = report.partsReceived,
+                truncated = report.truncated,
+            ),
+        ) as JsonObject
+
+        val findings = ArrayList<Finding>()
+        if (sent.isNotEmpty() && report.received == 0) {
+            findings.add(
+                Finding(
+                    id = ids.uuid(),
+                    code = FindingRegistry.UDP_UNREACHABLE_UPSTREAM.code,
+                    category = FindingRegistry.UDP_UNREACHABLE_UPSTREAM.category,
+                    severity = FindingRegistry.UDP_UNREACHABLE_UPSTREAM.severity,
+                    confidence = Confidence.HIGH,
+                    title = "The server received none of ${sent.size} upstream packets",
+                    description = "Every train packet vanished on the way out, while the " +
+                        "report request's reply made it back — the outbound path drops this " +
+                        "traffic, the return path works.",
+                    evidenceRefs = listOf(EvidenceRef(testId)),
+                ),
+            )
+        } else if (lossPct >= 2.0) {
+            findings.add(
+                Finding(
+                    id = ids.uuid(),
+                    code = FindingRegistry.LOSS_UPSTREAM.code,
+                    category = FindingRegistry.LOSS_UPSTREAM.category,
+                    severity = FindingRegistry.LOSS_UPSTREAM.severity,
+                    confidence = Confidence.HIGH,
+                    title = "Upstream loss of ${round1(lossPct)} %",
+                    description = "The server received ${report.received} of the ${sent.size} " +
+                        "packets this device sent, and its per-sequence ledger names the " +
+                        "missing ones. This is outbound loss specifically; the return path " +
+                        "delivered the report.",
+                    evidenceRefs = listOf(EvidenceRef(testId)),
+                ),
+            )
+        }
+
+        val status = when {
+            report.received == 0 && sent.isNotEmpty() -> TestStatus.FAILED
+            report.partsReceived < report.partsExpected -> TestStatus.PARTIAL
+            else -> TestStatus.OK
+        }
+        return Test(
+            id = testId, type = TestType.TRAIN_UDP_UPDOWN, sessionRef = sessionRef, tier = Tier.APP,
+            startedMonoNs = started, endedMonoNs = ids.monoNs(),
+            status = status, evidence = evidence, metrics = metrics,
+        ) to findings
+    }
+
+    private fun round1(v: Double) = Math.round(v * 10.0) / 10.0
+}
+
+/** Metrics for train.udp_updown. */
+@Serializable
+data class UpstreamTrainMetrics(
+    val sent: Int,
+    /** The server's total count — includes packets past its row buffer (counted, not listed). */
+    @SerialName("received_by_server") val receivedByServer: Int,
+    @SerialName("loss_pct") val lossPct: Double,
+    @SerialName("report_parts_expected") val reportPartsExpected: Int,
+    @SerialName("report_parts_received") val reportPartsReceived: Int,
+    /** The server's row buffer overflowed: rows are a sample, the count is still complete. */
+    val truncated: Boolean,
+)

echolot-app/core-engine/src/test/kotlin/app/echo_lot/engine/LiveUpstreamTrainTest.kt

@@ -0,0 +1,65 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package app.echo_lot.engine
+
+import app.echo_lot.measurement.TestStatus
+import app.echo_lot.protocol.ControlClient
+import app.echo_lot.protocol.ProbeSession
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
+
+/**
+ * Upstream train (types 0x03-0x05) against a LIVE server. Self-skips without ECHOLOT_LIVE_*.
+ *
+ * What is asserted is the ledger property: the server's report must account for what was sent,
+ * per sequence number, because directional loss attribution is the entire reason trains exist —
+ * a test that only checked "a report came back" would pass against a server that counts nothing.
+ */
+class LiveUpstreamTrainTest {
+
+    private val url = System.getenv("ECHOLOT_LIVE_URL")
+    private val pin = System.getenv("ECHOLOT_LIVE_PIN")
+    private val cred = System.getenv("ECHOLOT_LIVE_CRED")
+    private val udp = System.getenv("ECHOLOT_LIVE_UDP")
+    private val target = System.getenv("ECHOLOT_LIVE_TARGET") ?: "fmr"
+
+    @Test
+    fun serverLedgerAccountsForTheTrain() {
+        if (url == null || pin == null || cred == null || udp == null) {
+            println("LiveUpstreamTrainTest skipped (no ECHOLOT_LIVE_* env)"); return
+        }
+        val control = ControlClient(url, setOf(pin), "0.2.0")
+        val session = control.createSession(cred, target)
+        val (host, port) = udp.split(":").let { it[0] to it[1].toInt() }
+
+        val (test, findings) = ProbeSession(cred, session, host, port).use { ps ->
+            ps.echo() // prime the session so its source is known
+            UpstreamTrainMeasurement(SystemIdSource()).run(
+                ps, sessionRef = "sess-1", count = 120, sizeBytes = 200, interPacketMs = 3,
+            )
+        }
+        control.deleteSession(cred, session.sessionId)
+
+        val m = assertNotNull(test.metrics).toString()
+        println("updown: ${test.status} $m")
+        for (f in findings) println("finding ${f.code} [${f.severity}] ${f.title}")
+
+        assertEquals(TestStatus.OK, test.status, "train report incomplete or absent: $m")
+
+        val sent = Regex(""""sent":(\d+)""").find(m)?.groupValues?.get(1)?.toInt()
+        val received = Regex(""""received_by_server":(\d+)""").find(m)?.groupValues?.get(1)?.toInt()
+        assertNotNull(sent); assertNotNull(received)
+        assertTrue(sent > 0, "nothing was sent: $m")
+        // Over a working path the ledger must be near-complete; a lossy wifi may drop a few, but
+        // a server that fails to count would show up as massive phantom loss here.
+        assertTrue(received >= sent * 9 / 10, "server counted $received of $sent: $m")
+
+        // The columnar evidence must carry a server timestamp for arrived packets — that column
+        // is what one-way delay math consumes after timesync.
+        val ev = assertNotNull(test.evidence).toString()
+        assertTrue(ev.contains("t_srv_rx_ns"), "no server rx column in evidence")
+    }
+}

echolot-app/core-probe/src/main/kotlin/app/echo_lot/probe/ConstraintDetector.kt

@@ -3,6 +3,11 @@
 
 package app.echo_lot.probe
 
+import android.content.Context
+import android.net.ConnectivityManager
+import android.net.NetworkCapabilities
+import android.system.ErrnoException
+import android.system.OsConstants
 import app.echo_lot.measurement.Constraints
 import app.echo_lot.measurement.Transport
 import java.net.DatagramSocket
@@ -20,22 +25,44 @@ import java.net.DatagramSocket
  */
 object ConstraintDetector {
 
-    fun detect(entries: List<NetworkInventory.Entry>): Constraints {
-        val vpnActive = entries.any { it.model.transport == Transport.VPN }
-        val unmeasured = ArrayList<String>()
+    fun detect(ctx: Context, entries: List<NetworkInventory.Entry>): Constraints {
+        val cm = ctx.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
+        // "A VPN holds the default route" is judged from the ACTIVE network, not from a VPN
+        // network merely existing in the list: a tunnel that was just disconnected lingers in
+        // allNetworks while it tears down, and counting it kept the app claiming "measured
+        // through a VPN" after the VPN was gone.
+        val vpnActive = runCatching {
+            cm.getNetworkCapabilities(cm.activeNetwork)
+                ?.hasTransport(NetworkCapabilities.TRANSPORT_VPN) == true
+        }.getOrDefault(false)
+
+        val refused = ArrayList<String>()
         for (e in entries) {
             // The tunnel itself stays bindable — it is the underlying networks the OS walls off.
             if (e.model.transport == Transport.VPN) continue
-            val bindable = runCatching {
+            val err = try {
                 DatagramSocket().use { s -> e.handle.bindSocket(s) }
-                true
-            }.getOrDefault(false)
-            if (!bindable) unmeasured.add(e.model.id)
+                null
+            } catch (t: Throwable) {
+                t
+            }
+            // Only the OS *refusing* counts as blocked (EPERM: the VPN wall). A network that
+            // happens to die mid-snapshot fails its bind too, but with a different errno, and
+            // calling that "per-network probing blocked" would flip a whole healthy run to
+            // INCONCLUSIVE over one network going away — the probes already record
+            // attempted:false for that case.
+            if (err != null && isPermissionRefusal(err)) refused.add(e.model.id)
         }
         return Constraints(
             vpnActive = vpnActive,
-            perNetworkBlocked = unmeasured.isNotEmpty(),
-            unmeasuredNetworks = unmeasured,
+            perNetworkBlocked = refused.isNotEmpty(),
+            unmeasuredNetworks = refused,
         )
     }
+
+    private fun isPermissionRefusal(t: Throwable): Boolean =
+        generateSequence(t) { it.cause }.any {
+            (it is ErrnoException && it.errno == OsConstants.EPERM) ||
+                (it.message?.contains("EPERM") == true)
+        }
 }

echolot-app/core-protocol/src/main/kotlin/app/echo_lot/protocol/ProbeSession.kt

@@ -152,6 +152,149 @@ class ProbeSession(
     /** What one upstream run put on the wire locally. */
     data class Sent(val packets: Int, val bytes: Long, val durationMs: Long, val kbps: Int)
 
+    // ---- upstream trains (spec §3.2, types 0x03-0x05) --------------------------------------
+
+    /** One TRAIN_DATA packet as sent: its wire seq, local tx time and size. */
+    data class TrainPacket(val seq: Int, val tTxNs: Long, val sizeBytes: Int)
+
+    /** One row of the server's received view. 255 in ttl/dscp/ecn means "not observed". */
+    data class TrainRow(
+        val seq: Int, val tRxNs: Long, val sizeBytes: Int,
+        val ttl: Int, val dscp: Int, val ecn: Int,
+    )
+
+    /**
+     * The server's account of one train. [received] counts every packet that arrived, buffered
+     * or not; [truncated] mirrors the wire flag (rows beyond the server's cap were counted but
+     * not kept). [partsExpected]/[partsReceived] make a lossy report path visible instead of
+     * letting missing rows masquerade as train loss.
+     */
+    data class TrainReport(
+        val trainId: Int,
+        val received: Int,
+        val truncated: Boolean,
+        val rows: List<TrainRow>,
+        val partsExpected: Int,
+        val partsReceived: Int,
+    )
+
+    /**
+     * Sends one paced upstream train. Nothing comes back per packet by design; pair with
+     * [trainReport] to learn what arrived. The absolute schedule (not sleep-per-packet) is the
+     * same anti-drift choice as [sendThroughput].
+     */
+    fun sendTrain(trainId: Int, count: Int, sizeBytes: Int = 200, interPacketMs: Long = 5): List<TrainPacket> {
+        val size = sizeBytes.coerceIn(Wire.HEADER_SIZE + 4, 1472)
+        val out = ArrayList<TrainPacket>(count)
+        val start = System.nanoTime()
+        var next = start
+        for (i in 0 until count) {
+            val payload = ByteArray(size - Wire.HEADER_SIZE)
+            payload[0] = (trainId ushr 24).toByte(); payload[1] = (trainId ushr 16).toByte()
+            payload[2] = (trainId ushr 8).toByte(); payload[3] = trainId.toByte()
+            val tTx = nowNs()
+            val pkt = Wire.build(Wire.TYPE_TRAIN_DATA, prefix, ++seq, tTx, key, payload)
+            try {
+                socket.send(DatagramPacket(pkt, pkt.size, server))
+            } catch (e: java.io.IOException) {
+                // A local send failure is our condition, not the path's: report what actually
+                // left rather than letting the server's report read as loss.
+                break
+            }
+            out.add(TrainPacket(seq, tTx, pkt.size))
+            next += interPacketMs * 1_000_000
+            val sleepNs = next - System.nanoTime()
+            if (sleepNs > 0) Thread.sleep(sleepNs / 1_000_000, (sleepNs % 1_000_000).toInt())
+        }
+        return out
+    }
+
+    /**
+     * Fetches the server's received view of a train (one REPORT_REQ, N REPORT datagrams).
+     *
+     * Returns null when no report arrives at all — indistinguishable between "report lost" and
+     * "server predates trains", and the caller must say so rather than choose. Missing parts of
+     * a multi-part report are tolerated and visible via partsReceived < partsExpected.
+     */
+    fun trainReport(trainId: Int, timeoutMs: Long = 3_000): TrainReport? {
+        val req = ByteArray(4)
+        req[0] = (trainId ushr 24).toByte(); req[1] = (trainId ushr 16).toByte()
+        req[2] = (trainId ushr 8).toByte(); req[3] = trainId.toByte()
+        val pkt = Wire.build(Wire.TYPE_TRAIN_REPORT_REQ, prefix, ++seq, nowNs(), key, req)
+        socket.send(DatagramPacket(pkt, pkt.size, server))
+
+        var received = 0
+        var truncated = false
+        var partsExpected = -1
+        val seenParts = HashSet<Int>()
+        val rows = ArrayList<TrainRow>()
+        val deadline = System.nanoTime() + timeoutMs * 1_000_000
+        val buf = ByteArray(2048)
+        val prevTimeout = socket.soTimeout
+        try {
+            while (partsExpected < 0 || seenParts.size < partsExpected) {
+                val remainMs = ((deadline - System.nanoTime()) / 1_000_000).toInt()
+                if (remainMs <= 0) break
+                socket.soTimeout = remainMs.coerceAtMost(1000)
+                val dp = DatagramPacket(buf, buf.size)
+                try {
+                    socket.receive(dp)
+                } catch (e: java.net.SocketTimeoutException) {
+                    continue
+                }
+                val p = Wire.parseVerified(buf, dp.length, key) ?: continue
+                if (p.type != Wire.TYPE_TRAIN_REPORT) continue
+                val part = parseReportPart(p.payload, trainId) ?: continue
+                if (!seenParts.add(part.part)) continue
+                received = part.received
+                truncated = truncated || part.truncated
+                partsExpected = part.parts
+                rows.addAll(part.rows)
+            }
+        } finally {
+            socket.soTimeout = prevTimeout
+        }
+        if (seenParts.isEmpty()) return null
+        rows.sortBy { it.seq }
+        return TrainReport(trainId, received, truncated, rows, partsExpected, seenParts.size)
+    }
+
+    private class ReportPart(
+        val part: Int, val parts: Int, val received: Int,
+        val truncated: Boolean, val rows: List<TrainRow>,
+    )
+
+    /** Mirrors the server's columnar layout (dataplane/train.go buildTrainReport). */
+    private fun parseReportPart(b: ByteArray, wantId: Int): ReportPart? {
+        if (b.size < 16) return null
+        fun u16(off: Int) = ((b[off].toInt() and 0xFF) shl 8) or (b[off + 1].toInt() and 0xFF)
+        fun u32(off: Int) = ((b[off].toLong() and 0xFF) shl 24) or ((b[off + 1].toLong() and 0xFF) shl 16) or
+            ((b[off + 2].toLong() and 0xFF) shl 8) or (b[off + 3].toLong() and 0xFF)
+        if (u32(0).toInt() != wantId) return null
+        val received = u32(4).toInt()
+        val part = u16(8)
+        val parts = u16(10)
+        val truncated = (b[12].toInt() and 0x01) != 0
+        val n = u16(14)
+        if (b.size < 16 + n * 17) return null
+        val rows = ArrayList<TrainRow>(n)
+        var off = 16
+        val seqs = IntArray(n) { u32(off + it * 4).toInt() }; off += n * 4
+        val tRx = LongArray(n) {
+            var v = 0L
+            for (j in 0 until 8) v = (v shl 8) or (b[off + it * 8 + j].toLong() and 0xFF)
+            v
+        }; off += n * 8
+        val sizes = IntArray(n) { u16(off + it * 2) }; off += n * 2
+        val ttls = IntArray(n) { b[off + it].toInt() and 0xFF }; off += n
+        val dscps = IntArray(n) { b[off + it].toInt() and 0xFF }; off += n
+        val ecns = IntArray(n) { b[off + it].toInt() and 0xFF }
+        for (i in 0 until n) {
+            rows.add(TrainRow(seqs[i], tRx[i], sizes[i], ttls[i], dscps[i], ecns[i]))
+        }
+        return ReportPart(part, parts, received, truncated, rows)
+    }
+
     /** One packet received from the server, with the wire size actually delivered. */
     data class Received(val type: Int, val seq: Int, val sizeBytes: Int, val tRxNs: Long)
 

echolot-app/core-protocol/src/main/kotlin/app/echo_lot/protocol/Wire.kt

@@ -23,6 +23,14 @@ object Wire {
 
     const val TYPE_ECHO_REQ: Int = 0x01
     const val TYPE_ECHO_RESP: Int = 0x02
+    /**
+     * Upstream train (spec §3.2): DATA is deliberately unanswered — a per-packet reply would
+     * double the traffic and drag the return path into a measurement of the outbound one. The
+     * server's received view comes back afterwards via REPORT_REQ → one or more REPORTs.
+     */
+    const val TYPE_TRAIN_DATA: Int = 0x03
+    const val TYPE_TRAIN_REPORT_REQ: Int = 0x04
+    const val TYPE_TRAIN_REPORT: Int = 0x05
     const val TYPE_TIMESYNC_REQ: Int = 0x07
     const val TYPE_TIMESYNC_RSP: Int = 0x08
     const val TYPE_MTU_PROBE: Int = 0x09

echolot-app/core-shizuku/build.gradle.kts

@@ -32,4 +32,8 @@ dependencies {
     implementation(libs.shizuku.provider)
     implementation(libs.kotlinx.coroutines.android)
     implementation(libs.kotlinx.serialization.json)
+    // JVM unit tests for the pure dump parsers (DumpParsers.kt) against the archived
+    // vendor fixtures — no device, no Android runtime.
+    testImplementation(libs.kotlin.test.junit)
+    testImplementation(libs.junit4)
 }

echolot-app/core-shizuku/src/main/kotlin/app/echo_lot/shizuku/DumpParsers.kt

@@ -0,0 +1,174 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package app.echo_lot.shizuku
+
+/**
+ * Parsed view of one IPv6 default route from `ip -6 route show table all`.
+ *
+ * [table] stays a string: Android's per-network route tables use ids past Int range (the Lenovo
+ * TB330FU prints `table 1000000015`), and `table local` is not a number at all — parsing to a
+ * numeric type either overflows or silently drops rows, and the id is only ever compared, never
+ * computed with.
+ */
+data class V6DefaultRoute(
+    /** Link-local address of the advertising router; null for gateway-less defaults (dummy0). */
+    val gateway: String?,
+    val dev: String,
+    val table: String?, // null = main table (`ip` omits the token there)
+    val proto: String?, // "ra" marks a route installed from a Router Advertisement
+    val metric: Long?,
+    /** Remaining RA route lifetime (`expires NNNsec`); null when the route does not age out. */
+    val expiresSec: Long?,
+)
+
+/** One `ip neigh show` row. [lladdr] is null for FAILED/INCOMPLETE entries — the kernel tried to
+ *  resolve and has nothing, which is itself signal. */
+data class NeighborEntry(
+    val ip: String,
+    val dev: String?,
+    val lladdr: String?,
+    val state: String?, // REACHABLE/STALE/FAILED/... — kept verbatim, the kernel's vocabulary
+    val router: Boolean,
+)
+
+/** A NEIGH transition seen inside the `ip monitor` window. */
+data class NeighborEvent(val entry: NeighborEntry, val deleted: Boolean)
+
+/**
+ * Pure-string parsers for the shell battery's `ip` command outputs. No Android imports on
+ * purpose: these run (and are unit-tested) on the JVM against the real vendor dumps archived
+ * from the prober, which is the only way to catch a vendor format drift before it ships.
+ *
+ * All parsers degrade to an empty result on missing or unrecognized input — the battery's
+ * captures are best-effort (the Lenovo's `ip monitor` times out under newProcess, the
+ * UserService path prepends a stray `uid=2000` line, evidence strings are trimmed mid-line
+ * at 1200 chars), so an exception here would turn a degraded capture into a lost test.
+ */
+object DumpParsers {
+
+    /** True when [raw] is real command output rather than an executor error sentinel. */
+    fun captureUsable(raw: String?): Boolean {
+        if (raw.isNullOrBlank()) return false
+        val t = raw.trimStart()
+        return !t.startsWith("SHIZUKU_") && !t.startsWith("EXEC_") && !t.startsWith("NEWPROCESS_")
+    }
+
+    /** Extracts every `default …` route from `ip -6 route show table all` output. */
+    fun parseV6DefaultRoutes(raw: String?): List<V6DefaultRoute> {
+        if (!captureUsable(raw)) return emptyList()
+        val routes = ArrayList<V6DefaultRoute>()
+        for (line in raw!!.lineSequence()) {
+            val tok = line.trim().split(WS)
+            if (tok.firstOrNull() != "default") continue
+            var gateway: String? = null; var dev: String? = null; var table: String? = null
+            var proto: String? = null; var metric: Long? = null; var expires: Long? = null
+            var i = 1
+            while (i < tok.size - 1) {
+                when (tok[i]) {
+                    "via" -> gateway = tok[i + 1]
+                    "dev" -> dev = tok[i + 1]
+                    "table" -> table = tok[i + 1]
+                    "proto" -> proto = tok[i + 1]
+                    "metric" -> metric = tok[i + 1].toLongOrNull()
+                    // `expires 1269sec` — the unit is glued to the number.
+                    "expires" -> expires = tok[i + 1].removeSuffix("sec").toLongOrNull()
+                }
+                i++
+            }
+            // A default route without a device is not something `ip` prints; treat it as a
+            // truncated/garbled line rather than fabricating a partial route.
+            if (dev != null) routes.add(V6DefaultRoute(gateway, dev, table, proto, metric, expires))
+        }
+        return routes
+    }
+
+    /**
+     * Maps interface name → link-layer address from `ip addr show`. Only `link/ether` counts:
+     * loopback/ipip/gre pseudo-addresses are not identities, and the RA-source cross-reference
+     * this feeds compares Ethernet MACs.
+     */
+    fun parseInterfaceMacs(raw: String?): Map<String, String> {
+        if (!captureUsable(raw)) return emptyMap()
+        val macs = LinkedHashMap<String, String>()
+        var current: String? = null
+        for (line in raw!!.lineSequence()) {
+            val header = STANZA_HEADER.find(line)
+            if (header != null) {
+                // "5: tunl0@NONE:" — the name is the part before an optional @suffix.
+                current = header.groupValues[1].substringBefore('@')
+                continue
+            }
+            val dev = current ?: continue
+            val tok = line.trim().split(WS)
+            if (tok.size >= 2 && tok[0] == "link/ether" && MAC.matches(tok[1])) {
+                macs.putIfAbsent(dev, tok[1])
+            }
+        }
+        return macs
+    }
+
+    /** Parses `ip neigh show` output into entries; non-neighbor lines (uid noise) are skipped. */
+    fun parseNeighbors(raw: String?): List<NeighborEntry> {
+        if (!captureUsable(raw)) return emptyList()
+        return raw!!.lineSequence()
+            .mapNotNull { parseNeighborTokens(it.trim().split(WS)) }
+            .toList()
+    }
+
+    /**
+     * Extracts NEIGH transitions from an `ip monitor all` capture. Each event line carries a
+     * `[NEIGH]` label (other families — ROUTE, ADDR, LINK — are ignored) and deletions are
+     * printed as `Deleted <entry>`. An empty result is normal: a quiet 5 s window sees nothing.
+     */
+    fun parseNeighborEvents(raw: String?): List<NeighborEvent> {
+        if (!captureUsable(raw)) return emptyList()
+        val events = ArrayList<NeighborEvent>()
+        for (line in raw!!.lineSequence()) {
+            val m = MONITOR_LABEL.find(line.trim()) ?: continue
+            if (!m.groupValues[1].equals("NEIGH", ignoreCase = true)) continue
+            var rest = line.trim().removeRange(m.range).trim()
+            val deleted = rest.startsWith("Deleted ", ignoreCase = true)
+            if (deleted) rest = rest.substring("Deleted ".length)
+            parseNeighborTokens(rest.split(WS))?.let { events.add(NeighborEvent(it, deleted)) }
+        }
+        return events
+    }
+
+    /**
+     * (ip → lladdr) for every neighbor that has one. This is the comparison surface the future
+     * gateway-MAC-change finding diffs across runs, so it is computed here — in the tested,
+     * pure layer — rather than re-derived from JSON by each consumer.
+     */
+    fun lladdrByIp(neighbors: List<NeighborEntry>): Map<String, String> =
+        neighbors.mapNotNull { n -> n.lladdr?.let { n.ip to it } }.toMap()
+
+    /** One neighbor row: `<ip> dev <if> [lladdr <mac>] [router] [proxy] <STATE>`. */
+    private fun parseNeighborTokens(tok: List<String>): NeighborEntry? {
+        val ip = tok.firstOrNull() ?: return null
+        // The first token must look like an address — this is what drops the UserService path's
+        // stray "uid=2000" line and any grep noise without needing to know every noise shape.
+        if (!IP_LIKE.matches(ip) || (!ip.contains('.') && !ip.contains(':'))) return null
+        var dev: String? = null; var lladdr: String? = null; var state: String? = null
+        var router = false
+        var i = 1
+        while (i < tok.size) {
+            when (tok[i]) {
+                "dev" -> { dev = tok.getOrNull(i + 1); i++ }
+                "lladdr" -> { lladdr = tok.getOrNull(i + 1); i++ }
+                "router" -> router = true
+                "proxy" -> {} // recorded nowhere: proxy entries have no bearing on ARP watching
+                else -> if (STATE.matches(tok[i])) state = tok[i]
+            }
+            i++
+        }
+        return NeighborEntry(ip, dev, lladdr, state, router)
+    }
+
+    private val WS = Regex("\\s+")
+    private val STANZA_HEADER = Regex("^\\d+:\\s+([^:\\s]+):")
+    private val MAC = Regex("^[0-9a-fA-F]{2}(:[0-9a-fA-F]{2}){5}$")
+    private val IP_LIKE = Regex("^[0-9a-fA-F:.]+(%[\\w-]+)?$")
+    private val STATE = Regex("^(REACHABLE|STALE|DELAY|PROBE|FAILED|INCOMPLETE|PERMANENT|NOARP|NONE)$")
+    private val MONITOR_LABEL = Regex("^\\[(\\w+)]")
+}

echolot-app/core-shizuku/src/main/kotlin/app/echo_lot/shizuku/ShizukuProbe.kt

@@ -10,15 +10,24 @@ import app.echo_lot.measurement.TestType
 import app.echo_lot.measurement.Tier
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.withContext
+import kotlinx.serialization.json.JsonObject
 import kotlinx.serialization.json.buildJsonObject
 import kotlinx.serialization.json.put
+import kotlinx.serialization.json.putJsonArray
+import kotlinx.serialization.json.putJsonObject
+import kotlinx.serialization.json.addJsonObject
 
 /**
  * The Shizuku shell-tier probe: runs the privileged command battery (neighbor table, RA routes
  * with lifetimes, netlink monitor, IpClient DHCP logs, wifi dump) that the app UID cannot, and
- * captures the real per-device dump formats the production parsers must handle. Emitted as a
- * shizuku-tier `link.ip_monitor` test (the representative shell-tier link test); `exec_path`
- * records whether the UserService or the newProcess fallback carried it.
+ * captures the real per-device dump formats the production parsers must handle. Emits three
+ * shizuku-tier tests from the one battery:
+ *  - `link.ip_monitor` — the raw captures (the shell tier's ground truth), `exec_path` records
+ *    whether the UserService or the newProcess fallback carried it;
+ *  - `link.ra_source` — parsed from the v6 route table + `ip addr`: who advertises IPv6 here;
+ *  - `sec.arp_watch` — parsed from the neighbor table + monitor window: (ip → lladdr) pairs for
+ *    gateway-MAC-change detection.
+ * The battery runs once; the derived tests parse its captures, so they share its time window.
  */
 class ShizukuProbe {
     val type = TestType.LINK_IP_MONITOR
@@ -34,24 +43,34 @@ class ShizukuProbe {
         "wifi_dump" to "dumpsys wifi 2>/dev/null | grep -iA1 -m 20 -e 'mDhcpResults' -e 'Gateway' -e 'DNS' || true",
     )
 
-    /** Runs the battery and returns a Test. [uuid]/[monoNs] come from the run's id/clock source. */
-    suspend fun run(context: Context, uuid: () -> String, monoNs: () -> Long): Test = withContext(Dispatchers.IO) {
-        val id = uuid()
+    /**
+     * Runs the battery and returns the three tests, battery first. [uuid]/[monoNs] come from the
+     * run's id/clock source. When the shell tier is unavailable all three come back UNSUPPORTED —
+     * one silent test would leave the other two types missing from the document, which reads as
+     * "never attempted" rather than "tier absent".
+     */
+    suspend fun run(context: Context, uuid: () -> String, monoNs: () -> Long): List<Test> = withContext(Dispatchers.IO) {
         val started = monoNs()
         val runner = ShizukuRunner(context)
         val st = runner.status()
 
-        fun envelope(status: TestStatus, evidence: kotlinx.serialization.json.JsonObject, metrics: kotlinx.serialization.json.JsonObject? = null) =
-            Test(id = id, type = type, tier = tier, startedMonoNs = started, endedMonoNs = monoNs(),
+        fun envelope(type: String, status: TestStatus, evidence: JsonObject, metrics: JsonObject? = null) =
+            Test(id = uuid(), type = type, tier = tier, startedMonoNs = started, endedMonoNs = monoNs(),
                 status = status, evidence = evidence, metrics = metrics)
 
+        fun allUnsupported(evidence: JsonObject) = listOf(
+            envelope(TestType.LINK_IP_MONITOR, TestStatus.UNSUPPORTED, evidence),
+            envelope(TestType.LINK_RA_SOURCE, TestStatus.UNSUPPORTED, evidence),
+            envelope(TestType.SEC_ARP_WATCH, TestStatus.UNSUPPORTED, evidence),
+        )
+
         if (!st.binderAlive) {
-            return@withContext envelope(TestStatus.UNSUPPORTED, buildJsonObject {
+            return@withContext allUnsupported(buildJsonObject {
                 put("binder_alive", false); put("detail", "Shizuku not running")
             })
         }
         if (!st.permissionGranted && !runner.requestPermission()) {
-            return@withContext envelope(TestStatus.UNSUPPORTED, buildJsonObject {
+            return@withContext allUnsupported(buildJsonObject {
                 put("binder_alive", true); put("permission", false)
             })
         }
@@ -77,6 +96,125 @@ class ShizukuProbe {
             ok >= 1 -> TestStatus.PARTIAL
             else -> TestStatus.FAILED
         }
-        envelope(status, evidence, metrics)
+        listOf(
+            envelope(type, status, evidence, metrics),
+            raSourceTest(batch, ::envelope),
+            arpWatchTest(batch, ::envelope),
+        )
+    }
+
+    /** `link.ra_source` from the battery's `ip -6 route` / `ip addr` / `ip neigh` captures. */
+    private fun raSourceTest(
+        batch: ShizukuRunner.BatchResult,
+        envelope: (String, TestStatus, JsonObject, JsonObject?) -> Test,
+    ): Test {
+        val routeRaw = batch.results["ip6_route"]
+        if (!DumpParsers.captureUsable(routeRaw)) {
+            // The source command failed (executor sentinel or empty) — say so instead of
+            // presenting "no default routes" as a measurement of the network.
+            return envelope(TestType.LINK_RA_SOURCE, TestStatus.SKIPPED, buildJsonObject {
+                put("exec_path", batch.execPath)
+                put("reason", "ip -6 route capture unavailable: ${(routeRaw ?: "absent").take(80)}")
+            }, null)
+        }
+        val routes = DumpParsers.parseV6DefaultRoutes(routeRaw)
+        val macs = DumpParsers.parseInterfaceMacs(batch.results["ip_addr"])
+        // The RA sender's own identity: its link-local gateway address resolved through the
+        // neighbor table gives the router's MAC, which is what survives address renumbering.
+        val neighMacs = DumpParsers.lladdrByIp(DumpParsers.parseNeighbors(batch.results["ip_neigh"]))
+
+        val evidence = buildJsonObject {
+            put("exec_path", batch.execPath)
+            putJsonArray("default_routes") {
+                for (r in routes) addJsonObject {
+                    r.gateway?.let { put("gateway", it) }
+                    put("dev", r.dev)
+                    r.table?.let { put("table", it) }
+                    r.proto?.let { put("proto", it) }
+                    r.metric?.let { put("metric", it) }
+                    r.expiresSec?.let { put("expires_sec", it) }
+                    r.gateway?.let { gw -> neighMacs[gw]?.let { put("gateway_lladdr", it) } }
+                }
+            }
+            putJsonObject("interface_mac") {
+                // Only interfaces that actually carry a default route: the full MAC inventory
+                // belongs to the raw capture, not to this test's claim.
+                for (dev in routes.map { it.dev }.distinct()) macs[dev]?.let { put(dev, it) }
+            }
+        }
+        val metrics = buildJsonObject {
+            put("routes_total", routes.size)
+            put("routes_ra", routes.count { it.proto == "ra" })
+        }
+        val status = when {
+            routes.any { it.proto == "ra" } -> TestStatus.OK
+            // Routes parsed but none RA-installed, or a capture we couldn't parse a single
+            // default from: could be a genuinely RA-less link, could be vendor format drift —
+            // PARTIAL keeps it visible either way instead of quietly claiming success.
+            else -> TestStatus.PARTIAL
+        }
+        return envelope(TestType.LINK_RA_SOURCE, status, evidence, metrics)
+    }
+
+    /** `sec.arp_watch` from the battery's `ip neigh` snapshot + `ip monitor` window. */
+    private fun arpWatchTest(
+        batch: ShizukuRunner.BatchResult,
+        envelope: (String, TestStatus, JsonObject, JsonObject?) -> Test,
+    ): Test {
+        val neighRaw = batch.results["ip_neigh"]
+        val monitorRaw = batch.results["ip_monitor"]
+        val neighUsable = DumpParsers.captureUsable(neighRaw)
+        // The monitor window is best-effort (EXEC_TIMEOUT under newProcess on the Lenovo); the
+        // snapshot alone still yields the (ip → lladdr) pairs the MAC-change finding diffs.
+        val monitorRan = DumpParsers.captureUsable(monitorRaw)
+        if (!neighUsable && !monitorRan) {
+            return envelope(TestType.SEC_ARP_WATCH, TestStatus.SKIPPED, buildJsonObject {
+                put("exec_path", batch.execPath)
+                put("reason", "ip neigh capture unavailable: ${(neighRaw ?: "absent").take(80)}")
+            }, null)
+        }
+        val neighbors = DumpParsers.parseNeighbors(neighRaw)
+        val events = DumpParsers.parseNeighborEvents(monitorRaw)
+
+        val evidence = buildJsonObject {
+            put("exec_path", batch.execPath)
+            putJsonArray("neighbors") {
+                for (n in neighbors) addJsonObject {
+                    put("ip", n.ip)
+                    n.dev?.let { put("dev", it) }
+                    n.lladdr?.let { put("lladdr", it) }
+                    n.state?.let { put("state", it) }
+                    if (n.router) put("router", true)
+                }
+            }
+            // The comparison surface, precomputed: a MAC-change finding diffs this map between
+            // runs without re-walking the neighbor array.
+            putJsonObject("lladdr_by_ip") {
+                for ((ip, mac) in DumpParsers.lladdrByIp(neighbors)) put(ip, mac)
+            }
+            put("monitor_ran", monitorRan)
+            if (!monitorRan) put("monitor_reason", (monitorRaw ?: "absent").take(80))
+            putJsonArray("monitor_events") {
+                for (e in events) addJsonObject {
+                    put("ip", e.entry.ip)
+                    e.entry.dev?.let { put("dev", it) }
+                    e.entry.lladdr?.let { put("lladdr", it) }
+                    e.entry.state?.let { put("state", it) }
+                    if (e.deleted) put("deleted", true)
+                }
+            }
+        }
+        val metrics = buildJsonObject {
+            put("neighbors_total", neighbors.size)
+            put("neighbors_with_lladdr", neighbors.count { it.lladdr != null })
+            put("monitor_events", events.size)
+        }
+        val status = when {
+            neighbors.isNotEmpty() -> TestStatus.OK
+            // A snapshot that parsed to nothing (or a monitor-only capture) is thin evidence:
+            // usable command output with zero entries is unusual enough to flag, not to fail.
+            else -> TestStatus.PARTIAL
+        }
+        return envelope(TestType.SEC_ARP_WATCH, status, evidence, metrics)
     }
 }

echolot-app/core-shizuku/src/test/kotlin/app/echo_lot/shizuku/DumpParsersTest.kt

@@ -0,0 +1,303 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package app.echo_lot.shizuku
+
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+/**
+ * The fixtures below are the REAL shell-battery captures from the two archived prober reports
+ * (echolot-prober/reports/CPH2747-android16-sdk36-build5.json — OnePlus 15, UserService path;
+ * TB330FU-android15-sdk35-build5.json — Lenovo TB330FU, newProcess fallback), trimmed to the
+ * relevant lines but otherwise verbatim. That includes their warts on purpose: the UserService
+ * path's stray `uid=2000` first line, the 1200-char evidence trim cutting the last line mid-word,
+ * the Lenovo's 10-digit route table ids and its `EXEC_TIMEOUT(newProcess)` monitor sentinel.
+ * A parser that only survives clean textbook output has not been tested.
+ */
+class DumpParsersTest {
+
+    // ---- OnePlus 15 (CPH2747, Android 16) — UserService exec path ----
+
+    private val onePlusIp6Route = """
+        uid=2000
+        fe80::/64 dev wlan0 table 1028 proto kernel metric 256 pref medium
+        fe80::/64 dev wlan0 table 1028 proto static metric 1024 pref medium
+        default via fe80::7a9a:18ff:fe54:b8f9 dev wlan0 table 1028 proto ra metric 1024 expires 1269sec pref medium
+        fe80::/64 dev vgate0 table 1031 proto kernel metric 256 pref medium
+        2001:4bb8:417:bd78::/64 dev rmnet_data4 table 1032 proto kernel metric 256 pref medium
+        2001:4bb8:417:bd78::/64 dev rmnet_data4 table 1032 proto static metric 1024 pref medium
+        fe80::/64 dev rmnet_data4 table 1032 proto kernel metric 256 pref medium
+        default via fe80::246f:12be:21ef:1b54 dev rmnet_data4 table 1032 proto ra metric 1024 expires 64373sec hoplimit 255 pref medium
+        2001:4bb8:2fb:fe4c::/64 dev rmnet_data2 table 1000000022 proto static metric 1024 pref medium
+        fe80::/64 dev wlan0 table 1000000028 proto static metric 1024 pref medium
+        2001:4bb8:417:bd78::/64 dev rmnet_data4 table 1000000032 proto static metric 1024 pref medium
+        fe80::/64 dev dummy0 table 1002 proto kernel metric 256 pref medium
+        default dev dummy0 table 1002 proto static metric 1024 pref medium
+        fe80::/64 dev ifb0 table 1003 proto kernel metric 256 pref medium
+        fe80::/64 dev ifb1 table 1004 proto kerne
+    """.trimIndent()
+
+    private val onePlusIpNeigh = """
+        uid=2000
+        10.13.102.111 dev wlan0  FAILED
+        10.13.102.50 dev wlan0 lladdr 50:57:9c:4f:7a:3c STALE
+        10.13.102.31 dev wlan0 lladdr 98:5f:d3:f6:f1:75 STALE
+        10.13.102.116 dev wlan0 lladdr 0c:08:b4:03:68:0e STALE
+        10.13.102.120 dev wlan0 lladdr 0e:d8:14:58:6c:8b STALE
+        10.13.102.5 dev wlan0 lladdr 90:09:d0:1a:83:e4 STALE
+        10.13.102.1 dev wlan0 lladdr 78:9a:18:54:b8:f9 REACHABLE
+        10.13.102.21 dev wlan0 lladdr c8:7f:54:01:94:7c STALE
+        fe80::babe:f4ff:febc:caf9 dev wlan0 lladdr b8:be:f4:bc:ca:f9 REACHABLE
+        fe80::7a9a:18ff:fe54:b8f9 dev wlan0 lladdr 78:9a:18:54:b8:f9 router STALE
+        fe80::babe:f4ff:febc:cacf dev wlan0 lladdr b8:be:f4:bc:ca:cf REACHABLE
+        fe80::babe:f4ff:fec2:bf14 dev wlan0 lladdr b8:be:f4:c2:bf:14 REACHABLE
+    """.trimIndent()
+
+    // The 1200-char trim cut this capture off before wlan0's stanza — so the real archived
+    // evidence has NO MAC for the interface that carries the default route. The parser must
+    // yield what is there and nothing else; the probe records the gap instead of inventing one.
+    private val onePlusIpAddr = """
+        uid=2000
+        1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue state UNKNOWN group default qlen 1000
+            link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00
+            inet 127.0.0.1/8 scope host lo
+               valid_lft forever preferred_lft forever
+            inet6 ::1/128 scope host
+               valid_lft forever preferred_lft forever
+        2: dummy0: <BROADCAST,NOARP,UP,LOWER_UP> mtu 1500 qdisc noqueue state UNKNOWN group default qlen 1000
+            link/ether be:3d:e2:93:78:b9 brd ff:ff:ff:ff:ff:ff
+            inet6 fe80::bc3d:e2ff:fe93:78b9/64 scope link
+               valid_lft forever preferred_lft forever
+        3: ifb0: <BROADCAST,NOARP,UP,LOWER_UP> mtu 1500 qdisc htb state UNKNOWN group default qlen 1000
+            link/ether ba:6e:46:b5:3d:bb brd ff:ff:ff:ff:ff:ff
+        4: ifb1: <BROADCAST,NOARP,UP,LOWER_UP> mtu 1500 qdisc htb state UNKNOWN group default qlen 1000
+            link/ether d6:2a:e2:f5:93:8f brd ff:ff:ff:ff:ff:ff
+        5: tunl0@NONE: <NOARP> mtu 1480 qdisc noop state DOWN group default qlen 1000
+            link/ipip 0.0.0.0 brd 0.0.0.0
+        6: gre0@NONE: <NO
+    """.trimIndent()
+
+    // A monitor window that ran but saw nothing: the capture is "usable", just empty of events.
+    private val onePlusIpMonitor = "uid=2000"
+
+    // ---- Lenovo TB330FU (Android 15) — newProcess fallback ----
+
+    private val lenovoIp6Route = """
+        fe80::/64 dev wlan0 table 1000000015 proto static metric 1024 pref medium
+        fe80::/64 dev dummy0 table 1002 proto kernel metric 256 pref medium
+        default dev dummy0 table 1002 proto static metric 1024 pref medium
+        fe80::/64 dev wlan0 table 1015 proto kernel metric 256 pref medium
+        fe80::/64 dev wlan0 table 1015 proto static metric 1024 pref medium
+        default via fe80::7a9a:18ff:fe54:b8f9 dev wlan0 table 1015 proto ra metric 1024 expires 1622sec pref medium
+        local ::1 dev lo table local proto kernel metric 0 pref medium
+        local fe80::416:b9ff:feac:5b65 dev wlan0 table local proto kernel metric 0 pref medium
+        local fe80::1450:43ff:feec:93c4 dev dummy0 table local proto kernel metric 0 pref medium
+        multicast ff00::/8 dev dummy0 table local proto kernel metric 256 pref medium
+        multicast ff00::/8 dev wlan0 table local proto kernel metric 256 pref medium
+    """.trimIndent()
+
+    private val lenovoIpNeigh = """
+        10.13.102.21 dev wlan0 lladdr c8:7f:54:01:94:7c STALE
+        10.13.102.64 dev wlan0 lladdr b8:be:f4:c2:bf:14 STALE
+        10.13.102.5 dev wlan0 lladdr 90:09:d0:1a:83:e4 STALE
+        10.13.102.1 dev wlan0 lladdr 78:9a:18:54:b8:f9 STALE
+        10.13.102.79 dev wlan0 lladdr 02:11:32:25:63:bb STALE
+        fe80::babe:f4ff:febc:cacf dev wlan0 lladdr b8:be:f4:bc:ca:cf STALE
+        fe80::7a9a:18ff:fe54:b8f9 dev wlan0 lladdr 78:9a:18:54:b8:f9 router STALE
+        fe80::babe:f4ff:fec2:bf14 dev wlan0 lladdr b8:be:f4:c2:bf:14 STALE
+        fe80::babe:f4ff:febc:caf9 dev wlan0 lladdr b8:be:f4:bc:ca:f9 STALE
+    """.trimIndent()
+
+    private val lenovoIpAddr = """
+        1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue state UNKNOWN group default qlen 1000
+            link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00
+            inet 127.0.0.1/8 scope host lo
+               valid_lft forever preferred_lft forever
+        2: dummy0: <BROADCAST,NOARP,UP,LOWER_UP> mtu 1500 qdisc noqueue state UNKNOWN group default qlen 1000
+            link/ether 16:50:43:ec:93:c4 brd ff:ff:ff:ff:ff:ff
+            inet6 fe80::1450:43ff:feec:93c4/64 scope link
+               valid_lft forever preferred_lft forever
+        3: ifb0: <BROADCAST,NOARP> mtu 1500 qdisc noop state DOWN group default qlen 32
+            link/ether f6:d4:d4:9b:51:9c brd ff:ff:ff:ff:ff:ff
+        4: ifb1: <BROADCAST,NOARP> mtu 1500 qdisc noop state DOWN group default qlen 32
+            link/ether fe:16:ea:60:a2:d1 brd ff:ff:ff:ff:ff:ff
+        5: tunl0@NONE: <NOARP> mtu 1480 qdisc noop state DOWN group default qlen 1000
+            link/ipip 0.0.0.0 brd 0.0.0.0
+        7: gretap0@NONE: <BROADCAST,MULTICAST> mtu 1462 qdisc noop state DOWN group default qlen 1000
+            link/ether 00:00:00:00:00:00 brd ff:ff:ff:ff:f
+    """.trimIndent()
+
+    // On the Lenovo the 5 s monitor window exceeds the newProcess exec timeout — the executor's
+    // sentinel is all we get, and the arp_watch test must still stand on the snapshot alone.
+    private val lenovoIpMonitor = "EXEC_TIMEOUT(newProcess)"
+
+    // ---- link.ra_source: v6 default routes ----
+
+    @Test
+    fun onePlusDefaultRoutesParsed() {
+        val routes = DumpParsers.parseV6DefaultRoutes(onePlusIp6Route)
+        assertEquals(3, routes.size)
+
+        val wlan = routes.single { it.dev == "wlan0" }
+        assertEquals("fe80::7a9a:18ff:fe54:b8f9", wlan.gateway)
+        assertEquals("1028", wlan.table)
+        assertEquals("ra", wlan.proto)
+        assertEquals(1024L, wlan.metric)
+        assertEquals(1269L, wlan.expiresSec)
+
+        // The cellular default: `hoplimit 255` sits between expires and pref and must not derail
+        // the token walk.
+        val rmnet = routes.single { it.dev == "rmnet_data4" }
+        assertEquals("fe80::246f:12be:21ef:1b54", rmnet.gateway)
+        assertEquals("1032", rmnet.table)
+        assertEquals(64373L, rmnet.expiresSec)
+
+        // Android's gateway-less dummy0 default is a real route; it is the proto that tells a
+        // consumer it is not an RA.
+        val dummy = routes.single { it.dev == "dummy0" }
+        assertNull(dummy.gateway)
+        assertEquals("static", dummy.proto)
+        assertNull(dummy.expiresSec)
+    }
+
+    @Test
+    fun lenovoNumberedTablesAllCaptured() {
+        val routes = DumpParsers.parseV6DefaultRoutes(lenovoIp6Route)
+        // Two default routes: the RA one in table 1015 and the dummy0 one in 1002. The 10-digit
+        // table 1000000015 and the `table local` rows carry no default and must neither appear
+        // nor break parsing.
+        assertEquals(setOf("1002", "1015"), routes.map { it.table }.toSet())
+
+        val ra = routes.single { it.proto == "ra" }
+        assertEquals("fe80::7a9a:18ff:fe54:b8f9", ra.gateway)
+        assertEquals("wlan0", ra.dev)
+        assertEquals("1015", ra.table)
+        assertEquals(1622L, ra.expiresSec)
+    }
+
+    @Test
+    fun tenDigitTableIdOnADefaultRouteSurvives() {
+        // Not seen on a default route in the wild yet, but the Lenovo proves vendors put routes
+        // in tables past Int range — the day one holds a default, it must not overflow away.
+        val routes = DumpParsers.parseV6DefaultRoutes(
+            "default via fe80::1 dev wlan0 table 1000000015 proto ra metric 1024 expires 100sec pref medium"
+        )
+        assertEquals(1, routes.size)
+        assertEquals("1000000015", routes[0].table)
+        assertEquals(100L, routes[0].expiresSec)
+    }
+
+    // ---- link.ra_source: interface MACs ----
+
+    @Test
+    fun onePlusInterfaceMacsParsed() {
+        val macs = DumpParsers.parseInterfaceMacs(onePlusIpAddr)
+        assertEquals("be:3d:e2:93:78:b9", macs["dummy0"])
+        assertEquals("ba:6e:46:b5:3d:bb", macs["ifb0"])
+        // link/loopback and link/ipip are not identities.
+        assertFalse("lo" in macs)
+        assertFalse("tunl0" in macs)
+        // The capture is cut mid-stanza-header ("6: gre0@NONE: <NO") — no exception, no entry.
+        assertNull(macs["gre0"])
+    }
+
+    @Test
+    fun lenovoInterfaceMacsParsed() {
+        val macs = DumpParsers.parseInterfaceMacs(lenovoIpAddr)
+        assertEquals("16:50:43:ec:93:c4", macs["dummy0"])
+        assertEquals("fe:16:ea:60:a2:d1", macs["ifb1"])
+        // The @-suffixed stanza name resolves to the bare interface name.
+        assertEquals("00:00:00:00:00:00", macs["gretap0"])
+    }
+
+    // ---- sec.arp_watch: neighbor snapshot ----
+
+    @Test
+    fun onePlusNeighborsParsed() {
+        val n = DumpParsers.parseNeighbors(onePlusIpNeigh)
+        assertEquals(12, n.size) // the `uid=2000` noise line is not a neighbor
+
+        val failed = n.single { it.ip == "10.13.102.111" }
+        assertNull(failed.lladdr)
+        assertEquals("FAILED", failed.state)
+        assertEquals("wlan0", failed.dev)
+
+        val gw = n.single { it.ip == "10.13.102.1" }
+        assertEquals("78:9a:18:54:b8:f9", gw.lladdr)
+        assertEquals("REACHABLE", gw.state)
+
+        val v6gw = n.single { it.ip == "fe80::7a9a:18ff:fe54:b8f9" }
+        assertTrue(v6gw.router)
+        assertEquals("78:9a:18:54:b8:f9", v6gw.lladdr)
+    }
+
+    @Test
+    fun lenovoNeighborsParsed() {
+        val n = DumpParsers.parseNeighbors(lenovoIpNeigh)
+        assertEquals(9, n.size)
+        assertTrue(n.all { it.lladdr != null && it.state == "STALE" })
+        assertEquals(1, n.count { it.router })
+    }
+
+    @Test
+    fun lladdrByIpIsTheComparisonSurface() {
+        val pairs = DumpParsers.lladdrByIp(DumpParsers.parseNeighbors(onePlusIpNeigh))
+        // 12 neighbors, 11 with a MAC — the FAILED entry must drop out, or a diff against a
+        // later run would flag "null → MAC" as a gateway change.
+        assertEquals(11, pairs.size)
+        assertEquals("78:9a:18:54:b8:f9", pairs["10.13.102.1"])
+        assertFalse("10.13.102.111" in pairs)
+    }
+
+    // ---- sec.arp_watch: monitor window ----
+
+    @Test
+    fun monitorSentinelIsUnusableAndYieldsNoEvents() {
+        assertFalse(DumpParsers.captureUsable(lenovoIpMonitor))
+        assertTrue(DumpParsers.parseNeighborEvents(lenovoIpMonitor).isEmpty())
+    }
+
+    @Test
+    fun quietMonitorWindowIsUsableButEmpty() {
+        // OnePlus: the monitor ran (only the uid noise line came back) — "ran and saw nothing"
+        // must stay distinguishable from "never ran".
+        assertTrue(DumpParsers.captureUsable(onePlusIpMonitor))
+        assertTrue(DumpParsers.parseNeighborEvents(onePlusIpMonitor).isEmpty())
+    }
+
+    @Test
+    fun monitorNeighEventsParsedFromLabeledLines() {
+        // Synthetic, in `ip monitor all` label format — neither archived run caught a live
+        // transition, but the format is fixed by iproute2's print_neigh/print_headers.
+        val sample = """
+            [NEIGH]10.13.102.1 dev wlan0 lladdr 78:9a:18:54:b8:f9 REACHABLE
+            [NEIGH]Deleted 10.13.102.5 dev wlan0 lladdr 90:09:d0:1a:83:e4 STALE
+            [ROUTE]default via 10.13.102.1 dev wlan0 table 1015
+            [NEIGH]fe80::7a9a:18ff:fe54:b8f9 dev wlan0 lladdr 78:9a:18:54:b8:f9 router STALE
+        """.trimIndent()
+        val events = DumpParsers.parseNeighborEvents(sample)
+        assertEquals(3, events.size) // the ROUTE line belongs to a different family
+        assertEquals("10.13.102.1", events[0].entry.ip)
+        assertFalse(events[0].deleted)
+        assertTrue(events[1].deleted)
+        assertEquals("90:09:d0:1a:83:e4", events[1].entry.lladdr)
+        assertTrue(events[2].entry.router)
+    }
+
+    // ---- degradation: missing or garbage input ----
+
+    @Test
+    fun missingAndGarbageInputYieldsEmptyResultsNotExceptions() {
+        for (bad in listOf(null, "", "   \n ", "EXEC_TIMEOUT(newProcess)", "SHIZUKU_BINDER_DEAD",
+            "NEWPROCESS_UNAVAILABLE", "total garbage\nno routes here at all\ndefault", "default")) {
+            assertTrue(DumpParsers.parseV6DefaultRoutes(bad).isEmpty(), "routes from: $bad")
+            assertTrue(DumpParsers.parseNeighbors(bad).isEmpty(), "neighbors from: $bad")
+            assertTrue(DumpParsers.parseInterfaceMacs(bad).isEmpty(), "macs from: $bad")
+            assertTrue(DumpParsers.parseNeighborEvents(bad).isEmpty(), "events from: $bad")
+        }
+    }
+}

echolot-app/gradle/libs.versions.toml

@@ -8,6 +8,7 @@ lifecycle = "2.8.7"
 activityCompose = "1.9.3"
 composeBom = "2024.10.01"
 shizuku = "13.1.5"
+junit4 = "4.13.2"
 
 [libraries]
 kotlinx-serialization-json = { group = "org.jetbrains.kotlinx", name = "kotlinx-serialization-json", version.ref = "kotlinxSerialization" }
@@ -23,6 +24,10 @@ androidx-ui-tooling = { group = "androidx.compose.ui", name = "ui-tooling" }
 androidx-ui-tooling-preview = { group = "androidx.compose.ui", name = "ui-tooling-preview" }
 androidx-material3 = { group = "androidx.compose.material3", name = "material3" }
 shizuku-api = { group = "dev.rikka.shizuku", name = "api", version.ref = "shizuku" }
+# Android-module unit tests run on JUnit 4 (AGP's default); the JVM modules use kotlin("test")
+# with the JUnit Platform instead — that helper isn't available under AGP 9's built-in Kotlin.
+kotlin-test-junit = { group = "org.jetbrains.kotlin", name = "kotlin-test-junit", version.ref = "kotlin" }
+junit4 = { group = "junit", name = "junit", version.ref = "junit4" }
 shizuku-provider = { group = "dev.rikka.shizuku", name = "provider", version.ref = "shizuku" }
 
 [plugins]