throughput: the upstream direction, counted by the only party that can

The client generates the traffic and the server counts it. No grant is involved
- the client is sending its own packets, so there is nothing to amplify - but it
does need the server's tally, because only the far end knows how much arrived.
Without that number a sender measures how fast it can transmit, which is usually
just the speed of the local NIC and is not the question being asked.

A new wire type the server counts and deliberately never answers: a reply would
double the traffic and drag the return path into a measurement that is
specifically about the outbound one.

The tally is a counter, not a list, and short-circuits before the observation
log. A five-second run at 20 Mbps is around ten thousand packets; one struct
each would turn a measurement into an allocation storm on a shared server, and
nothing needs the per-packet detail since the client holds the send-side record.
The gap between the two counts is the loss.

direction=up on the throughput action sends nothing - it zeroes the counter, so
a second run in one session measures itself instead of inheriting the first.

Same honesty rule as downstream: measures_network is false when what arrived
matches what was offered, because then the path was never the constraint.

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

docs/build-status.md

@@ -728,3 +728,216 @@ tenfold asymmetry that a round-trip measurement cannot see at all.
 
 10 unit tests on the arithmetic (a wrong denominator here does not crash, it produces a plausible
 number pointing at the wrong half of the network) plus the live correlation check.
+
+### frag_send: crafted IP fragments, so *ordering* is testable (server-v0.6.0, 2026-08-01)
+`big_send` with `df=false` answers one question — do fragments get through. It cannot answer the
+more interesting one, because the kernel always emits fragments in order, first one first.
+
+The classic middlebox fault is exactly about that ordering. Only the **first** fragment carries the
+UDP header, and therefore the ports; a stateful firewall or NAT that has not seen it has no flow to
+match the rest against, and many simply drop them. That is invisible to every in-order test, and in
+the field it looks like "large DNS answers fail on this network" or "the tunnel breaks when the MTU
+drops" — it works until the network reorders, then fails intermittently, which is the hardest kind
+of fault to chase.
+
+So the server builds the fragments itself (raw socket, `IP_HDRINCL`) and controls their order:
+`in_order` (baseline), `reversed` (last fragment first), `first_last` (first fragment held back
+250 ms). The datagram is assembled and **signed whole** before being cut up, so what the client
+reassembles is indistinguishable from an ordinary packet — otherwise the test would be measuring
+our sender rather than the path. New test type `mtu.frag_ordering`; findings
+`mtu.fragments_blocked` and `mtu.fragment_reorder_sensitive`.
+
+Two details that would otherwise produce confidently wrong answers:
+- **The UDP checksum is computed, not left zero.** Zero is legal in IPv4 and would be less code,
+  but zero-checksum datagrams are dropped by some middleboxes — and that drop would be recorded as
+  a fragmentation failure, which is the wrong conclusion entirely.
+- **Fragment offsets are in 8-byte units**, so non-final fragments are rounded down to a multiple
+  of 8. A 100-byte fragment is not an error; it is a datagram no host will ever reassemble.
+
+`frag-send` is advertised only when a raw socket can actually be opened — checked by opening one,
+because a permission model has more ways to say no (userns, seccomp, LSM) than a capability bit has
+to say yes. fmr runs as root with `cap_net_raw` in its bounding set, so it is available there.
+
+Fragment ordering runs only after `mtu.frag_delivery` shows fragments arrive at all; otherwise the
+three orderings would each report "not delivered" and read as three faults instead of one.
+
+The header arithmetic is unit-tested (reassembly coverage with no gaps or double-delivery, MF
+flags, shared IP ID, 8-byte offsets, checksum verification over odd and even lengths). Because the
+code is `//go:build linux`, the tests are **cross-compiled and run on fmr** — there is no Go
+toolchain there, so `go test -c` plus scp is the loop.
+
+Live against fmr: 4 fragments per burst, and all three orderings reassembled — a healthy path, and
+the baseline against which a mobile network will be interesting.
+
+### Testing state (2026-08-01)
+Six live tests against fmr, all green, no device involved: `LiveServerTest`, `LiveMeasurement`,
+`LiveGranted`, `LiveDownstream`, `LiveUpload`, `LiveCompat`, `LiveEnrollment`. Plus 74 client unit
+tests and the full Go suite. Everything in the last several entries is verified from the PC; the
+app's UI (settings, history, deep-link enrollment) and `mtu.pmtud_up` remain device-only.
+
+### throughput: a rate, plus the qualifier that makes it a measurement (server-v0.6.1 … v0.6.2)
+A throughput test reports the *smallest* limit on the path — and the sender's own ceiling is one of
+the candidates. If the server is asked for 50 Mbps and 50 Mbps arrives, the network was never the
+constraint and "50 Mbps" says nothing about it. So `perf.throughput_udp` always carries
+`limited_by` (duration | budget | rate | send_error) and `measures_network`, and a finding is
+raised only when the path is actually implicated. The live run against fmr reports 20 Mbit/s with
+`measures_network: false`, which is the correct and useful answer.
+
+Loss is computed against the **sender's own count**, fetched from the observations API, not against
+the requested rate. A receiver alone cannot tell "the network dropped it" from "the sender never
+sent it", and guessing turns a healthy server-side limit into a phantom network fault. The server
+keeps one summary per action rather than per-packet records — a ten-second run at 50 Mbps is half a
+million packets, and a struct each would turn a measurement into memory exhaustion.
+
+Sending is **paced**, on an absolute schedule. Unpaced would measure the server's NIC and the first
+queue it meets, then collapse into loss that reads as a network fault; sleep-per-packet would
+accumulate scheduler error and drift the rate down over a ten-second run.
+
+Throughput gets its own grant budget sized from the request, so every *other* action stays bounded
+at 8 MiB. When the byte cap binds before the clock does, the **duration is shortened and reported**
+rather than the run being truncated: promising thirty seconds and delivering twenty-one is the same
+information with a surprise attached, and it keeps "the clock ended the run" as the normal case —
+the only case where the rate is a clean property of the path. That behaviour came out of a test
+that failed honestly (30 s at 100 Mbps needs 375 MB against a 256 MB cap).
+
+It is **opt-in** in the run config, default off. A 5-second run at 50 Mbps moves ~30 MB; on a
+metered mobile connection that is the user's money, and a tool that spends it without being asked
+is not one people keep installed.
+
+#### The bug the live test found
+The first live run delivered 104 packets and stopped after 50 ms. The grant's rate check exempted
+the first 50 ms entirely, meaning to be lenient at startup — the effect was the opposite. A sender
+could dump an unbounded burst into that free window, and the instant the check switched on it
+compared those bytes against 50 ms worth of allowance and refused everything until real time caught
+up. **Every short test passed** (downtrain sends 50 packets, big_send seven); every sustained send
+died fifty milliseconds in.
+
+Replaced with a token bucket (`allowance = burst + rate × elapsed`), which is smooth from t=0.
+The burst is 100 ms of the allowed rate, floored at one ordinary datagram — deliberately one, since
+at 8 kbps a 64 KB floor is sixty-four seconds' worth, exactly the instant dump the ceiling exists to
+prevent. The pre-existing rate test caught that when I first tried the generous floor, and it was
+right to. Second half of the same bug: callers treated *any* refusal as terminal, so `TryAllow` now
+says why — a sender paces through a transient "too fast just now" and still stops dead on a spent
+budget or an expired grant. Both halves are pinned by regression tests.
+
+### Findings registry (2026-08-01)
+Closes open item 1 of measurement-schema.md §9. A finding code is the stable, machine-readable half
+of a result — what a dashboard groups by and what someone greps a year of archived runs for — and
+that only holds if a code means exactly one thing forever. Ad-hoc string literals at fifteen call
+sites cannot promise that, and by the time the registry was written the failure had already
+happened.
+
+**Two emitters had independently produced `connectivity.downstream_loss` and
+`connectivity.loss_downstream` for the same claim**, and nothing anywhere objected. Anyone
+aggregating either one would have silently seen half their data. Merged into
+`connectivity.loss_downstream`, paired with `loss_upstream` so the two directions read as a set.
+
+**Two codes were also renamed out of `nat.*`.** `nat.udp_unreachable` is not about NAT — it means
+no replies came back — but the prefix determines the category, and the category determines which
+verdict light the finding rolls up into (§7.3). A `nat.*` code landing under *connectivity* is not
+a naming quibble; it changes which light turns red. Cheap to fix now, a breaking change later.
+
+Codes are now declared as typed `FindingSpec`s carrying their category and default severity, and
+emitters reference the spec instead of retyping the string — so a typo is a compile error and two
+call sites cannot disagree about a finding's category.
+
+`docs/findings-registry.md` is the contract, and a test reads it: it fails when the document and
+the registry have codes the other lacks, or when a severity differs. Documentation that drifts from
+its implementation is worse than none, because it still looks authoritative. The check scopes
+itself to table rows, so the prose can keep explaining which codes were retired and why.
+
+Six tests: uniqueness, declared-vs-listed, prefix↔category agreement, naming convention, a
+word-order-anagram check (the shape the duplication actually took), and the document agreement.
+
+### A real privacy leak, found by starting on the machine-readable schema (2026-08-01)
+The intent was `measurement.schema.json` (§8's promised companion). The first step — checking
+whether the anonymizer actually covers the fields the schema declares as sensitive — found that it
+did not, so that became the work.
+
+**At the `balanced` level, five identifying values were being uploaded verbatim:**
+
+| value | field | why it matters |
+|---|---|---|
+| `2001:…::150` | `networks[].link.addresses[].addr` | the device's own global IPv6 address — a strong, geolocatable device identifier |
+| `2a02:…::1` | `networks[].link.routes[].gateway` | identifies the ISP allocation |
+| `203.0.113.77` | `networks[].link.dns.servers[]` | the configured resolver |
+| `nas.example.lan` | `private_dns_hostname` | an internal hostname |
+| `example.lan` | `search_domains[]` | the internal domain |
+
+The settings screen describes that level as pseudonymizing addresses. It was not.
+
+**Root cause:** classification keyed on field *names*, and the schema's actual names (`addr`,
+`gateway`, `dst`, `servers`, `search_domains`, `private_dns_hostname`) had never been added to the
+table. Not a subtle bug — just an unfalsifiable design. The existing tests all passed, because each
+one checked a field somebody had remembered to write a case for.
+
+**Two fixes, one of them structural:**
+1. The missing names were added.
+2. More importantly, a **shape-based backstop**: when a field name is unrecognised, the *value* is
+   inspected, and anything shaped like an IPv4/IPv6 address or a MAC is treated as one. A name
+   table can only protect fields someone thought of, which is precisely the wrong property for a
+   privacy control. Hostnames are deliberately *not* inferred by shape — `train.udp_updown` is
+   indistinguishable from a domain, and mangling a test type would corrupt the document to protect
+   nothing.
+
+`LeakTest` is the new guard and is written to fail for fields nobody has considered: it plants
+identifying values wherever one can actually occur and asserts none survive, rather than checking
+a list of known cases. It also pins that RFC1918 addresses still come through readable, so the
+test cannot pass by over-redacting everything.
+
+Route prefixes and the unspecified address needed care in the transform: `0.0.0.0/0` and `::/0`
+must stay themselves, or a routing table becomes unreadable for no privacy gain.
+
+**Still outstanding:** `measurement.schema.json` itself. Worth noting what this episode implies for
+it — much of a document's payload lives in `evidence`/`metrics`/`params`, which are per-test-type
+`JsonObject` by design and therefore *outside* any schema. A schema-driven anonymizer would have
+less coverage there than the name-plus-shape one now does, so the schema should be built for
+validation and external tooling, not as a replacement for the classifier.
+
+### ULA prefixes are pseudonymized whole (2026-08-01)
+Spotted in a real uploaded run from the phone: the server had
+`fda1:3fb1:ff92:6696::2662` for a DNS server. The general IPv6 path preserves the leading two
+groups (deliberately — for a global address that keeps the ISP allocation, which is the
+diagnostically useful part), and for a ULA that passed through **32 of the 40 random bits** of the
+global ID.
+
+ULA looks like the v6 equivalent of RFC1918 and the instinct is to treat it the same. That
+reasoning does not carry over, and the difference is the whole point: an RFC1918 prefix is shared
+by millions of networks and identifies none of them, while a ULA global ID is random and unique to
+one network by construction (RFC 4193). The prefix *is* the identifier — it is a network
+fingerprint that was surviving redaction.
+
+Now pseudonymized as a unit, so two addresses on the same ULA subnet still land on the same
+pseudonymous prefix: "these hosts are on one network" survives, "this is *that* network" does not.
+Three tests, one of which uses the exact value observed on the wire.
+
+Worth recording as a reasoning trap: I had originally raised this as "ULA should probably be kept
+verbatim, like RFC1918, for consistency". The surface analogy pointed the wrong way, and the
+correct answer was the opposite.
+
+### Registry adopted everywhere; v6 findings renamed; Back works (2026-08-01)
+The findings registry was only adopted in `core-engine`. The app module still emitted seven codes
+as raw strings, so the registry test passed while codes existed outside it — including
+`ipv6.broken`, which fired on a real network and was in no registry at all.
+
+All seven now reference registry entries for code, category and severity, so those three cannot
+disagree at a call site. A grep for `code = "…"` across the app, engine and probe modules returns
+nothing.
+
+**`ipv6.*` → `v6.*`.** The third instance of rule 1: they declared `Category.IPV6` while the prefix
+map only knows `v6`, so `TestType.category("ipv6.broken")` fell through to *connectivity* and the
+finding rolled up under the wrong verdict light. The test-type registry already used `v6.`.
+
+Two severities reconciled while merging:
+- `connectivity.captive_portal` is **medium**, not high. The registry had guessed high; the probe
+  that emits it had always said medium, and the probe was the considered value — a captive portal
+  on hotel wifi is what should be there, and logging in clears it. `connectivity.no_internet` is
+  the high one, because nothing the user does locally fixes that.
+- `v6.not_offered` is **info, and the registry says it must stay info**. Most networks still do not
+  offer IPv6 and that is not a fault; a warning here lights a yellow verdict on a healthy network,
+  which teaches people to ignore the light.
+
+Also: a `BackHandler` now returns from Settings/History to the run screen. The screen was a plain
+state variable with nothing connecting it to the back stack, so the system Back gesture left the
+app entirely. Enabled only when there is somewhere to go back to, so Back still exits from the run
+screen.

docs/findings-registry.md

@@ -0,0 +1,106 @@
+<!--
+SPDX-FileCopyrightText: 2026 Echolot contributors
+SPDX-License-Identifier: CC-BY-4.0
+-->
+
+# Echolot findings registry
+
+Closes open item 1 of `measurement-schema.md` §9.
+
+A **finding code** is the stable, machine-readable half of a result. The prose around it changes
+freely; the code is what a dashboard groups by, what a diff between two runs keys on, and what
+someone greps a year of archived runs for. That only works if a code means exactly one thing,
+forever.
+
+This document is the contract. It is kept in step with
+`echolot-app/core-measurement/.../FindingRegistry.kt` by a test that fails when either side has a
+code the other does not — a registry that drifts from its documentation is worse than none,
+because it looks authoritative.
+
+## Rules
+
+1. **The prefix determines the category**, and the category determines which verdict light the
+   finding rolls up into (§7.3). A `nat.*` code appearing under *connectivity* is not a naming
+   quibble; it changes which light turns red. Two codes were renamed from `nat.*` to
+   `connectivity.*` for exactly this reason.
+2. **One code per concept.** Two emitters independently produced `connectivity.downstream_loss`
+   and `connectivity.loss_downstream` for the same claim before this registry existed. Anyone
+   aggregating either would have silently seen half their data.
+3. **Codes are declared, not typed.** Emitters reference a `FindingSpec`, so a typo is a compile
+   error and no two call sites can disagree about a finding's category or default severity.
+4. **Severity in the registry is the default.** An emitter may escalate for a specific run; it may
+   not quietly reclassify the finding in general.
+5. **Say what is ruled out**, where that is the useful half. "Loss upstream" is worth far more
+   when it also states that the return path is clean, because that halves where to look next.
+6. **Renaming a code is a breaking change** once runs are archived at scale. Before 1.0 it is
+   cheap; after, it needs an alias and a deprecation window.
+
+## Registry
+
+### connectivity
+
+| code | severity | means | rules out |
+|---|---|---|---|
+| `connectivity.udp_unreachable` | high | No UDP echo replies came back from the server at all. | — |
+| `connectivity.udp_unreachable_upstream` | high | The server received none of the probes, so traffic is dropped on the way out. | The return path: nothing arrived to be replied to. |
+| `connectivity.udp_loss` | medium | A large fraction of round-trip probes were lost, direction unknown. | — |
+| `connectivity.loss_upstream` | medium | Probes were lost on the way to the server. | The return path: replies came back for everything that arrived. |
+| `connectivity.loss_downstream` | medium | Packets were lost on the way back from the server. | The outbound path: the server received what it was answering. |
+| `connectivity.downstream_blocked` | high | Server-initiated packets never arrive, although round trips work. | Basic reachability: the path forwards replies, just not unsolicited traffic. |
+| `connectivity.downstream_reorder` | low | Downstream packets arrive in a different order than they were sent. | — |
+| `connectivity.captive_portal` | medium | A captive portal is intercepting connectivity checks. | — |
+| `connectivity.no_internet` | high | Android's own connectivity checks fail on this network. | — |
+
+### mtu
+
+| code | severity | means | rules out |
+|---|---|---|---|
+| `mtu.reduced_downstream` | low | The downstream path MTU is below the usual 1500 bytes. | — |
+| `mtu.downstream_blackhole` | medium | Datagrams above the path MTU are dropped downstream, fragmented or not. | — |
+| `mtu.fragments_blocked` | medium | IP fragments do not reach this device even when sent in order. | — |
+| `mtu.fragment_reorder_sensitive` | low | Fragments are delivered in order but dropped when reordered or delayed. | Fragmentation itself: in-order fragments arrive fine. |
+
+### nat
+
+| code | severity | means | rules out |
+|---|---|---|---|
+| `nat.udp_rebinding` | medium | A NAT remapped the UDP source port mid-flow. | — |
+| `nat.symmetric` | medium | The NAT assigns a different external port per destination. | — |
+
+### perf
+
+| code | severity | means | rules out |
+|---|---|---|---|
+| `perf.throughput_no_delivery` | high | No throughput traffic arrived, although the server sent it. | — |
+| `perf.throughput_below_offered` | low | Less throughput arrived than the server sent for the whole run. | — |
+
+### dns
+
+| code | severity | means | rules out |
+|---|---|---|---|
+| `dns.answer_rewritten` | high | A resolver returned an answer that differs from the authoritative record. | — |
+| `dns.authoritative_unreachable` | medium | The canary zone's authoritative server could not be reached. | — |
+
+### v6
+
+The prefix is `v6.`, matching the test-type registry (`v6.brokenness`, `v6.happy_eyeballs`, …).
+These were `ipv6.*` while declaring `Category.IPV6`; since the prefix map only knows `v6`, they
+rolled up under *connectivity* instead — the third occurrence of rule 1 being broken.
+
+| code | severity | means | rules out |
+|---|---|---|---|
+| `v6.broken` | medium | IPv6 is configured on this network but does not work. | Absence of IPv6: it is provisioned, it simply fails. |
+| `v6.not_offered` | info | This network does not offer IPv6. | — |
+
+`v6.not_offered` is **info and must stay info**. Most networks still do not offer IPv6 and that is
+not a fault; reporting it as a warning lights a yellow verdict on a healthy network, which teaches
+people to ignore the light — the one thing a diagnostic must never do.
+
+## Adding a finding
+
+1. Add a `FindingSpec` to `FindingRegistry`, and to its `all` list.
+2. Add the row here, under the section its prefix names.
+3. Emit it with `finding(FindingRegistry.YOUR_CODE, …)`.
+
+The registry test checks 1 and 2 agree, that every prefix maps to the category it claims, and that
+no two entries share a code.

docs/measurement-schema.md

@@ -269,7 +269,7 @@ The JSON Schema (machine-readable companion, `measurement.schema.json`, generate
 
 | type | example fields | v2 anonymizer transform |
 |---|---|---|
-| `ip4`, `ip6` | addresses, routes, hops, DNS answers | prefix-preserving pseudonymization, consistent per document; well-known/reserved ranges kept verbatim |
+| `ip4`, `ip6` | addresses, routes, hops, DNS answers | prefix-preserving pseudonymization, consistent per document; well-known/reserved ranges kept verbatim. **Exception: ULA (`fc00::/7`) has its whole prefix pseudonymized as a unit.** It resembles RFC1918 but is not analogous: a ULA global ID is 40 random bits, unique to one network by construction (RFC 4193), so the prefix *is* the identifier, whereas `192.168.0.0/16` is shared by millions of networks and identifies none. Pseudonymizing it as a unit keeps "these hosts are on one subnet" while dropping "this is that subnet". |
 | `mac`, `bssid` | wifi, arp_watch | OUI kept, NIC part pseudonymized |
 | `fqdn` | DNS names, reverse lookups | per-label pseudonyms, public-suffix kept |
 | `ssid` | wifi | pseudonym |
@@ -279,7 +279,8 @@ Free-text fields (`notes`, `error.detail`, dump excerpts from Shizuku parsers) c
 
 ## 9. Open items
 
-1. Findings registry document — start alongside the first implemented tests.
+1. ~~Findings registry document~~ — done: `findings-registry.md`, kept in step with
+   `FindingRegistry.kt` by a test that fails when the two disagree.
 2. Whether Shizuku raw-dump excerpts (dumpsys/ip output) are embedded in `evidence` verbatim (auditable, but large and hard to anonymize) or parsed-only with an optional "attach raw dumps" toggle. Proposal: toggle, default on for local archive, default off for export.
 3. Peer-mode documents: each device produces its own run; the coordinator embeds the peer's findings summary and cross-references by `run.id`. Full merge format deferred.
 4. Size guardrails: soft cap 20 MB uncompressed per run; trains beyond that downsample evidence (keep aggregates + first/last N + all anomalies) and record `"evidence_truncated": true`.

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

@@ -4,6 +4,7 @@
 package app.echo_lot.app
 
 import androidx.compose.foundation.layout.Arrangement
+import androidx.compose.foundation.layout.safeDrawingPadding
 import androidx.compose.foundation.layout.Column
 import androidx.compose.foundation.layout.Row
 import androidx.compose.foundation.layout.fillMaxWidth
@@ -39,7 +40,7 @@ fun HistoryScreen(
     onDelete: (String) -> Unit,
     onBack: () -> Unit,
 ) {
-    Column(Modifier.fillMaxWidth().padding(16.dp), verticalArrangement = Arrangement.spacedBy(10.dp)) {
+    Column(Modifier.fillMaxWidth().safeDrawingPadding().padding(16.dp), verticalArrangement = Arrangement.spacedBy(10.dp)) {
         Row(verticalAlignment = Alignment.CenterVertically) {
             TextButton(onClick = onBack) { Text("‹ Back") }
             Text("History", style = MaterialTheme.typography.titleLarge)
@@ -71,12 +72,20 @@ fun HistoryScreen(
                             )
                         }
                         Text(
-                            "${r.findingCount} finding(s) · ${r.sizeBytes / 1024} kB · ${r.anonymization}",
+                            "${r.findingCount} finding(s) · ${r.sizeBytes / 1024} kB · " +
+                                "kept complete on this device",
                             style = MaterialTheme.typography.bodySmall,
                         )
+                        // The upload line names the level the upload was made at, not the
+                        // archive's. They describe different documents, and showing the archive's
+                        // level here claimed more had left the device than actually did.
                         Text(
-                            if (r.uploaded) "uploaded to ${r.uploadedTo ?: "a server"}"
+                            if (r.uploaded) {
-                            else "on this device only",
+                                "uploaded to ${r.uploadedTo ?: "a server"}" +
+                                    (r.uploadedAs?.let { " as $it" } ?: "")
+                            } else {
+                                "on this device only"
+                            },
                             style = MaterialTheme.typography.bodySmall,
                             color = if (r.uploaded) Color(0xFF7FD17F) else Color(0xFFBBBBBB),
                         )

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

@@ -85,10 +85,18 @@ class MainActivity : ComponentActivity() {
                             finish()
                         }
                     }
+                    // Without this, the system Back gesture leaves the activity from Settings or
+                    // History instead of returning to the run screen — the screen is a plain state
+                    // variable, so nothing connects it to the back stack. Registered only when
+                    // there is somewhere to go back to, so Back still exits from the run screen.
+                    androidx.activity.compose.BackHandler(enabled = screen != Screen.RUN) {
+                        screen = Screen.RUN
+                    }
+
                     when (screen) {
                       Screen.SETTINGS -> SettingsScreen(
                         settings = vm.settings,
-                        archivedRuns = vm.state.history.size,
+                        archivedRuns = vm.archivedRunCount(),
                         archivedBytes = vm.archivedBytes(),
                         onApplyRetention = vm::applyRetention,
                         onDeleteAll = vm::deleteAllRuns,

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

@@ -164,8 +164,10 @@ class RunStore(context: Context, private val settings: Settings) {
             )
             val body = redactedForUpload(docJson, level)
             val reply = client.uploadRun(settings.serverCredential, body)
-            archive.markUploaded(runId, profile.name)
+            archive.markUploaded(runId, profile.name, level.wire)
-            UploadOutcome.Sent(profile.name, "as $level, ${body.toByteArray().size} bytes: ${reply.take(120)}")
+            // Deliberately not echoing `reply`: it is the server's index entry as raw JSON, and
+            // it ended up rendered verbatim in the UI. Size and level are what a person wants.
+            UploadOutcome.Sent(profile.name, "as $level, ${body.toByteArray().size} bytes")
         } catch (e: VersionRefused) {
             UploadOutcome.Incompatible(e.message ?: "the server refused this app's version")
         } catch (e: UploadRefused) {

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

@@ -196,6 +196,10 @@ class RunViewModel(app: Application) : AndroidViewModel(app) {
 
     fun archivedBytes(): Long = store.totalBytes()
 
+    /** Counted from the archive itself, not from [UiState.history], which is empty until the
+     *  history screen has been opened - the two disagreeing read as data loss. */
+    fun archivedRunCount(): Int = store.list().size
+
     /** Redeems an enrollment link, from a paste or from an echolot:// deep link. */
     fun enroll(link: String, deviceName: String? = android.os.Build.MODEL) {
         viewModelScope.launch {
@@ -352,8 +356,9 @@ class RunViewModel(app: Application) : AndroidViewModel(app) {
                 when {
                     ev.contains("\"captive_portal\"") -> out.add(
                         Finding(
-                            id = ids.uuid(), code = "connectivity.captive_portal", category = Category.CONNECTIVITY,
+                            id = ids.uuid(), code = FindingRegistry.CAPTIVE_PORTAL.code,
-                            severity = Severity.MEDIUM, confidence = Confidence.HIGH,
+                            category = FindingRegistry.CAPTIVE_PORTAL.category,
+                            severity = FindingRegistry.CAPTIVE_PORTAL.severity, confidence = Confidence.HIGH,
                             title = "Captive portal intercepting connections",
                             description = "The generate_204 check returned a redirect or a page instead of HTTP 204 — a captive portal (login/splash page) is intercepting traffic on this network.",
                             evidenceRefs = listOf(EvidenceRef(t.id)),
@@ -361,8 +366,9 @@ class RunViewModel(app: Application) : AndroidViewModel(app) {
                     )
                     t.status == TestStatus.FAILED -> out.add(
                         Finding(
-                            id = ids.uuid(), code = "connectivity.no_internet", category = Category.CONNECTIVITY,
+                            id = ids.uuid(), code = FindingRegistry.NO_INTERNET.code,
-                            severity = Severity.HIGH, confidence = Confidence.HIGH,
+                            category = FindingRegistry.NO_INTERNET.category,
+                            severity = FindingRegistry.NO_INTERNET.severity, confidence = Confidence.HIGH,
                             title = "No working internet on any network",
                             description = "Android's own generate_204 connectivity checks failed on every active network (no HTTP 204) — this device has no validated internet path.",
                             evidenceRefs = listOf(EvidenceRef(t.id)),
@@ -375,8 +381,9 @@ class RunViewModel(app: Application) : AndroidViewModel(app) {
                 if (ev.contains("MISMATCH")) {
                     out.add(
                         Finding(
-                            id = ids.uuid(), code = "dns.answer_rewritten", category = Category.DNS,
+                            id = ids.uuid(), code = FindingRegistry.DNS_ANSWER_REWRITTEN.code,
-                            severity = Severity.HIGH, confidence = Confidence.HIGH,
+                            category = FindingRegistry.DNS_ANSWER_REWRITTEN.category,
+                            severity = FindingRegistry.DNS_ANSWER_REWRITTEN.severity, confidence = Confidence.HIGH,
                             title = "DNS answers are being rewritten",
                             description = "A canary reference record returned different RDATA than the spec-defined ground truth — something on the path is rewriting DNS answers (interception, filtering, or a middlebox).",
                             evidenceRefs = listOf(EvidenceRef(t.id)),
@@ -385,8 +392,9 @@ class RunViewModel(app: Application) : AndroidViewModel(app) {
                 } else if (ev.contains("\"reached_authoritative\":false")) {
                     out.add(
                         Finding(
-                            id = ids.uuid(), code = "dns.authoritative_unreachable", category = Category.DNS,
+                            id = ids.uuid(), code = FindingRegistry.DNS_AUTHORITATIVE_UNREACHABLE.code,
-                            severity = Severity.MEDIUM, confidence = Confidence.MEDIUM,
+                            category = FindingRegistry.DNS_AUTHORITATIVE_UNREACHABLE.category,
+                            severity = FindingRegistry.DNS_AUTHORITATIVE_UNREACHABLE.severity, confidence = Confidence.MEDIUM,
                             title = "Canary queries don't reach the authoritative server",
                             description = "A per-run nonce name (which cannot be cached) was not answered by the canary server — the resolver is intercepting or failing to reach it.",
                             evidenceRefs = listOf(EvidenceRef(t.id)),
@@ -399,8 +407,9 @@ class RunViewModel(app: Application) : AndroidViewModel(app) {
                 if (ev.contains("address/port-dependent (symmetric NAT")) {
                     out.add(
                         Finding(
-                            id = ids.uuid(), code = "nat.symmetric", category = Category.NAT,
+                            id = ids.uuid(), code = FindingRegistry.NAT_SYMMETRIC.code,
-                            severity = Severity.MEDIUM, confidence = Confidence.HIGH,
+                            category = FindingRegistry.NAT_SYMMETRIC.category,
+                            severity = FindingRegistry.NAT_SYMMETRIC.severity, confidence = Confidence.HIGH,
                             title = "Symmetric NAT — peer-to-peer connections need a relay",
                             description = "The NAT assigns a different external port per destination (address/port-dependent mapping). Direct peer-to-peer connections (calls, games, file transfer) will usually fail and fall back to relays.",
                             evidenceRefs = listOf(EvidenceRef(t.id)),
@@ -417,8 +426,9 @@ class RunViewModel(app: Application) : AndroidViewModel(app) {
                 if (ipv6Provisioned(networks)) {
                     out.add(
                         Finding(
-                            id = ids.uuid(), code = "ipv6.broken", category = Category.IPV6,
+                            id = ids.uuid(), code = FindingRegistry.V6_BROKEN.code,
-                            severity = Severity.MEDIUM, confidence = Confidence.HIGH,
+                            category = FindingRegistry.V6_BROKEN.category,
+                            severity = FindingRegistry.V6_BROKEN.severity, confidence = Confidence.HIGH,
                             title = "IPv6 is configured but not working",
                             description = "This network advertises IPv6 (a global address and/or a default route), but ICMPv6 got no reply on any network. Half-configured IPv6 is worse than none: connections try IPv6 first and stall before falling back.",
                             evidenceRefs = listOf(EvidenceRef(t.id)),
@@ -427,8 +437,9 @@ class RunViewModel(app: Application) : AndroidViewModel(app) {
                 } else {
                     out.add(
                         Finding(
-                            id = ids.uuid(), code = "ipv6.not_offered", category = Category.IPV6,
+                            id = ids.uuid(), code = FindingRegistry.V6_NOT_OFFERED.code,
-                            severity = Severity.INFO, confidence = Confidence.HIGH,
+                            category = FindingRegistry.V6_NOT_OFFERED.category,
+                            severity = FindingRegistry.V6_NOT_OFFERED.severity, confidence = Confidence.HIGH,
                             title = "IPv4-only network (no IPv6 offered)",
                             description = "No IPv6 address or default route was provisioned, so IPv6 tests could not run. This is normal — many networks are still IPv4-only and it is not a fault.",
                             evidenceRefs = listOf(EvidenceRef(t.id)),

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

@@ -4,6 +4,7 @@
 package app.echo_lot.app
 
 import androidx.compose.foundation.layout.Arrangement
+import androidx.compose.foundation.layout.safeDrawingPadding
 import androidx.compose.foundation.layout.Column
 import androidx.compose.foundation.layout.Row
 import androidx.compose.foundation.layout.Spacer
@@ -66,7 +67,7 @@ fun SettingsScreen(
     var serverCred by remember { mutableStateOf(settings.serverCredential) }
 
     Column(
-        Modifier.fillMaxWidth().verticalScroll(rememberScrollState()).padding(16.dp),
+        Modifier.fillMaxWidth().safeDrawingPadding().verticalScroll(rememberScrollState()).padding(16.dp),
         verticalArrangement = Arrangement.spacedBy(12.dp),
     ) {
         Row(verticalAlignment = Alignment.CenterVertically) {

echolot-app/core-archive/src/main/kotlin/app/echo_lot/archive/RunArchive.kt

@@ -31,10 +31,23 @@ data class ArchivedRun(
     val verdict: String? = null,
     @SerialName("finding_count") val findingCount: Int = 0,
     @SerialName("size_bytes") val sizeBytes: Long = 0,
+    /**
+     * How the *archived* document is redacted. Always "full" in practice, because the archive
+     * deliberately keeps the unredacted run - see the package doc. This is not what was uploaded.
+     */
     val anonymization: String = "full",
     /** Whether this run has been accepted by a server, so history can show what is backed up. */
     val uploaded: Boolean = false,
     @SerialName("uploaded_to") val uploadedTo: String? = null,
+    /**
+     * The level the run was *uploaded* at, which is a different document from the archived one.
+     *
+     * Kept separately because conflating the two is actively misleading: the history row showed
+     * the archive's own level ("full") directly beneath "uploaded to fmr", which reads as "the
+     * complete data was uploaded" when a redacted copy had been sent. A privacy display that
+     * overstates what left the device is worse than none.
+     */
+    @SerialName("uploaded_as") val uploadedAs: String? = null,
 )
 
 /**
@@ -119,7 +132,7 @@ class RunArchive(private val dir: File, private val now: () -> Long = System::cu
     fun deleteAll(): Int = list().count { delete(it.id) }
 
     /** Records that a server accepted this run, so history can distinguish backed-up from local. */
-    fun markUploaded(id: String, serverName: String) {
+    fun markUploaded(id: String, serverName: String, uploadedAs: String? = null) {
         val f = File(dir, safe(id) + META_EXT)
         val meta = runCatching { json.decodeFromString(ArchivedRun.serializer(), f.readText()) }.getOrNull()
             ?: return
@@ -127,7 +140,7 @@ class RunArchive(private val dir: File, private val now: () -> Long = System::cu
             f,
             json.encodeToString(
                 ArchivedRun.serializer(),
-                meta.copy(uploaded = true, uploadedTo = serverName),
+                meta.copy(uploaded = true, uploadedTo = serverName, uploadedAs = uploadedAs),
             ),
         )
     }
@@ -181,7 +194,10 @@ class RunArchive(private val dir: File, private val now: () -> Long = System::cu
             id = id,
             savedAtEpochMs = now(),
             startedAt = run["started_at"]?.jsonPrimitive?.content,
-            verdict = doc["summary"]?.jsonObject?.get("verdict")?.jsonPrimitive?.content,
+            // The schema calls it `overall` (Summary.overall); reading `verdict` here silently
+            // yielded null for every run, so the history list's most prominent element - the
+            // coloured verdict - was blank on every row.
+            verdict = doc["summary"]?.jsonObject?.get("overall")?.jsonPrimitive?.content,
             findingCount = (doc["findings"] as? kotlinx.serialization.json.JsonArray)?.size ?: 0,
             sizeBytes = size,
             anonymization = run["privacy"]?.jsonObject?.get("anonymization")?.jsonPrimitive?.content ?: "full",

echolot-app/core-archive/src/test/kotlin/app/echo_lot/archive/RunArchiveTest.kt

@@ -25,7 +25,7 @@ class RunArchiveTest {
     private fun doc(id: String, findings: Int = 1, pad: Int = 0): String {
         val f = (1..findings).joinToString(",") { """{"id":"f$it"}""" }
         return """{"run":{"id":"$id","started_at":"2026-08-01T10:00:00Z","privacy":{"anonymization":"balanced"}},""" +
-            """"findings":[$f],"summary":{"verdict":"warn"},"pad":"${"x".repeat(pad)}"}"""
+            """"findings":[$f],"summary":{"overall":"warn"},"pad":"${"x".repeat(pad)}"}"""
     }
 
     @Test
@@ -120,13 +120,40 @@ class RunArchiveTest {
     fun uploadStateIsRecorded() {
         val a = archive()
         a.save(doc("run-1"))
-        a.markUploaded("run-1", "fmr")
+        a.markUploaded("run-1", "fmr", "balanced")
         val meta = a.list().single()
         assertTrue(meta.uploaded)
         assertEquals("fmr", meta.uploadedTo)
         assertEquals("run-1", meta.id, "marking upload must not disturb the rest of the entry")
     }
 
+    // The archive's own level and the level a run was uploaded at describe *different documents*.
+    // Showing the archive's ("full", because the archive is deliberately unredacted) next to
+    // "uploaded to fmr" reads as "the complete data was uploaded" when a redacted copy was sent —
+    // a privacy display that overstates what left the device is worse than none.
+    @Test
+    fun theUploadedLevelIsRecordedSeparatelyFromTheArchivedOne() {
+        val a = archive()
+        // A real archived document carries no privacy stamp: the anonymizer never runs on the
+        // archive. The shared doc() fixture has one, which is exactly the unrealism that let this
+        // confusion through in the first place.
+        a.save("""{"run":{"id":"run-1"},"findings":[],"summary":{"overall":"green"}}""")
+        a.markUploaded("run-1", "fmr", "balanced")
+        val meta = a.list().single()
+        assertEquals("full", meta.anonymization, "the archived copy is unredacted, by design")
+        assertEquals("balanced", meta.uploadedAs, "the uploaded copy was redacted, and must say so")
+    }
+
+    // The verdict is read from `summary.overall` — the schema's actual field name. Reading
+    // `summary.verdict` silently yielded null for every run, so the history list's most prominent
+    // element was blank on every row while everything else looked fine.
+    @Test
+    fun theVerdictComesFromTheSchemasOverallField() {
+        val a = archive()
+        a.save("""{"run":{"id":"r1"},"findings":[],"summary":{"overall":"yellow"}}""")
+        assertEquals("yellow", a.list().single().verdict)
+    }
+
     @Test
     fun deleteRemovesBothFiles() {
         val a = archive()

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

@@ -118,7 +118,7 @@ class DownstreamMeasurement(private val ids: IdSource) {
         if (!inOrder) {
             findings.add(
                 finding(
-                    "mtu.fragments_blocked", Category.MTU, Severity.MEDIUM, testId,
+                    FindingRegistry.FRAGMENTS_BLOCKED, testId,
                     "IP fragments do not reach this device",
                     "A fragmented datagram sent in the normal order never arrived. Anything that " +
                         "relies on fragmentation — large DNS answers over UDP, some VPN traffic — " +
@@ -133,7 +133,7 @@ class DownstreamMeasurement(private val ids: IdSource) {
             }.joinToString(" or ")
             findings.add(
                 finding(
-                    "mtu.fragment_reorder_sensitive", Category.MTU, Severity.LOW, testId,
+                    FindingRegistry.FRAGMENT_REORDER_SENSITIVE, testId,
                     "Fragments are dropped when they arrive $which",
                     "In-order fragments are delivered, but the same datagram sent $which is not. " +
                         "Something on the path only reassembles when the first fragment (the one " +
@@ -198,7 +198,7 @@ class DownstreamMeasurement(private val ids: IdSource) {
             if (ipMtu < 1500) {
                 findings.add(
                     finding(
-                        "mtu.reduced_downstream", Category.MTU, Severity.LOW, df.test.id,
+                        FindingRegistry.MTU_REDUCED_DOWNSTREAM, df.test.id,
                         "Downstream path MTU is $ipMtu bytes, below 1500",
                         "The largest datagram that reached this device without fragmenting was " +
                             "$pathMtu bytes of payload ($ipMtu on the wire). Tunnels (PPPoE, VPN, " +
@@ -213,7 +213,7 @@ class DownstreamMeasurement(private val ids: IdSource) {
             if (fragLargest <= pathMtu && sizes.any { it > pathMtu }) {
                 findings.add(
                     finding(
-                        "mtu.downstream_blackhole", Category.MTU, Severity.MEDIUM, frag.test.id,
+                        FindingRegistry.MTU_DOWNSTREAM_BLACKHOLE, frag.test.id,
                         "Datagrams above $pathMtu bytes are dropped downstream, fragmented or not",
                         "Nothing larger than $pathMtu bytes arrived, even when the network was " +
                             "free to fragment it. Traffic that relies on large responses will " +
@@ -226,7 +226,7 @@ class DownstreamMeasurement(private val ids: IdSource) {
         if (train.received == 0) {
             findings.add(
                 finding(
-                    "connectivity.downstream_blocked", Category.CONNECTIVITY, Severity.HIGH, train.test.id,
+                    FindingRegistry.DOWNSTREAM_BLOCKED, train.test.id,
                     "No server-initiated packets arrived",
                     "The server sent ${train.sent} packets toward this device and none arrived, " +
                         "while the round-trip echo worked. Something on the path forwards replies " +
@@ -236,7 +236,7 @@ class DownstreamMeasurement(private val ids: IdSource) {
         } else if (train.lossPct >= 5.0) {
             findings.add(
                 finding(
-                    "connectivity.downstream_loss", Category.CONNECTIVITY, Severity.MEDIUM, train.test.id,
+                    FindingRegistry.LOSS_DOWNSTREAM, train.test.id,
                     "Downstream loss of ${round1(train.lossPct)}%",
                     "${train.sent - train.received} of ${train.sent} packets sent toward this " +
                         "device were lost. Downstream loss is invisible to a round-trip test, " +
@@ -247,7 +247,7 @@ class DownstreamMeasurement(private val ids: IdSource) {
         if (train.reordered > 0) {
             findings.add(
                 finding(
-                    "connectivity.downstream_reorder", Category.CONNECTIVITY, Severity.LOW, train.test.id,
+                    FindingRegistry.DOWNSTREAM_REORDER, train.test.id,
                     "${train.reordered} downstream packet(s) arrived out of order",
                     "Packets arrived in a different order than they were sent. Usually per-packet " +
                         "load balancing across links; harmless for most traffic, not for all of it.",
@@ -414,9 +414,17 @@ class DownstreamMeasurement(private val ids: IdSource) {
 
     // ---- helpers ----------------------------------------------------------------------
 
-    private fun finding(code: String, cat: Category, sev: Severity, testId: String, title: String, desc: String) =
+    /**
+     * Builds a finding from a registry entry, which supplies the code, category and severity.
+     *
+     * Taking a [FindingSpec] rather than three loose values is the point: a typo becomes a
+     * compile error, and two call sites cannot disagree about which category a finding belongs
+     * to - a disagreement that would split one fault across two verdict lights.
+     */
+    private fun finding(spec: FindingSpec, testId: String, title: String, desc: String) =
         Finding(
-            id = ids.uuid(), code = code, category = cat, severity = sev, confidence = Confidence.HIGH,
+            id = ids.uuid(), code = spec.code, category = spec.category, severity = spec.severity,
+            confidence = Confidence.HIGH,
             title = title, description = desc, evidenceRefs = listOf(EvidenceRef(testId)),
         )
 

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

@@ -47,6 +47,14 @@ class ServerMeasurement(
          * it is a flag rather than an assumption.
          */
         val downstream: Boolean = true,
+        /**
+         * Throughput moves real data — a 5-second run at 50 Mbps is about 30 MB — so it is off
+         * unless asked for. On a metered mobile connection that is the user's money, and a
+         * measurement tool that spends it without being told to is not one people keep installed.
+         */
+        val throughput: Boolean = false,
+        @Suppress("unused") val throughputSeconds: Int = 5,
+        @Suppress("unused") val throughputKbps: Int = 50_000,
     )
 
     fun run(cfg: Config): MeasurementDocument {
@@ -89,6 +97,15 @@ class ServerMeasurement(
                 tests.addAll(dsTests)
                 allFindings.addAll(dsFindings)
             }
+
+            if (cfg.throughput && profile.supports("throughput")) {
+                val (tpTest, tpFindings) = ThroughputMeasurement(ids).run(
+                    cfg.credential, session.sessionId, control, ps, sessionRef = "sess-1",
+                    durationS = cfg.throughputSeconds, kbps = cfg.throughputKbps,
+                )
+                tests.add(tpTest)
+                allFindings.addAll(tpFindings)
+            }
         }
 
         control.deleteSession(cfg.credential, session.sessionId)
@@ -191,11 +208,11 @@ class ServerMeasurement(
 
         val findings = ArrayList<Finding>()
         if (received == 0) {
-            findings.add(finding("nat.udp_unreachable", Category.CONNECTIVITY, Severity.HIGH, testId,
+            findings.add(finding(FindingRegistry.UDP_UNREACHABLE, testId,
                 "No UDP echo replies from the server",
                 "Every ECHO probe to the server's UDP data plane was lost — the path blocks or drops the session's UDP traffic."))
         } else if (lossPct >= 20.0) {
-            findings.add(finding("connectivity.udp_loss", Category.CONNECTIVITY, Severity.MEDIUM, testId,
+            findings.add(finding(FindingRegistry.UDP_LOSS, testId,
                 "High UDP loss to the server (${round1(lossPct)}%)",
                 "A large fraction of ECHO probes were lost, indicating an unreliable UDP path."))
         }
@@ -203,14 +220,14 @@ class ServerMeasurement(
         directional?.let { d ->
             when {
                 d.noneReachedServer && received == 0 -> findings.add(
-                    finding("nat.udp_unreachable_upstream", Category.CONNECTIVITY, Severity.HIGH, testId,
+                    finding(FindingRegistry.UDP_UNREACHABLE_UPSTREAM, testId,
                         "Nothing reached the server",
                         "The server received none of the ${d.sent} probes, so the traffic is being " +
                             "dropped on the way out, not on the way back. A firewall or NAT on " +
                             "this side of the path is the place to look."),
                 )
                 d.lossUpstreamPct >= 2.0 -> findings.add(
-                    finding("connectivity.loss_upstream", Category.CONNECTIVITY, Severity.MEDIUM, testId,
+                    finding(FindingRegistry.LOSS_UPSTREAM, testId,
                         "${d.lossUpstreamPct} % of probes were lost on the way to the server",
                         "${d.lostUpstream} of ${d.sent} probes never reached the server. The " +
                             "return path is not implicated: replies came back for everything that " +
@@ -219,7 +236,7 @@ class ServerMeasurement(
             }
             if (d.lossDownstreamPct >= 2.0) {
                 findings.add(
-                    finding("connectivity.loss_downstream", Category.CONNECTIVITY, Severity.MEDIUM, testId,
+                    finding(FindingRegistry.LOSS_DOWNSTREAM, testId,
                         "${d.lossDownstreamPct} % of replies were lost on the way back",
                         "The server received ${d.seenByServer} probes and answered them, but " +
                             "${d.lostDownstream} of those replies never arrived. The outbound path " +
@@ -229,7 +246,7 @@ class ServerMeasurement(
         }
 
         if (natRebinding) {
-            findings.add(finding("nat.udp_rebinding", Category.NAT, Severity.MEDIUM, testId,
+            findings.add(finding(FindingRegistry.NAT_UDP_REBINDING, testId,
                 "NAT remapped the UDP source port mid-flow",
                 "The server observed more than one source port for this session (${observedPorts.joinToString()}), i.e. a NAT with a short UDP mapping or per-packet remapping."))
         }
@@ -259,9 +276,17 @@ class ServerMeasurement(
         }
     }
 
-    private fun finding(code: String, cat: Category, sev: Severity, testId: String, title: String, desc: String) =
+    /**
+     * Builds a finding from a registry entry, which supplies the code, category and severity.
+     *
+     * Taking a [FindingSpec] rather than three loose values is the point: a typo becomes a
+     * compile error, and two call sites cannot disagree about which category a finding belongs
+     * to - a disagreement that would split one fault across two verdict lights.
+     */
+    private fun finding(spec: FindingSpec, testId: String, title: String, desc: String) =
         Finding(
-            id = ids.uuid(), code = code, category = cat, severity = sev, confidence = Confidence.HIGH,
+            id = ids.uuid(), code = spec.code, category = spec.category, severity = spec.severity,
+            confidence = Confidence.HIGH,
             title = title, description = desc, evidenceRefs = listOf(EvidenceRef(testId)),
         )
 

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

@@ -0,0 +1,338 @@
+// 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.ControlClient
+import app.echo_lot.protocol.ProbeSession
+import app.echo_lot.protocol.Wire
+import kotlinx.serialization.SerialName
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.encodeToJsonElement
+import kotlinx.serialization.json.jsonArray
+import kotlinx.serialization.json.jsonObject
+import kotlinx.serialization.json.jsonPrimitive
+
+/**
+ * Downstream throughput: the server sends at a paced rate for a bounded time and the client
+ * measures what arrives (`perf.throughput_udp`).
+ *
+ * The number this produces is only meaningful with a qualifier attached, and getting that
+ * qualifier right is most of the work here. A throughput test reports the *smallest* limit on the
+ * path, and the sender's own ceiling is one of the candidates: if the server was asked for 50 Mbps
+ * and 50 Mbps arrived, the network was never the constraint and "50 Mbps" says nothing about it.
+ * Reporting that as a capacity measurement would be a confident lie, so the result always carries
+ * [ThroughputMetrics.limitedBy] and a finding is only raised when the network is actually
+ * implicated.
+ *
+ * Comparing against the *sender's* count rather than the requested rate is the other half: the
+ * server reports how much it actually put on the wire, and the gap between that and what arrived
+ * is the loss. A receiver alone cannot tell "the network dropped it" from "the sender never sent
+ * it", and guessing turns a healthy server-side limit into a phantom network fault.
+ */
+class ThroughputMeasurement(private val ids: IdSource) {
+
+    private val json = Json { encodeDefaults = true; explicitNulls = true }
+
+    fun run(
+        credential: String,
+        sessionId: String,
+        control: ControlClient,
+        probe: ProbeSession,
+        sessionRef: String,
+        durationS: Int = 5,
+        kbps: Int = 50_000,
+        sizeBytes: Int = 1200,
+    ): Pair<Test, List<Finding>> {
+        val testId = ids.uuid()
+        val started = ids.monoNs()
+
+        val reply = runCatching {
+            control.action(
+                credential, sessionId,
+                """{"action":"throughput","direction":"down","duration_s":$durationS,""" +
+                    """"kbps":$kbps,"size_bytes":$sizeBytes}""",
+            )
+        }
+        if (reply.isFailure) {
+            return Test(
+                id = testId, type = TestType.PERF_THROUGHPUT_UDP, sessionRef = sessionRef, tier = Tier.APP,
+                startedMonoNs = started, endedMonoNs = ids.monoNs(),
+                status = TestStatus.UNSUPPORTED,
+                error = TestError("action_refused", reply.exceptionOrNull()?.message ?: "throughput refused"),
+            ) to emptyList()
+        }
+
+        // The server may have shortened the run to fit its own byte budget; listen for what it
+        // actually promised, not for what we asked.
+        val plannedMs = parseInt(reply.getOrNull(), "duration_ms") ?: (durationS * 1000)
+
+        // A margin past the planned end so the tail of the run is not counted as loss: packets
+        // still in flight when we stop listening were not dropped, they were merely late.
+        val received = probe.collectGranted(plannedMs + 1_500L)
+            .filter { it.type == Wire.TYPE_THROUGHPUT_DATA }
+
+        val bytes = received.sumOf { it.sizeBytes.toLong() }
+        val spanNs = if (received.size >= 2) {
+            received.maxOf { it.tRxNs } - received.minOf { it.tRxNs }
+        } else {
+            0L
+        }
+        // Measured over the arrival span rather than our listening window, which includes the
+        // request round trip and the trailing margin and would understate the rate.
+        val receivedKbps = if (spanNs > 0) (bytes * 8 * 1_000_000 / spanNs).toInt() else 0
+
+        val sender = senderReport(control, credential, sessionId)
+        val sentPackets = sender?.packets ?: 0
+        val lossPct = if (sentPackets > 0) {
+            round2((sentPackets - received.size).coerceAtLeast(0) * 100.0 / sentPackets)
+        } else {
+            null
+        }
+
+        // Only a run the *clock* ended measured the network. One stopped by our own byte budget
+        // or rate ceiling measured this server.
+        val limitedBy = sender?.limitedBy ?: "unknown"
+        val networkLimited = limitedBy == "duration" &&
+            sender != null && receivedKbps > 0 && receivedKbps < sender.kbps * 9 / 10
+
+        val metrics = json.encodeToJsonElement(
+            ThroughputMetrics(
+                requestedKbps = kbps,
+                plannedDurationMs = plannedMs,
+                packetsReceived = received.size,
+                bytesReceived = bytes,
+                receivedKbps = receivedKbps,
+                senderPackets = sender?.packets,
+                senderBytes = sender?.bytes,
+                senderKbps = sender?.kbps,
+                lossPct = lossPct,
+                limitedBy = limitedBy,
+                measuresNetwork = networkLimited,
+            ),
+        ) as JsonObject
+
+        val findings = ArrayList<Finding>()
+        when {
+            sender == null -> Unit // no sender report: nothing can be concluded, so nothing is
+            received.isEmpty() -> findings.add(
+                finding(
+                    FindingRegistry.THROUGHPUT_NO_DELIVERY, testId,
+                    "No throughput traffic arrived",
+                    "The server sent ${sender.packets} packets and none arrived. This is a " +
+                        "connectivity fault rather than a slow link.",
+                ),
+            )
+            networkLimited -> findings.add(
+                finding(
+                    FindingRegistry.THROUGHPUT_BELOW_OFFERED, testId,
+                    "Downstream throughput ${receivedKbps / 1000} Mbit/s, below the " +
+                        "${sender.kbps / 1000} Mbit/s offered",
+                    "The server sent at ${sender.kbps / 1000} Mbit/s for the full run and " +
+                        "${receivedKbps / 1000} Mbit/s arrived" +
+                        (lossPct?.let { ", losing $it % of packets" } ?: "") +
+                        ". The path could not carry what was offered.",
+                ),
+            )
+        }
+
+        return Test(
+            id = testId, type = TestType.PERF_THROUGHPUT_UDP, sessionRef = sessionRef, tier = Tier.APP,
+            startedMonoNs = started, endedMonoNs = ids.monoNs(),
+            status = if (received.isEmpty()) TestStatus.FAILED else TestStatus.OK,
+            metrics = metrics,
+        ) to findings
+    }
+
+    /**
+     * Upstream throughput: the client sends, the server counts.
+     *
+     * The mirror image of the downstream case, and it needs no grant — the client is generating
+     * its own traffic, so there is no amplification to gate. What it does need is the server's
+     * count: only the far end knows how much arrived, and without that number a sender can
+     * measure how fast it can *transmit*, which is not the same question and is usually just the
+     * speed of the local NIC.
+     */
+    fun runUpstream(
+        credential: String,
+        sessionId: String,
+        control: ControlClient,
+        probe: ProbeSession,
+        sessionRef: String,
+        durationS: Int = 5,
+        kbps: Int = 20_000,
+        sizeBytes: Int = 1200,
+    ): Pair<Test, List<Finding>> {
+        val testId = ids.uuid()
+        val started = ids.monoNs()
+
+        // Zeroes the server's counter so this run measures itself rather than inheriting the
+        // packets of an earlier one on the same session.
+        val reply = runCatching {
+            control.action(credential, sessionId, """{"action":"throughput","direction":"up"}""")
+        }
+        if (reply.isFailure) {
+            return Test(
+                id = testId, type = TestType.PERF_THROUGHPUT_UDP, sessionRef = sessionRef, tier = Tier.APP,
+                startedMonoNs = started, endedMonoNs = ids.monoNs(),
+                status = TestStatus.UNSUPPORTED,
+                error = TestError("action_refused", reply.exceptionOrNull()?.message ?: "refused"),
+            ) to emptyList()
+        }
+
+        val sent = probe.sendThroughput(durationS * 1000L, kbps, sizeBytes)
+        // A moment for the tail of the run to arrive; counting still-in-flight packets as lost
+        // would inflate the loss figure by whatever the path's delay happens to be.
+        Thread.sleep(500)
+        val seen = upstreamCount(control, credential, sessionId)
+
+        val lossPct = if (sent.packets > 0 && seen != null) {
+            round2((sent.packets - seen.packets).coerceAtLeast(0) * 100.0 / sent.packets)
+        } else {
+            null
+        }
+        // The receiver's rate is the measurement. The sender's is what we managed to emit, which
+        // is a property of this phone and its radio, not of the network.
+        val achievedKbps = seen?.kbps ?: 0
+
+        val metrics = json.encodeToJsonElement(
+            UpstreamThroughputMetrics(
+                requestedKbps = kbps,
+                sentPackets = sent.packets,
+                sentBytes = sent.bytes,
+                sentKbps = sent.kbps,
+                receivedPackets = seen?.packets,
+                receivedBytes = seen?.bytes,
+                receivedKbps = achievedKbps,
+                lossPct = lossPct,
+                // Same honesty rule as downstream: if what arrived matches what we offered, the
+                // path was never the constraint and this number says nothing about it.
+                measuresNetwork = seen != null && achievedKbps > 0 && achievedKbps < sent.kbps * 9 / 10,
+            ),
+        ) as JsonObject
+
+        val findings = ArrayList<Finding>()
+        if (seen != null && seen.packets == 0 && sent.packets > 0) {
+            findings.add(
+                finding(
+                    FindingRegistry.THROUGHPUT_NO_DELIVERY, testId,
+                    "No upstream traffic reached the server",
+                    "This device sent ${sent.packets} packets and the server received none. " +
+                        "That is a connectivity fault on the outbound path rather than a slow link.",
+                ),
+            )
+        } else if (lossPct != null && lossPct >= 2.0) {
+            findings.add(
+                finding(
+                    FindingRegistry.THROUGHPUT_BELOW_OFFERED, testId,
+                    "Upstream loss of $lossPct % at ${sent.kbps / 1000} Mbit/s",
+                    "The server received ${seen?.packets} of the ${sent.packets} packets this " +
+                        "device sent. The outbound path could not carry what was offered.",
+                ),
+            )
+        }
+
+        return Test(
+            id = testId, type = TestType.PERF_THROUGHPUT_UDP, sessionRef = sessionRef, tier = Tier.APP,
+            startedMonoNs = started, endedMonoNs = ids.monoNs(),
+            status = if (seen == null || seen.packets == 0) TestStatus.FAILED else TestStatus.OK,
+            metrics = metrics,
+        ) to findings
+    }
+
+    private data class UpstreamCount(val packets: Int, val bytes: Long, val kbps: Int)
+
+    /** The server's tally for this session's upstream run. */
+    private fun upstreamCount(
+        control: ControlClient, credential: String, sessionId: String,
+    ): UpstreamCount? = runCatching {
+        val o = Json.parseToJsonElement(control.observations(credential, sessionId))
+            .jsonObject["throughput_up"]?.jsonObject ?: return null
+        UpstreamCount(
+            packets = o["packets"]?.jsonPrimitive?.content?.toIntOrNull() ?: 0,
+            bytes = o["bytes"]?.jsonPrimitive?.content?.toLongOrNull() ?: 0,
+            kbps = o["kbps"]?.jsonPrimitive?.content?.toIntOrNull() ?: 0,
+        )
+    }.getOrNull()
+
+    private data class SenderReport(
+        val packets: Int, val bytes: Long, val kbps: Int, val limitedBy: String,
+    )
+
+    /** The server's own account of the run, from the observations API. */
+    private fun senderReport(
+        control: ControlClient, credential: String, sessionId: String,
+    ): SenderReport? = runCatching {
+        val arr = Json.parseToJsonElement(control.observations(credential, sessionId))
+            .jsonObject["throughput"]?.jsonArray ?: return null
+        val last = arr.lastOrNull()?.jsonObject ?: return null
+        SenderReport(
+            packets = last["packets"]?.jsonPrimitive?.content?.toIntOrNull() ?: 0,
+            bytes = last["bytes"]?.jsonPrimitive?.content?.toLongOrNull() ?: 0,
+            kbps = last["kbps"]?.jsonPrimitive?.content?.toIntOrNull() ?: 0,
+            limitedBy = last["limited_by"]?.jsonPrimitive?.content ?: "unknown",
+        )
+    }.getOrNull()
+
+    private fun parseInt(body: String?, key: String): Int? =
+        body?.let { Regex("\"$key\"\\s*:\\s*(-?\\d+)").find(it)?.groupValues?.get(1)?.toIntOrNull() }
+
+    /**
+     * Builds a finding from a registry entry, which supplies the code, category and severity.
+     *
+     * Taking a [FindingSpec] rather than three loose values is the point: a typo becomes a
+     * compile error, and two call sites cannot disagree about which category a finding belongs
+     * to - a disagreement that would split one fault across two verdict lights.
+     */
+    private fun finding(spec: FindingSpec, testId: String, title: String, desc: String) =
+        Finding(
+            id = ids.uuid(), code = spec.code, category = spec.category, severity = spec.severity,
+            confidence = Confidence.HIGH,
+            title = title, description = desc, evidenceRefs = listOf(EvidenceRef(testId)),
+        )
+
+    private fun round2(v: Double) = Math.round(v * 100.0) / 100.0
+}
+
+/** Metrics for perf.throughput_udp in the upstream direction. */
+@Serializable
+data class UpstreamThroughputMetrics(
+    val direction: String = "up",
+    @SerialName("requested_kbps") val requestedKbps: Int,
+    @SerialName("sent_packets") val sentPackets: Int,
+    @SerialName("sent_bytes") val sentBytes: Long,
+    /** What this device managed to emit — a property of the phone and its radio, not the path. */
+    @SerialName("sent_kbps") val sentKbps: Int,
+    @SerialName("received_packets") val receivedPackets: Int? = null,
+    @SerialName("received_bytes") val receivedBytes: Long? = null,
+    /** What arrived, measured by the only party that can measure it. This is the result. */
+    @SerialName("received_kbps") val receivedKbps: Int,
+    @SerialName("loss_pct") val lossPct: Double? = null,
+    @SerialName("measures_network") val measuresNetwork: Boolean,
+)
+
+/** Metrics for perf.throughput_udp. */
+@Serializable
+data class ThroughputMetrics(
+    val direction: String = "down",
+    @SerialName("requested_kbps") val requestedKbps: Int,
+    @SerialName("planned_duration_ms") val plannedDurationMs: Int,
+    @SerialName("packets_received") val packetsReceived: Int,
+    @SerialName("bytes_received") val bytesReceived: Long,
+    @SerialName("received_kbps") val receivedKbps: Int,
+    @SerialName("sender_packets") val senderPackets: Int? = null,
+    @SerialName("sender_bytes") val senderBytes: Long? = null,
+    @SerialName("sender_kbps") val senderKbps: Int? = null,
+    /** Against the sender's count, so a server-side limit is never counted as network loss. */
+    @SerialName("loss_pct") val lossPct: Double? = null,
+    /** What ended the run: duration | budget | rate | send_error | unknown. */
+    @SerialName("limited_by") val limitedBy: String,
+    /**
+     * Whether this number says anything about the network. False when the sender's own ceiling
+     * was the binding constraint — in which case the rate is a property of the test, not the path.
+     */
+    @SerialName("measures_network") val measuresNetwork: Boolean,
+)

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

@@ -44,7 +44,9 @@ class LiveDownstreamTest {
         for (t in tests) println("${t.type}  status=${t.status}  metrics=${t.metrics}")
         for (f in findings) println("finding ${f.code} [${f.severity}] ${f.title}")
 
-        assertEquals(3, tests.size, "expected pmtud_down, frag_delivery and a downstream train")
+        // Assert on what is present, not on how many: adding a measurement should not be a
+        // test edit. (It was, once — hence the note.)
+        assertTrue(tests.size >= 3, "expected at least the three downstream tests, got ${tests.size}")
         val byType = tests.associateBy { it.type }
 
         val pmtud = assertNotNull(byType[TestType.MTU_PMTUD_DOWN], "no mtu.pmtud_down test")
@@ -58,6 +60,17 @@ class LiveDownstreamTest {
         val frag = assertNotNull(byType[TestType.MTU_FRAG_DELIVERY], "no mtu.frag_delivery test")
         assertNotNull(frag.metrics?.get("largest_delivered_bytes"))
 
+        // Fragment ordering runs only when fragments arrive at all, and only against a server
+        // that can craft them — so it is checked when present rather than required.
+        byType[TestType.MTU_FRAG_ORDERING]?.let { fo ->
+            val m = fo.metrics?.toString() ?: ""
+            println("fragment ordering: ${fo.status} $m")
+            if (fo.status != TestStatus.UNSUPPORTED) {
+                assertTrue(m.contains("in_order"), "no per-ordering result: $m")
+                assertTrue(m.contains("reversed"), "reversed ordering was never attempted: $m")
+            }
+        }
+
         val train = assertNotNull(byType[TestType.TRAIN_UDP_DOWNSTREAM], "no downstream train")
         assertNotNull(train.evidence, "a train without columnar evidence is not recomputable")
         val received = train.metrics?.get("received")?.toString()?.toIntOrNull() ?: 0

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

@@ -0,0 +1,68 @@
+// 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
+
+/**
+ * Downstream throughput against a LIVE server. Self-skips without ECHOLOT_LIVE_*.
+ *
+ * The assertions are about *honesty* rather than speed: a rate is only a measurement if the run
+ * was ended by the clock and the sender's own count backs it up. A test that just asserted "some
+ * Mbps arrived" would pass equally well against a broken implementation.
+ */
+class LiveThroughputTest {
+
+    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 measuresDownstreamRateAndSaysWhatLimitedIt() {
+        if (url == null || pin == null || cred == null || udp == null) {
+            println("LiveThroughputTest 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 grant binds to the observed source
+            ThroughputMeasurement(SystemIdSource()).run(
+                cred, session.sessionId, control, ps, sessionRef = "sess-1",
+                durationS = 3, kbps = 20_000,
+            )
+        }
+        control.deleteSession(cred, session.sessionId)
+
+        val m = assertNotNull(test.metrics).toString()
+        println("throughput: ${test.status} $m")
+        for (f in findings) println("finding ${f.code} [${f.severity}] ${f.title}")
+
+        assertEquals(TestStatus.OK, test.status, "no throughput traffic arrived: $m")
+
+        // The sender's own count must be present — without it, loss cannot be attributed and the
+        // number is not a measurement.
+        assertTrue(m.contains("sender_packets"), "no sender report to compare against: $m")
+        assertTrue(m.contains("limited_by"), "the result must say what ended the run: $m")
+
+        val received = Regex(""""received_kbps":(\d+)""").find(m)?.groupValues?.get(1)?.toInt()
+        assertNotNull(received)
+        assertTrue(received > 0, "measured 0 kbps: $m")
+        println("received ${received / 1000} Mbit/s")
+
+        // A run this short and this far below the ceiling should end on the clock. Anything else
+        // means the grant was the constraint, and then the rate says nothing about the path.
+        assertTrue(m.contains(""""limited_by":"duration""""),
+            "the run did not end on the clock, so the rate measures the server, not the path: $m")
+    }
+}

echolot-app/core-measurement/src/main/kotlin/app/echo_lot/measurement/FindingRegistry.kt

@@ -0,0 +1,205 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package app.echo_lot.measurement
+
+/**
+ * The registry of finding codes (measurement-schema.md §9, open item 1).
+ *
+ * A finding code is the stable, machine-readable half of a result: the prose changes, the code is
+ * what a dashboard groups by and what someone greps a year of archived runs for. That only holds
+ * if a code means exactly one thing forever — which is not something ad-hoc string literals at
+ * fifteen call sites can promise.
+ *
+ * The failure this exists to prevent had already happened by the time it was written. Two
+ * independently-added emitters produced `connectivity.downstream_loss` and
+ * `connectivity.loss_downstream` for the same concept, and nothing anywhere objected. Anyone
+ * aggregating either one would have silently seen half their data.
+ *
+ * So codes are declared here as typed specs, each carrying its category and default severity, and
+ * emitters reference the spec rather than retyping the string. That makes a typo a compile error,
+ * and makes it impossible for two call sites to disagree about which category a finding belongs
+ * to — a disagreement that would otherwise split one fault across two verdict lights.
+ */
+data class FindingSpec(
+    val code: String,
+    val category: Category,
+    /** Severity when nothing about the specific run argues otherwise; emitters may escalate. */
+    val severity: Severity,
+    /** One line: what this finding asserts. Present tense, no hedging. */
+    val meaning: String,
+    /**
+     * What the finding rules *out*, where that is the useful half. "Loss upstream" is worth much
+     * more when it also says the return path is fine, because that halves where to look next.
+     */
+    val rulesOut: String? = null,
+)
+
+object FindingRegistry {
+
+    // ---- connectivity ----------------------------------------------------------------
+
+    // Renamed from nat.* before anything shipped: neither of these is about NAT, and the
+    // prefix is what decides which category - and therefore which verdict light - a finding
+    // rolls up into. A nat.* code landing under connectivity would be a permanent puzzle.
+    val UDP_UNREACHABLE = FindingSpec(
+        "connectivity.udp_unreachable", Category.CONNECTIVITY, Severity.HIGH,
+        "No UDP echo replies came back from the server at all.",
+    )
+
+    val UDP_UNREACHABLE_UPSTREAM = FindingSpec(
+        "connectivity.udp_unreachable_upstream", Category.CONNECTIVITY, Severity.HIGH,
+        "The server received none of the probes, so traffic is dropped on the way out.",
+        rulesOut = "The return path: nothing arrived to be replied to.",
+    )
+
+    val UDP_LOSS = FindingSpec(
+        "connectivity.udp_loss", Category.CONNECTIVITY, Severity.MEDIUM,
+        "A large fraction of round-trip probes were lost, direction unknown.",
+    )
+
+    val LOSS_UPSTREAM = FindingSpec(
+        "connectivity.loss_upstream", Category.CONNECTIVITY, Severity.MEDIUM,
+        "Probes were lost on the way to the server.",
+        rulesOut = "The return path: replies came back for everything that arrived.",
+    )
+
+    /**
+     * The single code for "lost on the return path", whichever measurement found it.
+     *
+     * Two emitters had independently invented `connectivity.downstream_loss` and
+     * `connectivity.loss_downstream` for this, and nothing objected. Anyone aggregating either
+     * one would have silently seen half their data. Paired with [LOSS_UPSTREAM] so the two
+     * directions read as a set.
+     */
+    val LOSS_DOWNSTREAM = FindingSpec(
+        "connectivity.loss_downstream", Category.CONNECTIVITY, Severity.MEDIUM,
+        "Packets were lost on the way back from the server.",
+        rulesOut = "The outbound path: the server received what it was answering.",
+    )
+
+    val DOWNSTREAM_BLOCKED = FindingSpec(
+        "connectivity.downstream_blocked", Category.CONNECTIVITY, Severity.HIGH,
+        "Server-initiated packets never arrive, although round trips work.",
+        rulesOut = "Basic reachability: the path forwards replies, just not unsolicited traffic.",
+    )
+
+    val DOWNSTREAM_REORDER = FindingSpec(
+        "connectivity.downstream_reorder", Category.CONNECTIVITY, Severity.LOW,
+        "Downstream packets arrive in a different order than they were sent.",
+    )
+
+    // MEDIUM, not HIGH: a captive portal is a condition to report, not necessarily a fault - on
+    // hotel or cafe wifi it is exactly what should be there, and logging in clears it. NO_INTERNET
+    // is the HIGH one, because nothing the user does locally fixes that. The registry first said
+    // HIGH; the probe emitting it had always said MEDIUM, and the probe was the considered value.
+    val CAPTIVE_PORTAL = FindingSpec(
+        "connectivity.captive_portal", Category.CONNECTIVITY, Severity.MEDIUM,
+        "A captive portal is intercepting connectivity checks.",
+    )
+
+    val NO_INTERNET = FindingSpec(
+        "connectivity.no_internet", Category.CONNECTIVITY, Severity.HIGH,
+        "Android's own connectivity checks fail on this network.",
+    )
+
+    // ---- mtu -------------------------------------------------------------------------
+
+    val MTU_REDUCED_DOWNSTREAM = FindingSpec(
+        "mtu.reduced_downstream", Category.MTU, Severity.LOW,
+        "The downstream path MTU is below the usual 1500 bytes.",
+    )
+
+    val MTU_DOWNSTREAM_BLACKHOLE = FindingSpec(
+        "mtu.downstream_blackhole", Category.MTU, Severity.MEDIUM,
+        "Datagrams above the path MTU are dropped downstream, fragmented or not.",
+    )
+
+    val FRAGMENTS_BLOCKED = FindingSpec(
+        "mtu.fragments_blocked", Category.MTU, Severity.MEDIUM,
+        "IP fragments do not reach this device even when sent in order.",
+    )
+
+    val FRAGMENT_REORDER_SENSITIVE = FindingSpec(
+        "mtu.fragment_reorder_sensitive", Category.MTU, Severity.LOW,
+        "Fragments are delivered in order but dropped when reordered or delayed.",
+        rulesOut = "Fragmentation itself: in-order fragments arrive fine.",
+    )
+
+    // ---- nat -------------------------------------------------------------------------
+
+    val NAT_UDP_REBINDING = FindingSpec(
+        "nat.udp_rebinding", Category.NAT, Severity.MEDIUM,
+        "A NAT remapped the UDP source port mid-flow.",
+    )
+
+    val NAT_SYMMETRIC = FindingSpec(
+        "nat.symmetric", Category.NAT, Severity.MEDIUM,
+        "The NAT assigns a different external port per destination.",
+    )
+
+    // ---- perf ------------------------------------------------------------------------
+
+    val THROUGHPUT_NO_DELIVERY = FindingSpec(
+        "perf.throughput_no_delivery", Category.PERFORMANCE, Severity.HIGH,
+        "No throughput traffic arrived, although the server sent it.",
+    )
+
+    val THROUGHPUT_BELOW_OFFERED = FindingSpec(
+        "perf.throughput_below_offered", Category.PERFORMANCE, Severity.LOW,
+        "Less throughput arrived than the server sent for the whole run.",
+    )
+
+    // ---- dns -------------------------------------------------------------------------
+
+    val DNS_ANSWER_REWRITTEN = FindingSpec(
+        "dns.answer_rewritten", Category.DNS, Severity.HIGH,
+        "A resolver returned an answer that differs from the authoritative record.",
+    )
+
+    val DNS_AUTHORITATIVE_UNREACHABLE = FindingSpec(
+        "dns.authoritative_unreachable", Category.DNS, Severity.MEDIUM,
+        "The canary zone's authoritative server could not be reached.",
+    )
+
+    // ---- v6 ----------------------------------------------------------------------------
+    //
+    // Prefix is `v6.`, matching the test-type registry (v6.brokenness, v6.happy_eyeballs, ...).
+    // These were `ipv6.*` while declaring Category.IPV6, but the prefix map only knows "v6", so
+    // they silently rolled up under connectivity: the third instance of a prefix disagreeing with
+    // its category and quietly moving a fault to a different verdict light.
+
+    val V6_BROKEN = FindingSpec(
+        "v6.broken", Category.IPV6, Severity.MEDIUM,
+        "IPv6 is configured on this network but does not work.",
+        rulesOut = "Absence of IPv6: it is provisioned, it simply fails.",
+    )
+
+    /**
+     * INFO deliberately, and it needs to stay that way.
+     *
+     * Most networks still do not offer IPv6, and that is not a fault. Reporting it as a warning
+     * lights a yellow verdict on a perfectly healthy network, which teaches people to ignore the
+     * light — the one thing a diagnostic must never do.
+     */
+    val V6_NOT_OFFERED = FindingSpec(
+        "v6.not_offered", Category.IPV6, Severity.INFO,
+        "This network does not offer IPv6.",
+    )
+
+    /** Every registered finding, in declaration order. */
+    val all: List<FindingSpec> = listOf(
+        UDP_UNREACHABLE, UDP_UNREACHABLE_UPSTREAM, UDP_LOSS, LOSS_UPSTREAM, LOSS_DOWNSTREAM,
+        DOWNSTREAM_BLOCKED, DOWNSTREAM_REORDER, CAPTIVE_PORTAL, NO_INTERNET,
+        MTU_REDUCED_DOWNSTREAM, MTU_DOWNSTREAM_BLACKHOLE, FRAGMENTS_BLOCKED,
+        FRAGMENT_REORDER_SENSITIVE,
+        NAT_UDP_REBINDING, NAT_SYMMETRIC,
+        THROUGHPUT_NO_DELIVERY, THROUGHPUT_BELOW_OFFERED,
+        DNS_ANSWER_REWRITTEN, DNS_AUTHORITATIVE_UNREACHABLE,
+        V6_BROKEN, V6_NOT_OFFERED,
+    )
+
+    private val byCode: Map<String, FindingSpec> = all.associateBy { it.code }
+
+    fun byCode(code: String): FindingSpec? = byCode[code]
+}

echolot-app/core-measurement/src/test/kotlin/app/echo_lot/measurement/FindingRegistryTest.kt

@@ -0,0 +1,133 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package app.echo_lot.measurement
+
+import java.io.File
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+import kotlin.test.fail
+
+/**
+ * Keeps the finding registry honest.
+ *
+ * The interesting test is the last one: it reads `docs/findings-registry.md` and fails when the
+ * document and the code disagree. Documentation that drifts from its implementation is worse than
+ * none, because it still looks authoritative — and a finding registry is precisely the artifact
+ * other people build tooling against.
+ */
+class FindingRegistryTest {
+
+    @Test
+    fun codesAreUnique() {
+        val dupes = FindingRegistry.all.groupBy { it.code }.filterValues { it.size > 1 }.keys
+        assertTrue(dupes.isEmpty(), "duplicate finding codes: $dupes")
+    }
+
+    @Test
+    fun everyDeclaredSpecIsInTheAllList() {
+        // Reflection over the object's properties: a spec that is declared but left out of `all`
+        // is invisible to the doc check and to any consumer enumerating the registry.
+        val declared = FindingRegistry::class.java.declaredMethods
+            .filter { it.parameterCount == 0 && it.returnType == FindingSpec::class.java }
+            .mapNotNull { runCatching { it.invoke(FindingRegistry) as FindingSpec }.getOrNull() }
+            .map { it.code }
+            .toSet()
+        val listed = FindingRegistry.all.map { it.code }.toSet()
+        assertEquals(declared, listed, "declared specs and the `all` list disagree")
+    }
+
+    // The prefix decides the category, and the category decides which verdict light the finding
+    // rolls up into. A code whose prefix disagrees with its category silently moves a fault to a
+    // different light — the exact bug that got two codes renamed out of nat.*.
+    @Test
+    fun everyPrefixMatchesItsCategory() {
+        for (spec in FindingRegistry.all) {
+            val fromPrefix = TestType.category(spec.code)
+            assertEquals(
+                fromPrefix, spec.category,
+                "${spec.code} is declared as ${spec.category} but its prefix maps to $fromPrefix",
+            )
+        }
+    }
+
+    @Test
+    fun codesFollowTheNamingConvention() {
+        val shape = Regex("^[a-z0-9]+\\.[a-z0-9_]+$")
+        for (spec in FindingRegistry.all) {
+            assertTrue(shape.matches(spec.code), "malformed code: ${spec.code}")
+            assertTrue(spec.meaning.isNotBlank(), "${spec.code} has no meaning")
+            assertTrue(
+                spec.meaning.trimEnd().endsWith("."),
+                "${spec.code}'s meaning should be a sentence: '${spec.meaning}'",
+            )
+        }
+    }
+
+    // Two near-identical codes are how one fault ends up split across two dashboards. This is a
+    // blunt check — it will not catch every synonym — but it catches the shape that already
+    // happened: the same words in a different order.
+    @Test
+    fun noTwoCodesAreAnagramsOfEachOther() {
+        val normalised = FindingRegistry.all.associate { spec ->
+            spec.code to spec.code.substringAfter('.').split('_').sorted().joinToString("_")
+        }
+        val clashes = normalised.entries.groupBy { it.value }.filterValues { it.size > 1 }
+        if (clashes.isNotEmpty()) {
+            fail("codes differing only in word order: ${clashes.values.map { g -> g.map { it.key } }}")
+        }
+    }
+
+    @Test
+    fun theDocumentAndTheRegistryAgree() {
+        val doc = findDoc() ?: run {
+            println("findings-registry.md not found from ${File(".").absolutePath} — skipping")
+            return
+        }
+        val text = doc.readText()
+
+        // Only table rows count as "documented". Prose may legitimately mention a code that no
+        // longer exists — the rules section explains why two were merged — and treating that as
+        // a registry entry would force the document to forget its own history.
+        val documented = text.lines()
+            .filter { it.trimStart().startsWith("|") }
+            .flatMap { row -> Regex("`([a-z0-9]+\\.[a-z0-9_]+)`").findAll(row).map { it.groupValues[1] } }
+            .toSet()
+        val registered = FindingRegistry.all.map { it.code }.toSet()
+
+        val missingFromDoc = registered - documented
+        val missingFromCode = documented - registered
+        assertTrue(
+            missingFromDoc.isEmpty(),
+            "these codes exist in FindingRegistry but not in docs/findings-registry.md: $missingFromDoc",
+        )
+        assertTrue(
+            missingFromCode.isEmpty(),
+            "docs/findings-registry.md documents codes that no longer exist: $missingFromCode",
+        )
+
+        // And the severities must match, or the document is describing a different system.
+        for (spec in FindingRegistry.all) {
+            val row = text.lines().firstOrNull {
+                it.trimStart().startsWith("|") && it.contains("`${spec.code}`")
+            } ?: continue
+            val severity = spec.severity.name.lowercase()
+            assertTrue(
+                row.contains("| $severity |"),
+                "${spec.code} is ${severity} in code but the doc row says otherwise: $row",
+            )
+        }
+    }
+
+    /** Walks up from the test's working directory to find the repo's docs/ folder. */
+    private fun findDoc(): File? {
+        var dir: File? = File(".").absoluteFile
+        repeat(6) {
+            val candidate = File(dir, "docs/findings-registry.md")
+            if (candidate.isFile) return candidate
+            dir = dir?.parentFile
+        }
+        return null
+    }
+}

echolot-app/core-privacy/src/main/kotlin/app/echo_lot/privacy/Anonymizer.kt

@@ -108,8 +108,14 @@ class Anonymizer(private val level: PrivacyLevel, private val salt: Salt) {
         is JsonObject -> walkObject(v, path)
         is JsonArray -> JsonArray(v.map { walk(key, it, path) })
         is JsonPrimitive ->
-            if (v.isString) transform(Classification.typeOf(key, path), v.content).let(::JsonPrimitive)
+            if (v.isString) {
-            else v
+                // Name first (it is precise), then shape (it is exhaustive). A field nobody
+                // classified must not be a field that leaks.
+                val type = Classification.typeOf(key, path) ?: Classification.inferFromValue(v.content)
+                JsonPrimitive(transform(type, v.content))
+            } else {
+                v
+            }
     }
 
     private fun transform(type: LogicalType?, value: String): String = when (type) {
@@ -147,6 +153,11 @@ class Anonymizer(private val level: PrivacyLevel, private val salt: Salt) {
      * Public addresses keep only their /16 so the network is still locatable at ISP granularity.
      */
     private fun ip4(value: String): String {
+        // A route destination carries a prefix length; pseudonymize the address and put it back,
+        // or "0.0.0.0/0" turns into nonsense and the routing table becomes unreadable.
+        value.substringAfter('/', "").takeIf { it.isNotEmpty() && value.contains('/') }?.let { len ->
+            return ip4(value.substringBefore('/')) + "/" + len
+        }
         val o = value.split(".")
         if (o.size != 4 || o.any { it.toIntOrNull() == null }) return value
         val n = o.map { it.toInt() }
@@ -167,8 +178,36 @@ class Anonymizer(private val level: PrivacyLevel, private val salt: Salt) {
      * is a device fingerprint, especially with EUI-64.
      */
     private fun ip6(value: String): String {
+        // Dotted quads reach here through the family-agnostic field names (addr, gateway, dst);
+        // hand them to the IPv4 path rather than mangling them as if they were v6.
+        if (value.count { it == ':' } < 2) return ip4(value)
+        if (value.contains('/')) {
+            return ip6(value.substringBefore('/')) + "/" + value.substringAfter('/')
+        }
         val v = value.lowercase(Locale.ROOT)
+        // The unspecified address and the default route are not identities; mangling them would
+        // make a routing table unreadable for no privacy gain.
         if (v == "::1" || v == "::" || v.startsWith("fe80:") || v.startsWith("ff")) return v
+
+        // Unique local addresses (fc00::/7) need the *whole* prefix replaced, not the tail.
+        //
+        // They look like the v6 equivalent of RFC1918, and the first instinct is to keep them for
+        // the same reason: private, topological, says nothing about anyone. That reasoning does
+        // not carry over. An RFC1918 prefix is shared by millions of networks and identifies
+        // none of them; a ULA global ID is 40 *random* bits, unique to one network by
+        // construction (RFC 4193). It is a network fingerprint. Passing the leading groups
+        // through - which is what the general path does - leaked 32 of those 40 bits.
+        //
+        // The prefix is pseudonymized as a unit, so two addresses on the same ULA subnet still
+        // land on the same pseudonymous prefix. "These hosts are on one network" survives;
+        // "this is *that* network" does not.
+        if (v.startsWith("fc") || v.startsWith("fd")) {
+            val groups = v.substringBefore('%').split(":")
+            val prefix = pseudo("ula-prefix", groups.take(3).joinToString(":")) { it }
+            val host = pseudo("ula-host", v) { it }
+            return "fd${prefix.substring(0, 2)}:${prefix.substring(2, 6)}:${prefix.substring(6, 10)}" +
+                "::${host.substring(0, 4)}"
+        }
         val groups = v.substringBefore('%').split(":")
         if (groups.size < 3) return v
         val h = pseudo("ip6", value) { it }

echolot-app/core-privacy/src/main/kotlin/app/echo_lot/privacy/Classification.kt

@@ -32,6 +32,16 @@ object Classification {
             "link_local", "ra_source", "prefix",
         ).forEach { put(it, LogicalType.IP6) }
 
+        // Family-agnostic address fields — the names the models actually use (Address.addr,
+        // Route.gateway, Route.dst, DnsConfig.servers). Their absence here was a real leak: the
+        // device's own global IPv6 address went out verbatim at the level whose description
+        // promises addresses are pseudonymized. Typed IP6 because the transform detects the
+        // family from the value, falling through to the IPv4 path for a dotted quad.
+        listOf(
+            "addr", "address", "gateway", "dst", "src", "servers", "server", "resolver",
+            "next_hop", "via", "public_ip", "observed_ip",
+        ).forEach { put(it, LogicalType.IP6) }
+
         listOf("mac", "hw_addr", "gateway_mac", "router_mac", "sender_mac", "peer_mac")
             .forEach { put(it, LogicalType.MAC) }
         listOf("bssid", "ap_mac").forEach { put(it, LogicalType.BSSID) }
@@ -40,6 +50,8 @@ object Classification {
         listOf(
             "fqdn", "hostname", "host", "name", "reverse_dns", "ptr", "domain", "query_name",
             "friendly_name", "server_name", "sni", "cname", "search_domain", "device_name",
+            // Plural and prefixed variants the models actually use.
+            "search_domains", "private_dns_hostname", "domains", "hostnames",
         ).forEach { put(it, LogicalType.FQDN) }
 
         listOf("session_id", "credential", "token", "device_id", "android_id", "serial", "imsi", "iccid")
@@ -78,6 +90,49 @@ object Classification {
         return null
     }
 
+    /**
+     * Last-resort classification from the *value*, when the field name is unrecognised.
+     *
+     * A name table can only protect fields somebody remembered to add, which is the wrong
+     * property for a privacy control: the dangerous field is the one nobody thought of. This
+     * exists because that failed once already — `addresses[].addr` holds the device's own global
+     * IPv6 address, the table had never heard of the name, and it went out verbatim.
+     *
+     * Only addresses and MACs are inferred, because only those have shapes that cannot be
+     * mistaken for something else. Hostnames deliberately are not: `train.udp_updown` is
+     * indistinguishable from a domain by shape, and mangling a test type would corrupt the
+     * document to protect nothing.
+     */
+    fun inferFromValue(value: String): LogicalType? {
+        val v = value.trim()
+        if (v.isEmpty() || v.length > 64) return null
+        if (looksLikeMac(v)) return LogicalType.MAC
+        if (looksLikeIp6(v)) return LogicalType.IP6
+        if (looksLikeIp4(v)) return LogicalType.IP4
+        return null
+    }
+
+    private fun isHex(c: Char) = c in '0'..'9' || c in 'a'..'f' || c in 'A'..'F'
+
+    private fun looksLikeMac(v: String): Boolean {
+        val parts = v.split(':', '-')
+        return parts.size == 6 && parts.all { p -> p.length == 2 && p.all(::isHex) }
+    }
+
+    private fun looksLikeIp4(v: String): Boolean {
+        val parts = v.substringBefore('/').split('.')
+        return parts.size == 4 && parts.all { p ->
+            p.isNotEmpty() && p.length <= 3 && p.all(Char::isDigit) && p.toInt() <= 255
+        }
+    }
+
+    private fun looksLikeIp6(v: String): Boolean {
+        val core = v.substringBefore('/').substringBefore('%')
+        // Two colons minimum, so a time or a MAC fragment does not qualify, and nothing but the
+        // characters an address may contain.
+        return core.count { it == ':' } >= 2 && core.all { it == ':' || isHex(it) }
+    }
+
     fun dropAtBalanced(path: List<String>): Boolean {
         if (path.isNotEmpty() && path.last() in droppedKeys) return true
         return droppedPaths.any { dropped -> dropped.all { path.contains(it) } }

echolot-app/core-privacy/src/test/kotlin/app/echo_lot/privacy/AnonymizerTest.kt

@@ -182,4 +182,47 @@ class AnonymizerTest {
         assertEquals(PrivacyLevel.BALANCED, PrivacyLevel.max(PrivacyLevel.BALANCED, PrivacyLevel.FULL))
         assertEquals(PrivacyLevel.FULL, PrivacyLevel.fromWire("nonsense"))
     }
+
+    // A ULA looks like the v6 RFC1918 and is not. Its global ID is 40 random bits, unique to one
+    // network by construction (RFC 4193), so the prefix IS the identifier - unlike 192.168.x,
+    // which millions of networks share. Passing the leading groups through leaked most of it.
+    @Test
+    fun ulaPrefixesArePseudonymizedWhole() {
+        val doc = json.parseToJsonElement(
+            """{"run":{"id":"r"},"networks":[{"link":{"dns":{"servers":["fda1:3fb1:ff92:6696::2662"]}}}]}"""
+        ).jsonObject
+        val out = flat(anon(PrivacyLevel.BALANCED, doc))
+        assertFalse(out.contains("fda1"), "the ULA global ID survived: $out")
+        assertFalse(out.contains("3fb1"), "part of the ULA global ID survived: $out")
+        assertTrue(out.contains("fd"), "the result should still read as a ULA: $out")
+    }
+
+    // Pseudonymizing the prefix as a unit keeps the one fact that is diagnostically useful:
+    // whether two addresses sit on the same network.
+    @Test
+    fun addressesOnOneUlaSubnetStayRelated() {
+        val doc = json.parseToJsonElement(
+            """{"run":{"id":"r"},"networks":[{"link":{"dns":{"servers":[
+               "fda1:3fb1:ff92:6696::1","fda1:3fb1:ff92:6696::2","fdff:9999:8888:7777::1"]}}}]}"""
+        ).jsonObject
+        val servers = anon(PrivacyLevel.BALANCED, doc)["networks"]!!.jsonArray[0].jsonObject["link"]!!
+            .jsonObject["dns"]!!.jsonObject["servers"]!!.jsonArray.map { it.jsonPrimitive.content }
+        val prefixOf = { s: String -> s.substringBeforeLast("::") }
+        assertEquals(prefixOf(servers[0]), prefixOf(servers[1]),
+            "two addresses on one ULA subnet should share a pseudonymous prefix")
+        assertNotEquals(prefixOf(servers[0]), prefixOf(servers[2]),
+            "a different ULA network must not collide with the first")
+    }
+
+    // RFC1918 stays readable, and this is the contrast that justifies it: a shared, meaningless
+    // prefix is topology; a unique random one is identity.
+    @Test
+    fun rfc1918StaysReadableUnlikeUla() {
+        val doc = json.parseToJsonElement(
+            """{"run":{"id":"r"},"networks":[{"link":{"dns":{"servers":["192.168.1.1","10.13.102.1"]}}}]}"""
+        ).jsonObject
+        val out = flat(anon(PrivacyLevel.BALANCED, doc))
+        assertTrue(out.contains("192.168.1.1"), "RFC1918 should survive: $out")
+        assertTrue(out.contains("10.13.102.1"), "RFC1918 should survive: $out")
+    }
 }

echolot-app/core-privacy/src/test/kotlin/app/echo_lot/privacy/LeakTest.kt

@@ -0,0 +1,113 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package app.echo_lot.privacy
+
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.jsonObject
+import kotlin.test.Test
+import kotlin.test.assertTrue
+
+/**
+ * The blunt instrument: build a document with identifying values in every place one can actually
+ * occur, anonymize it, and assert none of them survive.
+ *
+ * [AnonymizerTest] checks that the fields the classification table knows about are handled
+ * correctly. This checks the other half — the fields it does *not* know about. A per-field test
+ * can only fail for a field someone remembered to write a case for, which is exactly the wrong
+ * property for a privacy check: the dangerous field is the one nobody thought of.
+ *
+ * Concretely, this is written the way it is because the schema's own field names disagree with
+ * the classifier's. `Address.addr` carries an IP and is documented as such in
+ * measurement-schema.md §8, but the classifier keys on names like `ip4` and `gateway_ip4` and had
+ * never heard of `addr`.
+ */
+class LeakTest {
+
+    private val json = Json { prettyPrint = false }
+    private val salt = Salt.perRun(ByteArray(32) { 3 })
+
+    /**
+     * Every string here is something that identifies a person, a household or a device, placed
+     * where the real models actually put it (`core-measurement`'s Network/Link/Address/DnsConfig).
+     */
+    private val secrets = listOf(
+        "Rambossek WLAN",              // ssid
+        "78:9a:18:aa:bb:cc",           // bssid
+        "aa:bb:cc:dd:ee:11",           // gateway mac
+        "2001:1ad0:c4fe:6767::150",    // global v6 address on the interface
+        "2a02:1748:dead:beef::1",      // v6 default gateway
+        "203.0.113.77",                // public v4
+        "nas.rambossek.lan",           // private-dns hostname
+        "rambossek.lan",               // search domain
+        "Anna's Chromecast",           // neighbour name
+        "kitchen table",               // free-text note
+    )
+
+    private fun document(): String = """
+        {
+          "schema": "echolot/measurement",
+          "run": {
+            "id": "run-1", "trigger": "manual", "notes": "${secrets[9]}",
+            "device": {"manufacturer": "OnePlus", "model": "CPH2747"}
+          },
+          "networks": [{
+            "id": "net-1", "transport": "wifi",
+            "link": {
+              "mtu": 1500,
+              "addresses": [
+                {"addr": "${secrets[3]}", "prefix_len": 64, "scope": "global"},
+                {"addr": "192.168.1.44", "prefix_len": 24, "scope": "global"}
+              ],
+              "routes": [
+                {"dst": "::/0", "gateway": "${secrets[4]}", "iface": "wlan0"},
+                {"dst": "0.0.0.0/0", "gateway": "192.168.1.1", "iface": "wlan0"}
+              ],
+              "dns": {
+                "servers": ["${secrets[5]}", "192.168.1.1"],
+                "private_dns_hostname": "${secrets[6]}",
+                "search_domains": ["${secrets[7]}"]
+              }
+            },
+            "wifi": {"ssid": "${secrets[0]}", "bssid": "${secrets[1]}"},
+            "neighbors": [{"name": "${secrets[8]}", "mac": "${secrets[2]}"}]
+          }],
+          "tests": [{"id": "t1", "type": "train.udp_updown", "status": "ok",
+                     "metrics": {"rtt_ms_avg": 12.4}}],
+          "findings": [],
+          "summary": {"verdict": "ok"}
+        }
+    """.trimIndent()
+
+    private fun anonymized(level: PrivacyLevel): String =
+        json.encodeToString(
+            kotlinx.serialization.json.JsonObject.serializer(),
+            Anonymizer(level, salt).anonymize(json.parseToJsonElement(document()).jsonObject),
+        )
+
+    @Test
+    fun nothingIdentifyingSurvivesBalanced() {
+        val out = anonymized(PrivacyLevel.BALANCED)
+        val leaked = secrets.filter { out.contains(it) }
+        assertTrue(
+            leaked.isEmpty(),
+            "these identifying values were uploaded verbatim at BALANCED: $leaked\n\n$out",
+        )
+    }
+
+    @Test
+    fun nothingIdentifyingSurvivesStrict() {
+        val out = anonymized(PrivacyLevel.STRICT)
+        val leaked = secrets.filter { out.contains(it) }
+        assertTrue(leaked.isEmpty(), "leaked at STRICT: $leaked\n\n$out")
+    }
+
+    // Private addresses are kept on purpose — they describe the topology and not the person — so
+    // this pins that the leak test above is not passing by accident of over-redaction.
+    @Test
+    fun privateAddressesAreStillReadable() {
+        val out = anonymized(PrivacyLevel.BALANCED)
+        assertTrue(out.contains("192.168.1.1"), "RFC1918 gateway should survive: $out")
+        assertTrue(out.contains("192.168.1.44"), "RFC1918 interface address should survive: $out")
+    }
+}

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

@@ -109,6 +109,49 @@ class ProbeSession(
         return out
     }
 
+    /**
+     * Sends paced upstream traffic for [durationMs] and reports what was put on the wire.
+     *
+     * Paced rather than flat out, for the same reason the server paces: an unpaced burst measures
+     * the local NIC and the first queue it meets, then collapses into loss that reads as a network
+     * fault. The schedule is absolute rather than sleep-per-packet, which accumulates the
+     * scheduler's error and drifts the achieved rate below target over a multi-second run.
+     *
+     * Nothing comes back — the server counts and stays silent — so the result here is only the
+     * send side. The measurement is the gap between this and the server's tally.
+     */
+    fun sendThroughput(durationMs: Long, kbps: Int, sizeBytes: Int = 1200): Sent {
+        val size = sizeBytes.coerceIn(Wire.HEADER_SIZE + 16, 1472)
+        val payload = ByteArray(size - Wire.HEADER_SIZE)
+        val perPacketNs = (size.toLong() * 8 * 1_000_000 / kbps.coerceAtLeast(1)).coerceAtLeast(1_000)
+
+        val start = System.nanoTime()
+        val deadline = start + durationMs * 1_000_000
+        var next = start
+        var packets = 0
+        var bytes = 0L
+        while (System.nanoTime() < deadline) {
+            val pkt = Wire.build(Wire.TYPE_THROUGHPUT_UP, prefix, ++seq, nowNs(), 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. Stop and report what
+                // actually left, rather than counting the remainder as loss on the network.
+                break
+            }
+            packets++
+            bytes += pkt.size
+            next += perPacketNs
+            val sleepNs = next - System.nanoTime()
+            if (sleepNs > 0) Thread.sleep(sleepNs / 1_000_000, (sleepNs % 1_000_000).toInt())
+        }
+        val elapsedMs = (System.nanoTime() - start) / 1_000_000
+        return Sent(packets, bytes, elapsedMs, if (elapsedMs > 0) (bytes * 8 / elapsedMs).toInt() else 0)
+    }
+
+    /** What one upstream run put on the wire locally. */
+    data class Sent(val packets: Int, val bytes: Long, val durationMs: Long, val kbps: Int)
+
     /** 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

@@ -38,6 +38,16 @@ object Wire {
      */
     const val TYPE_FRAG_DATA: Int = 0x0D
 
+    /** One packet of a sustained-rate downstream run. */
+    const val TYPE_THROUGHPUT_DATA: Int = 0x0E
+
+    /**
+     * One packet of a client-driven upstream run. The server counts it and does not answer:
+     * a reply would double the traffic and drag the return path into a measurement that is
+     * specifically about the outbound one.
+     */
+    const val TYPE_THROUGHPUT_UP: Int = 0x0F
+
     /** The 8-byte on-the-wire prefix = first 16 hex chars of the session id, decoded. */
     fun wirePrefix(sessionId: String): ByteArray {
         require(sessionId.length >= 16) { "session id too short" }

server/cmd/echolot-server/main.go

@@ -111,7 +111,7 @@ func serve(cfg *config.Config) error {
 		TLSConfig: &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS12},
 	}
 
-	caps := []string{"udp-probe", "delayed-echo", "connect-back", "http-echo", "downtrain", "big-send"}
+	caps := []string{"udp-probe", "delayed-echo", "connect-back", "http-echo", "downtrain", "big-send", "throughput"}
 	// Crafted fragments need a raw socket. Advertised only when one can actually be opened —
 	// a capability we cannot deliver turns a missing feature into a failed measurement.
 	rawFrag := dataplane.RawFragSupported()
@@ -167,6 +167,7 @@ func serve(cfg *config.Config) error {
 	if rawFrag {
 		ctl.FragSend = dp.FragSend
 	}
+	ctl.DownThroughput = dp.DownThroughput
 
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer stop()

server/internal/control/control.go

@@ -62,6 +62,8 @@ type Server struct {
 	// FragSend emits one datagram as hand-built IP fragments in a chosen order (may be nil:
 	// needs a raw socket, so it is unavailable to an unprivileged server).
 	FragSend func(sess *session.Session, g *session.Grant, sizeBytes int, mode dataplane.FragMode, fragSize int) (dataplane.FragResult, error)
+	// DownThroughput sends paced traffic toward the client for a bounded time (may be nil).
+	DownThroughput func(sess *session.Session, g *session.Grant, durationMs, kbps, sizeBytes int) (dataplane.ThroughputResult, error)
 	// EgressMTU reports the server's own measured egress path MTU (0 = unknown). With DF set
 	// we cannot emit a datagram larger than this, so requested sizes above it are refused up
 	// front and reported as such — the client must not read that as a downstream path limit.
@@ -221,7 +223,11 @@ func (s *Server) observations(w http.ResponseWriter, r *http.Request) {
 		"udp":          map[string]any{"packets_seen": packetsSeen, "packets": udp},
 		"tcp":          tcp,
 		"connect_back": cb,
-		"dns_canary":   dnsCanary,
+		// The sender's own count, which is what makes the receiver's count mean something.
+		"throughput": sess.ThroughputReports(),
+		// The receiver's count for upstream runs — same idea, other direction.
+		"throughput_up": upstreamJSON(sess),
+		"dns_canary":    dnsCanary,
 		// TODO(spec §6): http echo records
 	})
 }
@@ -246,6 +252,10 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
 		DF         *bool  `json:"df"`
 		Mode       string `json:"mode"`
 		FragBytes  int    `json:"frag_bytes"`
+		Direction  string `json:"direction"`
+		DurationS  int    `json:"duration_s"`
+		Kbps       int    `json:"kbps"`
+		Streams    int    `json:"streams"`
 	}
 	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
 		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "bad body"})
@@ -417,6 +427,66 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
 			"grant": map[string]any{"max_bytes": g.MaxBytes, "max_kbps": g.MaxKbps},
 		})
 
+	case "throughput":
+		if s.DownThroughput == nil {
+			writeJSON(w, http.StatusNotImplemented, map[string]string{"error": "throughput not wired"})
+			return
+		}
+		// Only the downstream direction needs the server to send. Upstream is the client
+		// sending and the server counting, which needs no action at all — so asking for it here
+		// is a client bug worth naming rather than silently doing the other thing.
+		if req.Direction == "up" {
+			// Upstream needs nothing sent from here — the client generates the traffic and the
+			// server counts it. The only thing an action can usefully do is zero the counter so
+			// the run measures itself rather than inheriting an earlier one.
+			sess.ResetUpstream()
+			writeJSON(w, http.StatusAccepted, map[string]any{
+				"action_id": actionID, "direction": "up", "reset": true,
+				"note": "send TYPE_THROUGHPUT_UP packets, then read observations.throughput_up",
+			})
+			return
+		}
+		if req.Direction != "" && req.Direction != "down" {
+			writeJSON(w, http.StatusBadRequest, map[string]string{
+				"error": "direction must be up or down",
+			})
+			return
+		}
+		// Planned once, here, so the response promises exactly what the run will do. A request
+		// that would outlast the server's byte cap comes back with a shorter duration rather
+		// than being truncated halfway.
+		durationMs, kbps := dataplane.ThroughputPlan(
+			clamp(req.DurationS, 1, 30)*1000, clamp(req.Kbps, 100, 200_000))
+		size := clamp(req.SizeBytes, dataMinPacket, 1472)
+		if req.SizeBytes == 0 {
+			size = 1200
+		}
+		g := sess.NewGrant(actionID, 0, kbps, dataplane.ThroughputLimits(durationMs, kbps))
+		if g == nil {
+			writeJSON(w, http.StatusConflict, noDataPlaneYet)
+			return
+		}
+		// Answered before the run so the client can start listening, then reported through the
+		// observations API. Doing it the other way round would have the client miss the first
+		// second of a ten-second test.
+		writeJSON(w, http.StatusAccepted, map[string]any{
+			"action_id": actionID, "direction": "down",
+			"duration_s": durationMs / 1000, "duration_ms": durationMs,
+			"requested_duration_s": clamp(req.DurationS, 1, 30),
+			"kbps":                 kbps, "size_bytes": size,
+			"grant": map[string]any{"max_bytes": g.MaxBytes, "max_kbps": g.MaxKbps},
+		})
+		if f, ok := w.(http.Flusher); ok {
+			f.Flush()
+		}
+		go func() {
+			result, err := s.DownThroughput(sess, g, durationMs, kbps, size)
+			slog.Info("throughput finished", "action", actionID, "packets", result.Packets,
+				"bytes", result.Bytes, "kbps", result.Kbps, "limited_by", result.LimitedBy, "err", err)
+			sess.RecordThroughput(actionID, result.Packets, result.Bytes, result.DurationMs,
+				result.Kbps, result.LimitedBy)
+		}()
+
 	default:
 		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "unknown or unimplemented action"})
 	}
@@ -728,3 +798,13 @@ func (s *Server) EnrollmentLink(token string) string {
 		"&p=" + url.QueryEscape("pin-sha256:"+s.PinB64) +
 		"&t=" + url.QueryEscape(token)
 }
+
+// upstreamJSON renders the upstream tally with the derived figures already computed, so every
+// consumer does not have to repeat (and risk fumbling) the same arithmetic.
+func upstreamJSON(sess *session.Session) map[string]any {
+	u := sess.Upstream()
+	return map[string]any{
+		"packets": u.Packets, "bytes": u.Bytes,
+		"span_ms": u.SpanMs(), "kbps": u.Kbps(),
+	}
+}

server/internal/dataplane/throughput.go

@@ -0,0 +1,187 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package dataplane
+
+import (
+	"encoding/binary"
+	"fmt"
+	"time"
+
+	"echo-lot.app/server/internal/session"
+)
+
+// Sustained-rate sending (spec §5 throughput).
+//
+// This is the most expensive thing the server will do on a client's say-so, so it is also the
+// action where the §3.4 anti-amplification rules matter most. Three bounds apply, and all three
+// are enforced here rather than trusted to the caller:
+//
+//   - the destination is the session's *observed* data-plane source, verified by an HMAC-signed
+//     ECHO that arrived from that address, so this cannot be aimed at a third party;
+//   - the grant carries a byte budget and an average-rate ceiling, and the send stops the moment
+//     either is reached;
+//   - the duration is hard-capped, so a client that vanishes mid-test costs a bounded amount of
+//     traffic rather than an open-ended one.
+//
+// The measurement this produces is honest only if the client is told which limit it hit. A run
+// that saturates the grant ceiling has measured *us*, not the network, and reporting that as
+// throughput would be worse than not measuring at all — see ThroughputResult.LimitedBy.
+
+// ThroughputResult is what the server actually managed to send.
+type ThroughputResult struct {
+	Packets    int   `json:"packets"`
+	Bytes      int64 `json:"bytes"`
+	DurationMs int64 `json:"duration_ms"`
+	Kbps       int   `json:"kbps"`
+	// LimitedBy says what stopped it: "duration" (ran the full time, so the rate is the path's
+	// or ours to give), "budget" (hit the grant's byte ceiling), or "rate" (the pacing ceiling
+	// held it back). Only "duration" makes the number a property of the network.
+	LimitedBy string `json:"limited_by"`
+}
+
+// ThroughputLimits derives a grant sized for one throughput run.
+//
+// The default 8 MiB action budget is deliberately far too small for this — ten seconds at
+// 50 Mbps is 62 MB — so throughput gets its own budget computed from what it asked for, still
+// clamped to a ceiling. Sizing the budget to the request (rather than raising the global default)
+// keeps every *other* action bounded at 8 MiB.
+func ThroughputLimits(durationMs, kbps int) session.GrantLimits {
+	durationMs, kbps = ThroughputPlan(durationMs, kbps)
+	// bytes = kbps * 1000 / 8 * seconds, with a little headroom so the byte budget is not what
+	// stops a run that was meant to be stopped by the clock.
+	budget := int64(kbps) * 1000 / 8 * int64(durationMs) / 1000
+	budget = budget * 11 / 10
+	if budget > maxThroughputBytes {
+		budget = maxThroughputBytes
+	}
+	return session.GrantLimits{
+		MaxBytes: budget,
+		// A little above the pacing target on purpose: the pacer should be what controls the
+		// rate, and the grant should be the safety net. If they are equal, ordinary scheduling
+		// jitter trips the grant and the run is cut short for no real reason.
+		MaxKbps: kbps * 12 / 10,
+		MaxHold: time.Duration(durationMs)*time.Millisecond + 5*time.Second,
+	}
+}
+
+// ThroughputPlan reduces a request to what this server will actually run, and is the single
+// place that decides it.
+//
+// When the byte cap binds before the clock does, the *duration* is shortened rather than the run
+// being cut off partway. Truncating mid-run is not wrong exactly — the rate is still computed
+// over the elapsed time and limited_by says "budget" — but it means promising a client thirty
+// seconds and giving it twenty-one. Saying "twenty-one seconds" up front is the same information
+// without the surprise, and it keeps "the clock ended the run" as the normal case, which is the
+// only case where the number is a clean property of the network.
+func ThroughputPlan(durationMs, kbps int) (effectiveMs, effectiveKbps int) {
+	if durationMs <= 0 {
+		durationMs = 10_000
+	}
+	if durationMs > maxThroughputMs {
+		durationMs = maxThroughputMs
+	}
+	if kbps <= 0 || kbps > maxThroughputKbps {
+		kbps = maxThroughputKbps
+	}
+	bytesPerMs := int64(kbps) * 1000 / 8 / 1000
+	if bytesPerMs > 0 {
+		if maxMs := maxThroughputBytes / bytesPerMs; int64(durationMs) > maxMs {
+			durationMs = int(maxMs)
+		}
+	}
+	return durationMs, kbps
+}
+
+const (
+	maxThroughputMs    = 30_000
+	maxThroughputKbps  = 200_000
+	maxThroughputBytes = 256 << 20
+)
+
+// DownThroughput sends paced traffic toward the client for up to durationMs.
+//
+// Pacing is deliberate rather than "send as fast as possible": an unpaced burst measures the
+// server's NIC and the first queue it meets, then collapses into loss that looks like a network
+// fault. Spacing packets at the target rate makes loss mean what a reader will assume it means.
+func (s *Server) DownThroughput(
+	sess *session.Session, g *session.Grant, durationMs, kbps, sizeBytes int,
+) (ThroughputResult, error) {
+	res := ThroughputResult{}
+
+	target := sess.DataSource()
+	if !target.IsValid() {
+		return res, fmt.Errorf("no observed data-plane source")
+	}
+	conn := s.connFor(target, sess.DataLocal())
+	if conn == nil {
+		return res, fmt.Errorf("no data-plane socket matches target family")
+	}
+
+	// Same plan the grant was sized from, so the two cannot disagree.
+	durationMs, kbps = ThroughputPlan(durationMs, kbps)
+	if sizeBytes < HeaderSize+16 {
+		sizeBytes = 1200 // a size that survives every common path unfragmented
+	}
+	if sizeBytes > 1472 {
+		sizeBytes = 1472
+	}
+
+	// Nanoseconds between packets to hit the target rate.
+	perPacketNs := int64(sizeBytes) * 8 * 1_000_000 / int64(kbps)
+	if perPacketNs < 1_000 {
+		perPacketNs = 1_000
+	}
+
+	payload := make([]byte, sizeBytes-HeaderSize)
+	deadline := time.Now().Add(time.Duration(durationMs) * time.Millisecond)
+	start := time.Now()
+	next := start
+
+	var seq uint32
+	for time.Now().Before(deadline) {
+		ok, why := g.TryAllow(sizeBytes)
+		if !ok {
+			if why == session.RefusalRate {
+				// Transient: the bucket is momentarily empty. Wait for the next slot and carry
+				// on. Ending the run here would report a rate measured over a fraction of a
+				// second, which is worse than reporting no rate at all.
+				res.LimitedBy = "rate"
+				time.Sleep(time.Duration(perPacketNs))
+				continue
+			}
+			// Terminal: the budget is spent, or the grant expired.
+			res.LimitedBy = why
+			break
+		}
+		// Reaching here means the run is progressing normally; the clock will end it.
+		res.LimitedBy = "duration"
+		binary.BigEndian.PutUint32(payload[0:4], seq)
+		binary.BigEndian.PutUint64(payload[4:12], uint64(time.Since(s.start).Nanoseconds()))
+		if err := s.sendErr(conn, target, sess, TypeThroughputData, seq, payload); err != nil {
+			// A send error mid-run is a local condition (buffer full, route gone). Stop and
+			// report what got out rather than pretending the rest was lost on the path.
+			res.LimitedBy = "send_error"
+			break
+		}
+		res.Packets++
+		res.Bytes += int64(sizeBytes)
+		seq++
+
+		// Absolute schedule, not sleep-per-packet: sleeping a fixed interval accumulates the
+		// scheduler's error and drifts the achieved rate below the target over a 10-second run.
+		next = next.Add(time.Duration(perPacketNs))
+		if d := time.Until(next); d > 0 {
+			time.Sleep(d)
+		}
+	}
+
+	elapsed := time.Since(start)
+	res.DurationMs = elapsed.Milliseconds()
+	// bits per millisecond is kilobits per second, so no scaling constant is needed - and none
+	// can be got wrong. Guarded because a run that ends inside a millisecond has no rate.
+	if res.DurationMs > 0 {
+		res.Kbps = int(res.Bytes * 8 / res.DurationMs)
+	}
+	return res, nil
+}

server/internal/dataplane/throughput_test.go

@@ -0,0 +1,99 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package dataplane
+
+import (
+	"testing"
+	"time"
+)
+
+// The grant has to be big enough that the *clock* ends a throughput run, not the byte budget. Get
+// this wrong and the test still "works": it stops early, reports a rate computed over a truncated
+// window, and nothing anywhere says the number is meaningless. So the sizing is pinned.
+func TestThroughputBudgetOutlastsTheRequestedRun(t *testing.T) {
+	cases := []struct{ durationMs, kbps int }{
+		{1_000, 1_000},
+		{10_000, 50_000},
+		{10_000, 200_000},
+		{30_000, 100_000},
+	}
+	for _, c := range cases {
+		// Against the *planned* duration, which is what will actually be run: a request the
+		// server shortens is answered with the shorter number, not truncated halfway.
+		planMs, planKbps := ThroughputPlan(c.durationMs, c.kbps)
+		lim := ThroughputLimits(c.durationMs, c.kbps)
+		needed := int64(planKbps) * 1000 / 8 * int64(planMs) / 1000
+		if lim.MaxBytes < needed {
+			t.Errorf("%d ms at %d kbps (planned %d ms) needs %d bytes, budget is %d - the run "+
+				"would stop early and report a rate over a truncated window",
+				c.durationMs, c.kbps, planMs, needed, lim.MaxBytes)
+		}
+	}
+}
+
+// The pacer should control the rate and the grant should be the safety net. If the grant's
+// ceiling equals the pacing target, ordinary scheduling jitter trips it and cuts the run short
+// for no real reason.
+func TestGrantRateCeilingSitsAboveThePacingTarget(t *testing.T) {
+	lim := ThroughputLimits(10_000, 50_000)
+	if lim.MaxKbps <= 50_000 {
+		t.Fatalf("grant ceiling %d kbps is not above the 50000 kbps pacing target", lim.MaxKbps)
+	}
+}
+
+// A client asking for more than the server will do must get the server's number, not its own.
+func TestThroughputRequestsAreClamped(t *testing.T) {
+	lim := ThroughputLimits(10*60*1000, 10_000_000) // ten minutes at 10 Gbps
+	if lim.MaxBytes > maxThroughputBytes {
+		t.Errorf("byte budget %d exceeds the hard cap %d", lim.MaxBytes, maxThroughputBytes)
+	}
+	if lim.MaxKbps > maxThroughputKbps*12/10 {
+		t.Errorf("rate ceiling %d exceeds the hard cap", lim.MaxKbps)
+	}
+	// The hold has to outlast the planned run, or the grant expires mid-send and the run is
+	// reported as rate-limited when it was really time-limited.
+	planMs, _ := ThroughputPlan(10*60*1000, 10_000_000)
+	if lim.MaxHold < time.Duration(planMs)*time.Millisecond {
+		t.Errorf("hold %v is shorter than the planned run of %d ms", lim.MaxHold, planMs)
+	}
+}
+
+// When the byte cap binds before the clock does, the server shortens the run and says so, rather
+// than accepting thirty seconds and delivering twenty-one. Same information, no surprise - and it
+// keeps "the clock ended the run" as the normal case, which is the only case where the resulting
+// rate is a clean property of the network.
+func TestAnOversizedRequestComesBackShorterRatherThanTruncated(t *testing.T) {
+	const kbps = 200_000
+	askedMs := 30_000
+	planMs, planKbps := ThroughputPlan(askedMs, kbps)
+
+	if planKbps != kbps {
+		t.Errorf("rate was reduced to %d; the duration should absorb the cap, not the rate", planKbps)
+	}
+	if planMs >= askedMs {
+		t.Fatalf("plan kept the full %d ms at %d kbps, which exceeds the %d byte cap",
+			askedMs, kbps, maxThroughputBytes)
+	}
+	// And what it does promise must fit.
+	if got := int64(planKbps) * 1000 / 8 * int64(planMs) / 1000; got > maxThroughputBytes {
+		t.Errorf("planned run needs %d bytes, over the %d cap", got, maxThroughputBytes)
+	}
+}
+
+// A short, ordinary request must come back untouched - the clamping only exists for the extremes.
+func TestAnOrdinaryRequestIsNotRewritten(t *testing.T) {
+	planMs, planKbps := ThroughputPlan(10_000, 50_000)
+	if planMs != 10_000 || planKbps != 50_000 {
+		t.Errorf("10 s at 50 Mbps was rewritten to %d ms at %d kbps", planMs, planKbps)
+	}
+}
+
+// Every action other than throughput stays on the small default budget. Throughput needs a big
+// one; raising the global default to suit it would quietly unbound everything else.
+func TestOnlyThroughputGetsTheLargeBudget(t *testing.T) {
+	big := ThroughputLimits(10_000, 50_000)
+	if big.MaxBytes <= 8<<20 {
+		t.Fatalf("throughput budget %d is no larger than the default action budget", big.MaxBytes)
+	}
+}

server/internal/dataplane/udp.go

@@ -37,6 +37,12 @@ const (
 	TypeBigSend       = 0x0C
 	// TypeFragData is delivered only after IP reassembly, so its arrival IS the measurement.
 	TypeFragData = 0x0D
+	// TypeThroughputData is one packet of a sustained-rate downstream run.
+	TypeThroughputData = 0x0E
+	// TypeThroughputUp is one packet of a client-driven upstream run. The server counts it and
+	// deliberately does not answer: a reply would double the traffic and measure the return
+	// path at the same time, which is the one thing this test is trying not to do.
+	TypeThroughputUp = 0x0F
 )
 
 type Server struct {
@@ -150,6 +156,14 @@ func (s *Server) handle(conn *net.UDPConn, raddr netip.AddrPort, pkt []byte, tRx
 	if la, ok := conn.LocalAddr().(*net.UDPAddr); ok {
 		sess.NoteDataLocal(la.AddrPort())
 	}
+
+	// Upstream throughput short-circuits before the observation log. Recording one struct per
+	// packet here would mean tens of thousands of allocations for a single run; the counter is
+	// all anyone needs, since the client holds the send-side record.
+	if typ == TypeThroughputUp {
+		sess.CountUpstream(len(pkt), tRxNs)
+		return
+	}
 	sess.RecordUDP(session.UDPObservation{
 		Seq: seq, TRxNs: tRxNs, TTxNs: time.Since(s.start).Nanoseconds(),
 		Src: raddr.String(), Size: len(pkt), Type: typ,

server/internal/session/grant.go

@@ -74,22 +74,64 @@ func (s *Session) NewGrant(actionID string, wantBytes int64, wantKbps int, lim G
 // enforces the byte ceiling, the expiry, and the average rate (by refusing early sends rather
 // than sleeping, so callers stay in control of pacing).
 func (g *Grant) Allow(n int) bool {
+	ok, _ := g.TryAllow(n)
+	return ok
+}
+
+// Refusal reasons from TryAllow. The distinction is not cosmetic: "too fast just now" is
+// transient and a caller should pace and carry on, while "budget" and "expired" are terminal and
+// a caller that keeps trying is only wasting its own run.
+const (
+	RefusalNone    = ""
+	RefusalBudget  = "budget"
+	RefusalExpired = "expired"
+	RefusalRate    = "rate"
+)
+
+// TryAllow reports whether n more bytes may be sent now, consuming the budget when they may, and
+// says why not when they may not.
+//
+// The rate limit is a token bucket: allowance = burst + rate x elapsed. An earlier version
+// exempted the first 50 ms from the check entirely, meaning to be lenient at startup. The effect
+// was the opposite - a sender could dump an unbounded burst into that window, and the moment the
+// check switched on it compared those bytes against 50 ms worth of allowance and refused
+// everything until real time caught up. A sustained send died about fifty milliseconds in, having
+// looked perfectly fine in every short test. A bucket has no such cliff: it is smooth from t=0.
+func (g *Grant) TryAllow(n int) (bool, string) {
 	g.mu.Lock()
 	defer g.mu.Unlock()
 	if time.Now().After(g.ExpiresAt) {
-		return false
+		return false, RefusalExpired
 	}
 	if g.sentBytes+int64(n) > g.MaxBytes {
-		return false
+		return false, RefusalBudget
 	}
-	// Average-rate check: bytes allowed so far = kbps/8 * elapsed_seconds.
+	// kbps -> bytes/s is kbps*1000/8 = kbps*125.
+	bytesPerSec := float64(g.MaxKbps) * 125
 	elapsed := time.Since(g.started).Seconds()
-	allowed := float64(g.MaxKbps) * 125 * elapsed // kbps -> bytes/s is kbps*1000/8 = kbps*125
+	allowed := burstBytes(bytesPerSec) + bytesPerSec*elapsed
-	if elapsed > 0.05 && float64(g.sentBytes+int64(n)) > allowed {
+	if float64(g.sentBytes+int64(n)) > allowed {
-		return false
+		return false, RefusalRate
 	}
 	g.sentBytes += int64(n)
-	return true
+	return true, RefusalNone
+}
+
+// burstBytes is the bucket's depth: 100 ms of the allowed rate, floored at a single ordinary
+// datagram.
+//
+// The floor exists only so that one packet is never refused outright by a very slow grant — it is
+// deliberately one datagram and not more. A generous floor would undo the rate ceiling at low
+// rates: at 8 kbps a 64 KB burst is sixty-four seconds' worth, which is exactly the instant dump
+// the ceiling is there to prevent. One datagram is 1.5 seconds' worth at that rate and nothing at
+// any realistic one.
+func burstBytes(bytesPerSec float64) float64 {
+	const oneDatagram = 1500
+	b := bytesPerSec * 0.1
+	if b < oneDatagram {
+		b = oneDatagram
+	}
+	return b
 }
 
 // Sent returns how many bytes this grant has consumed.

server/internal/session/grant_test.go

@@ -92,3 +92,67 @@ func TestGrantEnforcesRate(t *testing.T) {
 		t.Fatalf("rate limit let %d bytes through in ~60ms at 8kbps", sent)
 	}
 }
+
+// The bug this pins: the rate check used to exempt the first 50 ms entirely, so a sender could
+// dump an unbounded burst into that window and then be refused for as long as it took real time
+// to catch up. Every short test passed; a sustained send died about fifty milliseconds in. A
+// token bucket has no such cliff, and the property that matters is that a sender pacing *at* the
+// allowed rate is never refused for long.
+func TestSustainedSendAtTheAllowedRateIsNotCutOff(t *testing.T) {
+	s := sessionWithSource(t)
+	const kbps = 8000   // 1 MB/s
+	const packet = 1200 // bytes
+	g := s.NewGrant("a1", 8<<20, kbps, DefaultGrantLimits)
+
+	// Pace at the allowed rate for a short run and count how much got through. A correct
+	// limiter passes essentially all of it; the old one stopped almost immediately.
+	perPacket := time.Duration(float64(packet) / (float64(kbps) * 125) * float64(time.Second))
+	deadline := time.Now().Add(300 * time.Millisecond)
+	sent, refusals := 0, 0
+	for time.Now().Before(deadline) {
+		if ok, why := g.TryAllow(packet); ok {
+			sent += packet
+		} else if why == RefusalRate {
+			refusals++
+		} else {
+			t.Fatalf("unexpected terminal refusal %q after %d bytes", why, sent)
+		}
+		time.Sleep(perPacket)
+	}
+
+	// 300 ms at 1 MB/s is ~300 KB. Allow generous slack for scheduler granularity, but a run
+	// that delivered only a few packets means the limiter cut it off.
+	if sent < 100_000 {
+		t.Fatalf("a sender pacing at the allowed rate got only %d bytes through in 300ms "+
+			"(%d rate refusals) — the limiter is cutting off sustained sends", sent, refusals)
+	}
+}
+
+// The other half: a rate refusal must be distinguishable from a spent budget, because one is
+// transient and one is terminal, and a caller that cannot tell them apart either gives up early
+// or spins forever.
+func TestRefusalReasonsAreDistinguishable(t *testing.T) {
+	s := sessionWithSource(t)
+
+	// Budget: tiny ceiling, plenty of rate.
+	g := s.NewGrant("a1", 1000, 100_000, DefaultGrantLimits)
+	for i := 0; i < 20; i++ {
+		g.TryAllow(100)
+	}
+	if ok, why := g.TryAllow(100); ok || why != RefusalBudget {
+		t.Errorf("spent budget reported as ok=%v why=%q, want %q", ok, why, RefusalBudget)
+	}
+
+	// Rate: huge ceiling, minimal rate, so only the bucket can refuse.
+	g2 := s.NewGrant("a2", 1<<20, 8, DefaultGrantLimits)
+	sawRate := false
+	for i := 0; i < 100; i++ {
+		if ok, why := g2.TryAllow(1000); !ok && why == RefusalRate {
+			sawRate = true
+			break
+		}
+	}
+	if !sawRate {
+		t.Error("a sender far above the rate ceiling never got a rate refusal")
+	}
+}

server/internal/session/session.go

@@ -40,6 +40,8 @@ type Session struct {
 	packetsSeen uint64
 	udpObs      []UDPObservation // ring, newest last, cap obsCap
 	connectBack []ConnectBackResult
+	throughput  []ThroughputReport
+	upstream    UpstreamCounter
 }
 
 const obsCap = 4096
@@ -54,6 +56,48 @@ type UDPObservation struct {
 	Type  uint8  `json:"type"`
 }
 
+// ThroughputReport is the server's own account of a sustained send: what it managed to put on
+// the wire, and what stopped it. The client needs this to interpret its own count — the gap
+// between the two IS the loss, and without the sender's number a receiver can only guess.
+type ThroughputReport struct {
+	ActionID   string `json:"action_id"`
+	Packets    int    `json:"packets"`
+	Bytes      int64  `json:"bytes"`
+	DurationMs int64  `json:"duration_ms"`
+	Kbps       int    `json:"kbps"`
+	LimitedBy  string `json:"limited_by"`
+}
+
+// UpstreamCounter is the server's tally of a client-driven throughput run.
+//
+// Deliberately a counter and not a list. A five-second upstream run at 20 Mbps is around ten
+// thousand packets; one observation struct each would turn a measurement into an allocation
+// storm on a shared server, and nothing downstream needs the per-packet detail - the client
+// already has its own send record. The gap between the two counts IS the loss.
+type UpstreamCounter struct {
+	Packets   int   `json:"packets"`
+	Bytes     int64 `json:"bytes"`
+	FirstRxNs int64 `json:"first_rx_ns"`
+	LastRxNs  int64 `json:"last_rx_ns"`
+}
+
+// SpanMs is the time between the first and last packet, which is the interval the rate should be
+// computed over - not the client's requested duration, which includes ramp-up and the tail.
+func (u UpstreamCounter) SpanMs() int64 {
+	if u.Packets < 2 || u.LastRxNs <= u.FirstRxNs {
+		return 0
+	}
+	return (u.LastRxNs - u.FirstRxNs) / 1_000_000
+}
+
+// Kbps is bits per millisecond, which is kilobits per second - no scaling constant to get wrong.
+func (u UpstreamCounter) Kbps() int {
+	if ms := u.SpanMs(); ms > 0 {
+		return int(u.Bytes * 8 / ms)
+	}
+	return 0
+}
+
 // ConnectBackResult records one connect-back action outcome.
 type ConnectBackResult struct {
 	ActionID string  `json:"action_id"`
@@ -87,6 +131,57 @@ func (s *Session) Observations() (packetsSeen uint64, udp []UDPObservation, cb [
 		append([]ConnectBackResult(nil), s.connectBack...)
 }
 
+// CountUpstream tallies one client-sent throughput packet.
+//
+// Called on the hot path for every packet of an upstream run, so it does exactly two additions
+// and two comparisons under the lock and allocates nothing.
+func (s *Session) CountUpstream(sizeBytes int, tRxNs int64) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.upstream.Packets == 0 {
+		s.upstream.FirstRxNs = tRxNs
+	}
+	s.upstream.Packets++
+	s.upstream.Bytes += int64(sizeBytes)
+	s.upstream.LastRxNs = tRxNs
+}
+
+// Upstream returns the tally so far.
+func (s *Session) Upstream() UpstreamCounter {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.upstream
+}
+
+// ResetUpstream clears the tally, so a second run in one session measures itself rather than
+// inheriting the first one's packets.
+func (s *Session) ResetUpstream() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.upstream = UpstreamCounter{}
+}
+
+// RecordThroughput stores the server's account of one sustained send.
+//
+// Kept as a per-action summary rather than per-packet records: a ten-second run at 50 Mbps is
+// half a million packets, and holding one struct each would turn a measurement into a memory
+// exhaustion. The client has the per-packet view; the server only needs to say how many it sent.
+func (s *Session) RecordThroughput(actionID string, packets int, bytes, durationMs int64, kbps int, limitedBy string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.throughput = append(s.throughput, ThroughputReport{
+		ActionID: actionID, Packets: packets, Bytes: bytes,
+		DurationMs: durationMs, Kbps: kbps, LimitedBy: limitedBy,
+	})
+}
+
+// ThroughputReports returns the server's account of every sustained send in this session.
+func (s *Session) ThroughputReports() []ThroughputReport {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([]ThroughputReport(nil), s.throughput...)
+}
+
 // DataSource returns the last verified data-plane source (invalid when the
 // session has not sent data-plane traffic yet).
 func (s *Session) DataSource() netip.AddrPort {