throughput: paced downstream rate, with the qualifier that makes it honest

A throughput number 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. So the result always carries limited_by and measures_network, and a finding is raised only when the path is actually implicated. Loss is computed against the *sender's* count, not the requested rate: the server reports what it put on the wire, and the gap 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. The count is stored per action, not per packet — half a million packets of structs would turn a measurement into memory exhaustion. Sending is paced rather than flat out. An unpaced burst measures the server's 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 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 halfway: 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 last behaviour came out of a test that failed honestly: 30 s at 100 Mbps needs 375 MB against a 256 MB cap. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
docs: record frag_send and the current testing state
2026-08-01 14:05:23 +02:00 · 2026-08-01 13:49:43 +02:00 · 2026-08-01 13:45:09 +02:00 · 2026-08-01 13:09:59 +02:00 · 2026-08-01 12:12:28 +02:00 · 2026-08-01 12:11:37 +02:00
23 changed files with 1844 additions and 13 deletions
@@ -162,3 +162,11 @@ First build downloads AGP/Compose/Shizuku from Google Maven + Maven Central.
   are SUPPORTED on both known devices via `Os.recvmsg` + `StructMsghdr` reflection.
 3. Fold the confirmed capabilities + Shizuku dump-format samples back into the production
   `core-probe` / `core-shizuku` modules.
 ## Enrolling a device with a server
 `echolot-app/scripts/enroll-link.sh [note]` mints a §2.1 bootstrap link on fmr over SSH and prints
 it (plus a QR if `qrencode` is installed, plus the `adb shell am start -a …VIEW -d '<uri>'` command
 when a device is attached). The link carries a single-use token — treat it as a secret until spent.
 Never hand-assemble one: the base64 pin needs percent-encoding, and a pin wrong by one character
 fails as an inscrutable TLS error rather than as a bad pin.
@@ -660,3 +660,117 @@ One user-visible bug caught in the process: Go's JSON encoder HTML-escapes `<`,
 default, so the refusal reached the client as `needs \u003e= 0.2.0`. Disabled at the encoder (this
 is an API, not a page), and the client now *parses* the error field instead of pattern-matching it,
 so it survives whatever a future encoder decides to escape.
 ### Enrollment: the server mints the bootstrap link (server-v0.5.3 … v0.5.4, 2026-08-01)
 Until now a device was configured by hand-typing a control URL, a base64 SPKI pin and a
 credential. That is the step that goes wrong, and it goes wrong quietly: a pin off by one
 character does not fail loudly, it just never matches, and surfaces days later as an inscrutable
 TLS error.
 `POST /admin/enroll-tokens` now returns the whole §2.1 bootstrap link alongside the token, because
 the server is the only party holding all three parts at once. The app takes it from a paste or an
 `echolot://enroll` deep link (so a QR scan configures a server in one action) and writes URL, pin
 and credential **together or not at all** — a half-applied server fails later, somewhere else,
 with an error pointing at the wrong thing.
 The control URL comes from `ECHOLOT_PUBLIC_URL` (set on fmr to `https://fmr-1.echo-lot.app:8443`),
 falling back to the first control listen address; a wildcard bind warns rather than emitting a
 link to `0.0.0.0`.
 **The encoding trap, which is the whole reason this is tested across both languages.** The pin is
 base64, so it contains `+`, `/` and `=` — each of which means something else in a query string. An
 unencoded `+` decodes to a space, leaving the pin wrong by exactly one character. Base64 has no
 spaces, so the parser restores them; that cannot damage a correctly-encoded pin and it rescues
 every hand-assembled link. `LiveEnrollmentTest` redeems a link the *server* produced, which is the
 only way to catch a disagreement between the Go assembler and the Kotlin parser — a unit test on
 either side alone cannot see it. It also asserts the token is refused the second time.
 Also fixed a spec divergence found while reading §2.1: the spec names the field
 `device_credential`, the first implementation shipped `credential`. The server now sends both and
 the client prefers the spec's; the alias goes once nothing reads it.
 Two process notes from this round:
 - An edit to the admin handler silently failed to apply and the endpoint kept returning just the
  token. Caught by deploying and *looking at the response*, not by trusting a green build.
 - The live suite is now six tests (`LiveServerTest`, `LiveMeasurement`, `LiveGranted`,
  `LiveUpload`, `LiveCompat`, `LiveEnrollment`), all green against fmr from the PC with no device.
 ### Directional loss: which way is the packet loss? (2026-08-01)
 A round trip can only report that *something* was lost somewhere, which is the least useful form
 of the answer — "3 % loss" sends an engineer looking in both directions at once. The server
 already records every packet it received per sequence number (§6), so the two cases are actually
 distinguishable, and `train.udp_updown` now reports them separately:
 - sent, never seen by the server → **upstream** loss
 - seen by the server, reply never arrived → **downstream** loss
 Findings name the direction and say what is *not* implicated, which is half the value:
 `connectivity.loss_upstream` ("the return path is not implicated: replies came back for everything
 that arrived"), `connectivity.loss_downstream`, `nat.udp_unreachable_upstream`.
 Two things the implementation gets deliberately right:
 - **Downstream loss is measured against what reached the server**, not against what was sent.
  Using "sent" as the denominator counts every upstream loss a second time and overstates the
  return path. Pinned by a test with loss in both directions at once.
 - **Per-direction jitter without synchronised clocks.** Absolute one-way delay would need clock
  sync and we deliberately have none (the two-clock rule). But `server_rx − client_tx` carries a
  constant unknown offset, and differencing successive samples cancels it — so RFC 3393 one-way
  delay variation *is* honestly attributable to a direction even though latency is not. A test
  pins that a 10-second clock offset changes nothing.
 Correlation is by **wire sequence number**, which is not the loop index: the counter is shared
 with every other packet type on the session, so "the nth echo" is not "sequence n". `ProbeSession`
 now exposes `lastSeq`, including for a probe that was lost — a lost packet still has a sequence
 number, and that number is exactly what tells you which way it was lost.
 Live against fmr: 20/20 both ways, and jitter of **0.08 ms upstream vs 0.85 ms downstream** — a
 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.
@@ -24,11 +24,34 @@ echolot://enroll?v=1&u=<control-URL, urlencoded>&p=pin-sha256:<b64 SPKI hash>&t=
 ```
 POST /v1/enroll        Authorization: Bearer <enrollment-token>
-→ 200 { "device_credential": "<random 256-bit, b64url>",
+→ 201 { "device_credential": "<random 256-bit, b64url>",
-        "device_id": "uuid",
+        "device_id": "uuid" }
        "profile": { ... §2.2 ... } }
 ```
 The **server assembles the bootstrap link**, because it is the only party holding all three parts
 at once, and the part an operator gets wrong by hand is the base64 pin — which does not fail
 loudly, it just never matches, and surfaces later as an inscrutable TLS error:
 ```
 POST /admin/enroll-tokens
 → { "token": "…", "expires_in_s": 86400,
    "enroll_uri": "echolot://enroll?v=1&u=…&p=…&t=…" }
 ```
 The control URL in the link comes from `ECHOLOT_PUBLIC_URL`, falling back to the first control
 listen address. A wildcard bind has no single right answer, so it warns rather than guessing.
 Encoding notes that matter in practice:
 - `u`, `p` and `t` are **percent-encoded**. The pin is base64, so it contains `+`, `/` and `=`,
  every one of which means something else in a query string.
 - A `+` that was *not* encoded decodes to a space. Base64 contains no spaces, so a parser SHOULD
  restore them — the alternative is a pin wrong by one character and a failure that points nowhere
  near the cause.
 - The control URL MUST be `https://`. The pin only protects a TLS connection; a cleartext URL
  would hand the token to anyone on the path.
 - **The link is a secret** while it is live: it carries a bearer token, so anyone who sees it
  before the device does can enroll instead.
 Enrollment tokens are single-use with expiry, created in the admin UI, scoped `enroll`. The device credential is a long-lived bearer secret, scoped `run-tests`; it is also the HKDF input for session keys. Revocation = deleting the device in the admin UI.
 ### 2.2 Profile
@@ -0,0 +1,107 @@
 // SPDX-FileCopyrightText: 2026 Echolot contributors
 // SPDX-License-Identifier: GPL-3.0-or-later
 package app.echo_lot.engine
 import kotlinx.serialization.SerialName
 import kotlinx.serialization.Serializable
 /**
 * Splits a round-trip train into its two directions using what the server witnessed.
 *
 * A round trip can only report that *something* was lost somewhere. That is the least useful form
 * of the answer: "3 % loss" sends an engineer looking in both directions at once. The server
 * records every packet it received, per sequence number (probe-protocol.md §6), so the two cases
 * are actually distinguishable:
 *
 *   - sent, never seen by the server  → **upstream** loss
 *   - seen by the server, reply never arrived → **downstream** loss
 *
 * The same records give one-way delay *variation* per direction. Absolute one-way delay would
 * need synchronised clocks and we deliberately have none (measurement-schema.md's two-clock rule),
 * but the variation does not: (server_rx − client_tx) contains an unknown constant clock offset,
 * and differencing successive samples cancels it. So jitter is honestly attributable to a
 * direction even though latency is not.
 */
 object Directional {
    /** One probe as the client saw it. [tRxNs] null means no reply came back. */
    data class Sample(val seq: Int, val tTxNs: Long, val tRxNs: Long?)
    /** One probe as the server saw it: its own receive and transmit stamps, on its own clock. */
    data class ServerSighting(val seq: Int, val tRxNs: Long, val tTxNs: Long)
    fun analyse(sent: List<Sample>, seen: List<ServerSighting>): DirectionalMetrics {
        val byServerSeq = seen.associateBy { it.seq }
        // Only sequences we actually sent count. A server record for a sequence we have no note
        // of is not evidence about this train — it is a bug or a stray, and silently folding it
        // in would produce loss percentages above 100 or below zero.
        val relevant = sent.filter { byServerSeq.containsKey(it.seq) }
        val nSent = sent.size
        val nSeen = relevant.size
        val nReplied = sent.count { it.tRxNs != null }
        // A reply can only exist if the request arrived, so downstream loss is measured against
        // what the server saw, not against what we sent — otherwise upstream loss is counted twice.
        val lostUp = nSent - nSeen
        val lostDown = (nSeen - nReplied).coerceAtLeast(0)
        val upDeltas = relevant.sortedBy { it.seq }
            .map { byServerSeq.getValue(it.seq).tRxNs - it.tTxNs }
        val downDeltas = sent.filter { it.tRxNs != null && byServerSeq.containsKey(it.seq) }
            .sortedBy { it.seq }
            .map { it.tRxNs!! - byServerSeq.getValue(it.seq).tTxNs }
        return DirectionalMetrics(
            sent = nSent,
            seenByServer = nSeen,
            repliesReceived = nReplied,
            lostUpstream = lostUp,
            lostDownstream = lostDown,
            lossUpstreamPct = pct(lostUp, nSent),
            // Denominator is what reached the server: of the packets that got there, how many
            // replies came back.
            lossDownstreamPct = pct(lostDown, nSeen),
            jitterUpstreamMs = jitterMs(upDeltas),
            jitterDownstreamMs = jitterMs(downDeltas),
            /** True when the server saw nothing at all, which is a different fault from loss. */
            noneReachedServer = nSent > 0 && nSeen == 0,
        )
    }
    /**
     * Mean absolute difference between consecutive one-way samples (RFC 3393 IPDV, averaged).
     *
     * Differencing is what makes this legitimate without synchronised clocks: each sample carries
     * the same unknown offset between the two clocks, and the difference cancels it. Fewer than
     * two samples yields null rather than zero — "no jitter" and "not enough data to say" are
     * different claims and only one of them is true here.
     */
    private fun jitterMs(oneWayNs: List<Long>): Double? {
        if (oneWayNs.size < 2) return null
        val deltas = oneWayNs.zipWithNext { a, b -> kotlin.math.abs(b - a) }
        return round2(deltas.average() / 1_000_000.0)
    }
    private fun pct(part: Int, whole: Int): Double =
        if (whole <= 0) 0.0 else round2(part * 100.0 / whole)
    private fun round2(v: Double) = Math.round(v * 100.0) / 100.0
 }
 /** Directional metrics for train.udp_updown; recomputable from the columnar evidence. */
@Serializable
 data class DirectionalMetrics(
    val sent: Int,
    @SerialName("seen_by_server") val seenByServer: Int,
    @SerialName("replies_received") val repliesReceived: Int,
    @SerialName("lost_upstream") val lostUpstream: Int,
    @SerialName("lost_downstream") val lostDownstream: Int,
    @SerialName("loss_upstream_pct") val lossUpstreamPct: Double,
    @SerialName("loss_downstream_pct") val lossDownstreamPct: Double,
    /** One-way delay variation (RFC 3393), per direction. Null when there were too few samples. */
    @SerialName("jitter_upstream_ms") val jitterUpstreamMs: Double? = null,
    @SerialName("jitter_downstream_ms") val jitterDownstreamMs: Double? = null,
    @SerialName("none_reached_server") val noneReachedServer: Boolean = false,
 )
@@ -37,6 +37,120 @@ class DownstreamMeasurement(private val ids: IdSource) {
    /** How long to wait for a granted burst after the server accepts the action. */
    private val collectWindowMs = 4_000L
    /**
     * Shorter, but long enough to cover the first_last mode's deliberate 250 ms hold plus a
     * reassembly. A fragment burst is one datagram: it is here quickly or not at all.
     */
    private val fragWindowMs = 1_500L
    /**
     * Asks the server to send one deliberately-fragmented datagram per ordering, and reports
     * which orderings survive the path.
     *
     * Kernel fragmentation always emits fragments in order, first one first, so an oversized
     * datagram can only answer "do fragments get through at all". The interesting fault is about
     * ordering: only the *first* fragment carries the UDP ports, so a stateful firewall that has
     * not seen it has nothing to match the rest against, and many drop them. That failure is
     * invisible to every in-order test and shows up in the field as "large DNS answers fail here"
     * or "the tunnel breaks when the MTU drops".
     */
    fun fragmentOrdering(
        credential: String,
        sessionId: String,
        control: ControlClient,
        probe: ProbeSession,
        sessionRef: String,
        sizeBytes: Int = 2000,
        fragBytes: Int = 576,
    ): Pair<Test, List<Finding>> {
        val testId = ids.uuid()
        val started = ids.monoNs()
        val delivered = LinkedHashMap<String, Boolean>()
        val fragmentCounts = LinkedHashMap<String, Int>()
        var unsupported = false
        for (mode in FRAG_MODES) {
            val reply = runCatching {
                control.action(
                    credential, sessionId,
                    """{"action":"frag_send","size_bytes":$sizeBytes,"mode":"$mode","frag_bytes":$fragBytes}""",
                )
            }
            if (reply.isFailure) {
                // A server without a raw socket says so; that is a missing capability, not a
                // property of the network, and must not be recorded as a failed delivery.
                unsupported = true
                break
            }
            parseInt(reply.getOrNull(), "fragments")?.let { fragmentCounts[mode] = it }
            // The burst is already on the wire when the action returns (it is sent
            // synchronously), so anything that survived is either here or lost.
            val got = probe.collectGranted(fragWindowMs).any { it.type == Wire.TYPE_FRAG_DATA }
            delivered[mode] = got
        }
        if (unsupported) {
            return Test(
                id = testId, type = TestType.MTU_FRAG_ORDERING, sessionRef = sessionRef, tier = Tier.APP,
                startedMonoNs = started, endedMonoNs = ids.monoNs(),
                status = TestStatus.UNSUPPORTED,
                error = TestError("no_raw_socket", "this server cannot craft fragments"),
            ) to emptyList()
        }
        val metrics = json.encodeToJsonElement(
            FragOrderingMetrics(
                sizeBytes = sizeBytes,
                fragBytes = fragBytes,
                fragmentsPerBurst = fragmentCounts,
                deliveredByMode = delivered,
                inOrderDelivered = delivered[FRAG_IN_ORDER] == true,
                reorderedDelivered = delivered[FRAG_REVERSED] == true,
                delayedFirstDelivered = delivered[FRAG_FIRST_LAST] == true,
            ),
        ) as JsonObject
        val findings = ArrayList<Finding>()
        val inOrder = delivered[FRAG_IN_ORDER] == true
        val reversed = delivered[FRAG_REVERSED] == true
        val firstLast = delivered[FRAG_FIRST_LAST] == true
        if (!inOrder) {
            findings.add(
                finding(
                    "mtu.fragments_blocked", Category.MTU, Severity.MEDIUM, 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 — " +
                        "will fail here rather than slow down.",
                ),
            )
        } else if (!reversed || !firstLast) {
            // The precise and useful finding: fragments work, but only if they arrive tidily.
            val which = buildList {
                if (!reversed) add("out of order")
                if (!firstLast) add("with the first fragment delayed")
            }.joinToString(" or ")
            findings.add(
                finding(
                    "mtu.fragment_reorder_sensitive", Category.MTU, Severity.LOW, 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 " +
                        "carrying the UDP ports) arrives first — typical of a stateful firewall " +
                        "or NAT. It works until the network reorders, then fails intermittently, " +
                        "which is the hardest kind of fault to chase.",
                ),
            )
        }
        return Test(
            id = testId, type = TestType.MTU_FRAG_ORDERING, sessionRef = sessionRef, tier = Tier.APP,
            startedMonoNs = started, endedMonoNs = ids.monoNs(),
            status = if (inOrder) TestStatus.OK else TestStatus.PARTIAL,
            metrics = metrics,
        ) to findings
    }
    /**
     * Runs all three against an already-primed session.
     *
@@ -66,6 +180,16 @@ class DownstreamMeasurement(private val ids: IdSource) {
        tests.add(df.test); tests.add(frag.test); tests.add(train.test)
        // Fragment ordering only makes sense once we know fragments arrive at all; when they do
        // not, the ordering variants would all report "not delivered" and read as three faults
        // instead of one.
        if (frag.largestDelivered != null) {
            val (fragTest, fragFindings) =
                fragmentOrdering(credential, sessionId, control, probe, sessionRef)
            tests.add(fragTest)
            findings.addAll(fragFindings)
        }
        // A downstream MTU below the classic 1500-byte Ethernet payload is worth saying out loud:
        // it is the usual cause of "small requests work, large responses hang".
        val pathMtu = df.largestDelivered
@@ -310,6 +434,11 @@ class DownstreamMeasurement(private val ids: IdSource) {
        /** IPv4 (20) + UDP (8). The v6 case is 48; reported per-family once v6 sessions land. */
        const val IP_UDP_OVERHEAD4 = 28
        const val FRAG_IN_ORDER = "in_order"
        const val FRAG_REVERSED = "reversed"
        const val FRAG_FIRST_LAST = "first_last"
        val FRAG_MODES = listOf(FRAG_IN_ORDER, FRAG_REVERSED, FRAG_FIRST_LAST)
        /** Straddles the usual suspects: 1500 Ethernet, 1492 PPPoE, 1400-ish tunnels. */
        val DEFAULT_SIZES = listOf(600, 1200, 1372, 1400, 1450, 1472, 1500, 2000, 4000)
@@ -330,6 +459,18 @@ data class BigSendMetrics(
    @SerialName("path_mtu_bytes") val pathMtuBytes: Int? = null,
 )
 /** Metrics for mtu.frag_ordering. */
@Serializable
 data class FragOrderingMetrics(
    @SerialName("size_bytes") val sizeBytes: Int,
    @SerialName("frag_bytes") val fragBytes: Int,
    @SerialName("fragments_per_burst") val fragmentsPerBurst: Map<String, Int>,
    @SerialName("delivered_by_mode") val deliveredByMode: Map<String, Boolean>,
    @SerialName("in_order_delivered") val inOrderDelivered: Boolean,
    @SerialName("reordered_delivered") val reorderedDelivered: Boolean,
    @SerialName("delayed_first_delivered") val delayedFirstDelivered: Boolean,
 )
 /** Metrics for train.udp_downstream. */
@Serializable
 data class DownTrainMetrics(
@@ -10,8 +10,13 @@ import app.echo_lot.measurement.*
 import app.echo_lot.protocol.ControlClient
 import app.echo_lot.protocol.ProbeSession
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonArray
 import kotlinx.serialization.json.JsonObject
 import kotlinx.serialization.json.encodeToJsonElement
 import kotlinx.serialization.json.intOrNull
 import kotlinx.serialization.json.jsonObject
 import kotlinx.serialization.json.jsonPrimitive
 import kotlinx.serialization.json.longOrNull
 /**
 * Runs the server-facing measurements against one target and assembles a [MeasurementDocument]:
@@ -71,7 +76,7 @@ class ServerMeasurement(
        // re-primed source is never recorded and every granted send goes to the old, closed port.
        // Session identity lives on the server; the socket must live as long as it does.
        ProbeSession(cfg.credential, session, cfg.udpHost, cfg.udpPort).use { ps ->
-            val (test, findings) = echoTrain(cfg, ps, startMono)
+            val (test, findings) = echoTrain(cfg, ps, startMono, control, session.sessionId)
            tests.add(test)
            allFindings.addAll(findings)
@@ -105,6 +110,7 @@ class ServerMeasurement(
    private fun echoTrain(
        cfg: Config, ps: ProbeSession, startMono: Long,
        control: ControlClient? = null, sessionId: String? = null,
    ): Pair<Test, List<Finding>> {
        val testId = ids.uuid()
        val seqs = ArrayList<Int>()
@@ -114,9 +120,15 @@ class ServerMeasurement(
        val rtts = ArrayList<Double>()
        val observedPorts = LinkedHashSet<Int>()
        // Wire sequence numbers, kept so the server's observations can be correlated packet by
        // packet. They are not 0..n-1: the counter is shared with every other packet type on the
        // session, so "the nth echo" is not "sequence n".
        val wireSeqs = ArrayList<Int>()
        for (i in 0 until cfg.echoCount) {
            val txMono = ids.monoNs() - startMono
            val r = ps.echo(cfg.echoPaddingBytes)
            wireSeqs.add(ps.lastSeq)
            seqs.add(i)
            tTx.add(txMono)
            sizes.add(Wire_HEADER + cfg.echoPaddingBytes)
@@ -129,6 +141,20 @@ class ServerMeasurement(
            }
        }
        // Ask the server what it actually received. This is what turns "3 % loss somewhere" into
        // "3 % loss upstream" - the least useful form of the answer into a usable one.
        val directional: DirectionalMetrics? =
            if (control != null && sessionId != null) {
                runCatching {
                    val samples = wireSeqs.indices.map {
                        Directional.Sample(wireSeqs[it], tTx[it] ?: 0L, tRx[it])
                    }
                    Directional.analyse(samples, serverSightings(control, cfg, sessionId))
                }.getOrNull() // an older server without the endpoint simply yields no split
            } else {
                null
            }
        val sent = cfg.echoCount
        val received = rtts.size
        val lossPct = if (sent == 0) 0.0 else (sent - received) * 100.0 / sent
@@ -138,6 +164,9 @@ class ServerMeasurement(
            epochMonoNs = startMono, seq = seqs, tTxNs = tTx, tRxNs = tRx, sizeBytes = sizes,
        ).toEvidence()
        val directionalJson = directional?.let {
            json.encodeToJsonElement(DirectionalMetrics.serializer(), it) as JsonObject
        }
        val metrics: JsonObject = json.encodeToJsonElement(
            EchoMetrics(
                sent = sent, received = received, lossPct = round1(lossPct),
@@ -147,7 +176,7 @@ class ServerMeasurement(
                observedPorts = observedPorts.toList(),
                natRebindingDetected = natRebinding,
            )
-        ) as JsonObject
+        ).let { base -> JsonObject((base as JsonObject) + (directionalJson ?: JsonObject(emptyMap()))) }
        val status = when {
            received == 0 -> TestStatus.FAILED
@@ -170,6 +199,35 @@ class ServerMeasurement(
                "High UDP loss to the server (${round1(lossPct)}%)",
                "A large fraction of ECHO probes were lost, indicating an unreliable UDP path."))
        }
        // Naming the direction is the entire value of the split, so the findings do.
        directional?.let { d ->
            when {
                d.noneReachedServer && received == 0 -> findings.add(
                    finding("nat.udp_unreachable_upstream", Category.CONNECTIVITY, Severity.HIGH, 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,
                        "${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 " +
                            "arrived."),
                )
            }
            if (d.lossDownstreamPct >= 2.0) {
                findings.add(
                    finding("connectivity.loss_downstream", Category.CONNECTIVITY, Severity.MEDIUM, 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 " +
                            "is fine; the fault is on the return leg."),
                )
            }
        }
        if (natRebinding) {
            findings.add(finding("nat.udp_rebinding", Category.NAT, Severity.MEDIUM, testId,
                "NAT remapped the UDP source port mid-flow",
@@ -178,6 +236,29 @@ class ServerMeasurement(
        return test to findings
    }
    /**
     * The server's per-packet record of this session's echoes (spec section 6). Filtered to
     * ECHO_REQ, because the observation list also holds MTU probes and anything else we sent -
     * counting those as train packets would invent loss that is not there.
     */
    private fun serverSightings(
        control: ControlClient, cfg: Config, sessionId: String,
    ): List<Directional.ServerSighting> {
        val body = control.observations(cfg.credential, sessionId)
        val packets = Json.parseToJsonElement(body).jsonObject["udp"]
            ?.jsonObject?.get("packets") as? JsonArray ?: return emptyList()
        return packets.mapNotNull { el ->
            val o = el as? JsonObject ?: return@mapNotNull null
            val type = o["type"]?.jsonPrimitive?.intOrNull ?: return@mapNotNull null
            if (type != ECHO_REQ_TYPE) return@mapNotNull null
            Directional.ServerSighting(
                seq = o["seq"]?.jsonPrimitive?.intOrNull ?: return@mapNotNull null,
                tRxNs = o["t_rx_ns"]?.jsonPrimitive?.longOrNull ?: return@mapNotNull null,
                tTxNs = o["t_tx_ns"]?.jsonPrimitive?.longOrNull ?: return@mapNotNull null,
            )
        }
    }
    private fun finding(code: String, cat: Category, sev: Severity, testId: String, title: String, desc: String) =
        Finding(
            id = ids.uuid(), code = code, category = cat, severity = sev, confidence = Confidence.HIGH,
@@ -186,6 +267,7 @@ class ServerMeasurement(
    private companion object {
        const val Wire_HEADER = 32
        const val ECHO_REQ_TYPE = 0x01
        fun round1(v: Double) = Math.round(v * 10.0) / 10.0
    }
 }
@@ -0,0 +1,201 @@
 // 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(
                    "perf.throughput_no_delivery", Category.PERFORMANCE, Severity.HIGH, 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(
                    "perf.throughput_below_offered", Category.PERFORMANCE, Severity.LOW, 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
    }
    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() }
    private fun finding(code: String, cat: Category, sev: Severity, testId: String, title: String, desc: String) =
        Finding(
            id = ids.uuid(), code = code, category = cat, severity = sev, 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. */
@Serializable
 data class ThroughputMetrics(
    @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,
 )
@@ -0,0 +1,155 @@
 // SPDX-FileCopyrightText: 2026 Echolot contributors
 // SPDX-License-Identifier: GPL-3.0-or-later
 package app.echo_lot.engine
 import app.echo_lot.engine.Directional.Sample
 import app.echo_lot.engine.Directional.ServerSighting
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertFalse
 import kotlin.test.assertNotNull
 import kotlin.test.assertNull
 import kotlin.test.assertTrue
 /**
 * The arithmetic that turns "3 % loss somewhere" into "3 % loss upstream". Getting a denominator
 * wrong here does not crash anything — it produces a plausible number pointing at the wrong half
 * of the network, which is worse than no number at all. Hence a test per claim.
 */
 class DirectionalTest {
    /** A clean train: every packet sent, seen and answered. Server clock offset by a constant. */
    private fun clean(n: Int, offsetNs: Long = 5_000_000_000L): Pair<List<Sample>, List<ServerSighting>> {
        val sent = (1..n).map { Sample(it, tTxNs = it * 10_000_000L, tRxNs = it * 10_000_000L + 4_000_000L) }
        val seen = (1..n).map {
            ServerSighting(it, tRxNs = offsetNs + it * 10_000_000L + 2_000_000L,
                tTxNs = offsetNs + it * 10_000_000L + 2_100_000L)
        }
        return sent to seen
    }
    @Test
    fun aCleanTrainReportsNoLossInEitherDirection() {
        val (sent, seen) = clean(10)
        val m = Directional.analyse(sent, seen)
        assertEquals(10, m.sent)
        assertEquals(10, m.seenByServer)
        assertEquals(10, m.repliesReceived)
        assertEquals(0.0, m.lossUpstreamPct)
        assertEquals(0.0, m.lossDownstreamPct)
        assertFalse(m.noneReachedServer)
    }
    // The whole point: a packet the server never saw was lost on the way there.
    @Test
    fun packetsTheServerNeverSawAreUpstreamLoss() {
        val (sent, seen) = clean(10)
        val m = Directional.analyse(sent, seen.filter { it.seq !in setOf(3, 7) })
        assertEquals(2, m.lostUpstream)
        assertEquals(0, m.lostDownstream)
        assertEquals(20.0, m.lossUpstreamPct)
        assertEquals(0.0, m.lossDownstreamPct, "a packet that never arrived cannot be lost coming back")
    }
    @Test
    fun repliesThatNeverArrivedAreDownstreamLoss() {
        val (sent, seen) = clean(10)
        val withHoles = sent.map { if (it.seq in setOf(2, 5)) it.copy(tRxNs = null) else it }
        val m = Directional.analyse(withHoles, seen)
        assertEquals(0, m.lostUpstream)
        assertEquals(2, m.lostDownstream)
        assertEquals(20.0, m.lossDownstreamPct)
    }
    // Downstream loss is measured against what actually reached the server. Using "sent" as the
    // denominator would count every upstream loss a second time and overstate the return path.
    @Test
    fun downstreamLossIsRelativeToWhatReachedTheServer() {
        val (sent, seen) = clean(10)
        // 5 lost on the way there; of the 5 that arrived, 1 reply is lost coming back.
        val seenPartial = seen.filter { it.seq > 5 }
        val withHole = sent.map {
            when {
                it.seq <= 5 -> it.copy(tRxNs = null) // never got there, so never came back
                it.seq == 6 -> it.copy(tRxNs = null) // arrived, reply lost
                else -> it
            }
        }
        val m = Directional.analyse(withHole, seenPartial)
        assertEquals(5, m.lostUpstream)
        assertEquals(50.0, m.lossUpstreamPct)
        assertEquals(1, m.lostDownstream)
        assertEquals(20.0, m.lossDownstreamPct, "1 of the 5 that arrived, not 1 of 10")
    }
    @Test
    fun aServerThatSawNothingIsCalledOutSeparately() {
        val (sent, _) = clean(6)
        val m = Directional.analyse(sent.map { it.copy(tRxNs = null) }, emptyList())
        assertTrue(m.noneReachedServer)
        assertEquals(100.0, m.lossUpstreamPct)
        assertEquals(0.0, m.lossDownstreamPct, "with nothing arriving there is no return path to blame")
    }
    // Jitter is legitimate without synchronised clocks because the offset cancels when successive
    // one-way samples are differenced. This pins that: a huge constant offset must not show up.
    @Test
    fun jitterIsUnaffectedByTheClockOffsetBetweenTheTwoMachines() {
        val (sent, near) = clean(10, offsetNs = 0)
        val (_, far) = clean(10, offsetNs = 9_999_999_999L)
        val a = Directional.analyse(sent, near)
        val b = Directional.analyse(sent, far)
        assertEquals(a.jitterUpstreamMs, b.jitterUpstreamMs,
            "a constant clock offset must cancel when consecutive samples are differenced")
        assertEquals(0.0, assertNotNull(a.jitterUpstreamMs), "an evenly spaced train has no jitter")
    }
    @Test
    fun jitterReflectsUnevenArrival() {
        val sent = listOf(
            Sample(1, 0, 10_000_000),
            Sample(2, 10_000_000, 20_000_000),
            Sample(3, 20_000_000, 30_000_000),
        )
        // Server receive times drift: +2ms, +7ms, +3ms relative to send.
        val seen = listOf(
            ServerSighting(1, 2_000_000, 2_100_000),
            ServerSighting(2, 17_000_000, 17_100_000),
            ServerSighting(3, 23_000_000, 23_100_000),
        )
        val m = Directional.analyse(sent, seen)
        // one-way samples: 2ms, 7ms, 3ms → |7-2| and |3-7| → mean 4.5ms
        assertEquals(4.5, assertNotNull(m.jitterUpstreamMs))
    }
    // "No jitter" and "not enough data to say" are different claims, and only one is true here.
    @Test
    fun tooFewSamplesReportsNoJitterRatherThanZero() {
        val m = Directional.analyse(
            listOf(Sample(1, 0, 10_000_000)),
            listOf(ServerSighting(1, 2_000_000, 2_100_000)),
        )
        assertNull(m.jitterUpstreamMs)
        assertNull(m.jitterDownstreamMs)
    }
    // A server record for a sequence we never sent is not evidence about this train; folding it
    // in would yield loss percentages outside 0–100.
    @Test
    fun strayServerRecordsAreIgnored() {
        val (sent, seen) = clean(5)
        val m = Directional.analyse(sent, seen + ServerSighting(99, 1, 2) + ServerSighting(100, 3, 4))
        assertEquals(5, m.seenByServer)
        assertEquals(0.0, m.lossUpstreamPct)
        assertTrue(m.lossDownstreamPct in 0.0..100.0)
    }
    @Test
    fun anEmptyTrainDoesNotDivideByZero() {
        val m = Directional.analyse(emptyList(), emptyList())
        assertEquals(0.0, m.lossUpstreamPct)
        assertEquals(0.0, m.lossDownstreamPct)
        assertFalse(m.noneReachedServer, "nothing sent is not the same as nothing arriving")
    }
 }
@@ -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
@@ -7,6 +7,7 @@ import app.echo_lot.measurement.*
 import kotlinx.serialization.json.Json
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
 /**
@@ -59,6 +60,18 @@ class LiveMeasurementTest {
        println("metrics: $metrics")
        assertTrue(metrics.toString().contains("rtt_ms_avg"))
        // The directional split is the point of asking the server what it saw: without it a
        // lossy path is reported as "loss" with no direction, which sends an engineer looking
        // in both at once. Correlation is by wire sequence number, so a mismatch here means the
        // two sides disagree about which packet is which.
        val m = metrics.toString()
        assertTrue(m.contains("seen_by_server"), "no directional split in the metrics: $m")
        val seen = Regex(""""seen_by_server":(\d+)""").find(m)?.groupValues?.get(1)?.toInt()
        assertNotNull(seen, "seen_by_server missing")
        assertEquals(20, seen, "the server should have seen every probe on a healthy path")
        assertTrue(m.contains("jitter_upstream_ms"), "no per-direction jitter: $m")
        println("directional: $m")
        assertTrue(doc.summary != null)
        // A healthy local->fmr path should be green (no loss, no rebinding) or yellow.
        println("summary: ${doc.summary}")
@@ -77,6 +77,8 @@ object TestType {
    const val MTU_BLACKHOLE = "mtu.blackhole"
    const val MTU_MSS_OBSERVED = "mtu.mss_observed"
    const val MTU_FRAG_DELIVERY = "mtu.frag_delivery"
    /** Whether fragments survive arriving out of order, not merely whether they survive. */
    const val MTU_FRAG_ORDERING = "mtu.frag_ordering"
    // nat
    const val NAT_STUN_5780 = "nat.stun_5780"
    const val NAT_MAPPING_LIFETIME_UDP = "nat.mapping_lifetime_udp"
@@ -44,13 +44,26 @@ class ProbeSession(
     */
    fun echo(paddingBytes: Int = 40): EchoResult? {
        val t0 = System.nanoTime()
-        val pkt = Wire.build(Wire.TYPE_ECHO_REQ, prefix, ++seq, nowNs(), key, ByteArray(paddingBytes))
+        val wireSeq = ++seq
        val pkt = Wire.build(Wire.TYPE_ECHO_REQ, prefix, wireSeq, nowNs(), key, ByteArray(paddingBytes))
        socket.send(DatagramPacket(pkt, pkt.size, server))
        // A lost probe still has a sequence number, and that number is what lets the server's
        // observations say whether it was lost going out or coming back — so report it either way.
        lastSeq = wireSeq
        val resp = receive(Wire.TYPE_ECHO_RESP) ?: return null
        val rttMs = (System.nanoTime() - t0) / 1_000_000.0
-        return EchoResult(rttMs, Observation.parse(resp.payload))
+        return EchoResult(rttMs, Observation.parse(resp.payload), wireSeq)
    }
    /**
     * The wire sequence number of the most recent [echo], including one that was lost.
     *
     * Exposed because the caller cannot derive it: the counter is shared with every other packet
     * type on this session, so "the nth echo" is not "sequence n".
     */
    var lastSeq: Int = 0
        private set
    /** One MTU probe of [totalSize] bytes (DF is set by the OS on the socket where supported).
     *  Returns the size the server acknowledged receiving, or null if the probe was lost. */
    fun mtuProbe(totalSize: Int): Int? {
@@ -112,5 +125,5 @@ class ProbeSession(
    override fun close() = socket.close()
-    data class EchoResult(val rttMs: Double, val observation: Observation?)
+    data class EchoResult(val rttMs: Double, val observation: Observation?, val seq: Int = 0)
 }
@@ -32,6 +32,15 @@ object Wire {
    const val TYPE_DOWNTRAIN_DATA: Int = 0x06
    const val TYPE_BIG_SEND: Int = 0x0C
    /**
     * A datagram the server deliberately fragmented. Its arrival IS the measurement: it can only
     * be delivered if every fragment survived the path and the local stack reassembled them.
     */
    const val TYPE_FRAG_DATA: Int = 0x0D
    /** One packet of a sustained-rate downstream run. */
    const val TYPE_THROUGHPUT_DATA: Int = 0x0E
    /** 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" }
@@ -0,0 +1,42 @@
 #!/usr/bin/env bash
 # SPDX-FileCopyrightText: 2026 Echolot contributors
 # SPDX-License-Identifier: GPL-3.0-or-later
 #
 # Mints an enrollment link on the probe server and prints it — as text, as a QR code if
 # `qrencode` is around, and as an adb command if a device is attached.
 #
 # The admin listener is localhost-only by design, so this goes over SSH. The link carries a
 # single-use bearer token: treat it like a password until it is redeemed.
 #
 # Usage: echolot-app/scripts/enroll-link.sh [note]
 set -euo pipefail
 SSH_HOST="${ECHOLOT_SSH:-claude-echolot}"
 NOTE="${1:-manual}"
 MINTED=$(ssh -o BatchMode=yes "$SSH_HOST" \
  "curl -s -X POST 'http://127.0.0.1:8444/admin/enroll-tokens?note=$NOTE'")
 URI=$(printf '%s' "$MINTED" | python -c 'import json,sys;print(json.load(sys.stdin).get("enroll_uri",""))')
 if [ -z "$URI" ]; then
  echo "server returned no enroll_uri (needs server-v0.5.4+):" >&2
  echo "$MINTED" >&2
  exit 1
 fi
 echo "$URI"
 echo
 # A QR is the point of the format: scanning beats pasting a 200-character string onto a phone.
 if command -v qrencode >/dev/null 2>&1; then
  qrencode -t ANSIUTF8 "$URI"
 else
  echo "(install qrencode to get a scannable QR here)"
 fi
 # With a device attached, the deep link can be delivered straight to the app — no typing at all.
 if command -v adb >/dev/null 2>&1 && [ -n "$(adb devices | sed -n '2p')" ]; then
  echo
  echo "attached device — deliver it directly with:"
  echo "  adb shell am start -a android.intent.action.VIEW -d '$URI'"
 fi
@@ -111,7 +111,15 @@ 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()
 	if rawFrag {
 		caps = append(caps, "frag-send")
 	} else {
 		slog.Info("frag-send unavailable: no raw socket (needs CAP_NET_RAW)")
 	}
 	if len(config.Addrs(cfg.TCPListen)) > 0 {
 		caps = append(caps, "tcp-echo", "tls-echo")
 	}
@@ -154,6 +162,12 @@ func serve(cfg *config.Config) error {
 		AppRange:         appRange,
 		PublicControlURL: publicControlURL(cfg),
 	}
 	// Left nil when there is no raw socket, so the handler answers "not implemented" with a
 	// reason rather than failing somewhere deeper.
 	if rawFrag {
 		ctl.FragSend = dp.FragSend
 	}
 	ctl.DownThroughput = dp.DownThroughput
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer stop()
@@ -59,6 +59,11 @@ type Server struct {
 	BigSend   func(sess *session.Session, g *session.Grant, sizes []int, df bool) ([]dataplane.BigSendResult, error)
 	// Runs stores uploaded measurement documents (may be nil: uploads unsupported).
 	Runs *runs.Store
 	// 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.
@@ -218,6 +223,8 @@ func (s *Server) observations(w http.ResponseWriter, r *http.Request) {
 		"udp":          map[string]any{"packets_seen": packetsSeen, "packets": udp},
 		"tcp":          tcp,
 		"connect_back": cb,
 		// The sender's own count, which is what makes the receiver's count mean something.
 		"throughput": sess.ThroughputReports(),
 		"dns_canary": dnsCanary,
 		// TODO(spec §6): http echo records
 	})
@@ -241,6 +248,12 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
 		IntervalUs int    `json:"interval_us"`
 		SizesBytes []int  `json:"sizes_bytes"`
 		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"})
@@ -373,6 +386,94 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
 			"grant": map[string]any{"max_bytes": g.MaxBytes, "max_kbps": g.MaxKbps},
 		})
 	case "frag_send":
 		if s.FragSend == nil {
 			writeJSON(w, http.StatusNotImplemented, map[string]string{
 				"error": "frag_send needs a raw socket, which this server does not have",
 			})
 			return
 		}
 		size := clamp(req.SizeBytes, 1600, 8000) // must exceed the path MTU or nothing fragments
 		mode := dataplane.FragMode(req.Mode)
 		switch mode {
 		case dataplane.FragInOrder, dataplane.FragReversed, dataplane.FragFirstLast:
 		default:
 			mode = dataplane.FragInOrder
 		}
 		fragBytes := clamp(req.FragBytes, 8, 1400)
 		g := sess.NewGrant(actionID, int64(size), 0, session.DefaultGrantLimits)
 		if g == nil {
 			writeJSON(w, http.StatusConflict, noDataPlaneYet)
 			return
 		}
 		// Synchronous: the whole burst is a few kB and at most a few hundred milliseconds, and
 		// the caller wants to know it was actually emitted before it starts listening. An
 		// asynchronous send would make "nothing arrived" ambiguous between a path drop and a
 		// send that never happened — the one distinction this test exists to make.
 		result, err := s.FragSend(sess, g, size, mode, fragBytes)
 		slog.Info("frag_send finished", "action", actionID, "mode", mode,
 			"size", size, "fragments", result.Fragments, "err", err)
 		if err != nil {
 			writeJSON(w, http.StatusConflict, map[string]any{
 				"error": err.Error(), "action_id": actionID, "result": result,
 			})
 			return
 		}
 		writeJSON(w, http.StatusAccepted, map[string]any{
 			"action_id": actionID, "mode": string(mode), "size_bytes": size,
 			"frag_bytes": fragBytes, "fragments": result.Fragments,
 			"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 != "" && req.Direction != "down" {
 			writeJSON(w, http.StatusBadRequest, map[string]string{
 				"error": "only direction=down is an action; upstream throughput is measured by sending and reading observations",
 			})
 			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"})
 	}
@@ -0,0 +1,263 @@
 // SPDX-FileCopyrightText: 2026 Echolot contributors
 // SPDX-License-Identifier: GPL-3.0-or-later
 //go:build linux
 package dataplane
 import (
 	"encoding/binary"
 	"fmt"
 	"net/netip"
 	"sync/atomic"
 	"syscall"
 	"time"
 	"echo-lot.app/server/internal/session"
 )
 // Crafted IPv4 fragmentation (spec §5 frag_send).
 //
 // Letting the kernel fragment an oversized datagram — which is what big_send with df=false does —
 // answers one question: do fragments get through at all. It cannot answer the more interesting
 // one, because the kernel always emits fragments in order, first one first.
 //
 // The classic middlebox fault is precisely 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 later fragments against. Plenty of implementations drop them. Others hold them
 // briefly and reassemble; others leak. The difference is invisible to any test that sends
 // fragments in order, and it shows up in the real world as "large DNS answers fail on this
 // network" or "the VPN works until the MTU drops".
 //
 // So this builds the fragments by hand and controls their order and timing. That needs a raw
 // socket (CAP_NET_RAW); when we do not have one the capability is not advertised, rather than
 // advertised and failing later.
 // FragMode is how a fragmented datagram is put on the wire.
 type FragMode string
 const (
 	// FragInOrder is the baseline: first fragment first, as the kernel would. A path that fails
 	// this fails everything, and it tells the others apart from a path that drops all fragments.
 	FragInOrder FragMode = "in_order"
 	// FragReversed sends the last fragment first. This is the one that finds stateful devices
 	// which need the first fragment to build state.
 	FragReversed FragMode = "reversed"
 	// FragFirstLast holds the first fragment back until the others have arrived, which tests
 	// whether the path buffers non-first fragments at all and for how long.
 	FragFirstLast FragMode = "first_last"
 )
 var fragIPID atomic.Uint32
 // RawFragSupported reports whether crafted fragments can actually be sent here.
 //
 // Checked by opening the socket rather than by inspecting capabilities: the question is "will
 // this work", and a permission model has more ways to say no than a capability bit has to say yes
 // (user namespaces, seccomp, LSM). Advertising a capability we cannot deliver would turn a
 // missing feature into a failed measurement.
 func RawFragSupported() bool {
 	fd, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_RAW, syscall.IPPROTO_RAW)
 	if err != nil {
 		return false
 	}
 	_ = syscall.Close(fd)
 	return true
 }
 // FragResult is what happened to one crafted fragment burst.
 type FragResult struct {
 	Mode      FragMode `json:"mode"`
 	SizeBytes int      `json:"size_bytes"`
 	Fragments int      `json:"fragments"`
 	Sent      bool     `json:"sent"`
 	Err       string   `json:"err,omitempty"`
 }
 // FragSend emits one ELT1 packet of sizeBytes as hand-built IPv4 fragments, in the given order.
 //
 // The datagram is assembled whole and then cut up, so what the client reassembles — if it
 // reassembles — is a normal, HMAC-valid packet indistinguishable from any other. That matters:
 // the client must not be able to tell a crafted fragment burst from a kernel one, or it would be
 // measuring our sender rather than the path.
 func (s *Server) FragSend(
 	sess *session.Session, g *session.Grant, sizeBytes int, mode FragMode, fragSize int,
 ) (FragResult, error) {
 	res := FragResult{Mode: mode, SizeBytes: sizeBytes}
 	target := sess.DataSource()
 	if !target.IsValid() {
 		return res, fmt.Errorf("no observed data-plane source")
 	}
 	if !target.Addr().Unmap().Is4() {
 		// IPv6 has no in-network fragmentation: only the source may fragment, via an extension
 		// header. Worth building, but it is a different mechanism and belongs in its own code
 		// path rather than pretending this one covers it.
 		return res, fmt.Errorf("crafted fragmentation is IPv4-only for now")
 	}
 	conn := s.connFor(target, sess.DataLocal())
 	if conn == nil {
 		return res, fmt.Errorf("no data-plane socket matches target family")
 	}
 	local := sess.DataLocal()
 	if !local.IsValid() {
 		return res, fmt.Errorf("session has no recorded local address")
 	}
 	if sizeBytes < HeaderSize+8 {
 		sizeBytes = HeaderSize + 8
 	}
 	if sizeBytes > 8000 {
 		sizeBytes = 8000
 	}
 	if !g.Allow(sizeBytes) {
 		return res, fmt.Errorf("grant exhausted")
 	}
 	// The ELT1 packet, signed exactly as any other, then wrapped in UDP.
 	payload := make([]byte, sizeBytes-HeaderSize)
 	binary.BigEndian.PutUint32(payload[0:4], uint32(sizeBytes))
 	copy(payload[4:], mode)
 	elt := s.buildPacket(sess, TypeFragData, 0, payload)
 	udp := buildUDP(local, target, elt)
 	// Fragment offsets are in 8-byte units, so every fragment except the last must be a multiple
 	// of 8. A payload that is not is not an error — it is a fragment that no host will reassemble.
 	if fragSize <= 0 {
 		fragSize = 576
 	}
 	fragSize = (fragSize / 8) * 8
 	if fragSize < 8 {
 		fragSize = 8
 	}
 	fragments := splitIPv4(local.Addr(), target.Addr(), udp, fragSize, uint16(fragIPID.Add(1)))
 	res.Fragments = len(fragments)
 	fd, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_RAW, syscall.IPPROTO_RAW)
 	if err != nil {
 		res.Err = err.Error()
 		return res, err
 	}
 	defer syscall.Close(fd)
 	if err := syscall.SetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_HDRINCL, 1); err != nil {
 		res.Err = err.Error()
 		return res, err
 	}
 	dst := syscall.SockaddrInet4{}
 	copy(dst.Addr[:], target.Addr().Unmap().AsSlice())
 	send := func(pkt []byte) error { return syscall.Sendto(fd, pkt, 0, &dst) }
 	switch mode {
 	case FragReversed:
 		for i := len(fragments) - 1; i >= 0; i-- {
 			if err := send(fragments[i]); err != nil {
 				res.Err = err.Error()
 				return res, err
 			}
 			time.Sleep(time.Millisecond)
 		}
 	case FragFirstLast:
 		for i := 1; i < len(fragments); i++ {
 			if err := send(fragments[i]); err != nil {
 				res.Err = err.Error()
 				return res, err
 			}
 			time.Sleep(time.Millisecond)
 		}
 		// Long enough to be a real test of whether anything holds fragments, short enough to stay
 		// inside the usual 30-second reassembly timeout by a wide margin.
 		time.Sleep(250 * time.Millisecond)
 		if err := send(fragments[0]); err != nil {
 			res.Err = err.Error()
 			return res, err
 		}
 	default:
 		for _, f := range fragments {
 			if err := send(f); err != nil {
 				res.Err = err.Error()
 				return res, err
 			}
 			time.Sleep(time.Millisecond)
 		}
 	}
 	res.Sent = true
 	return res, nil
 }
 // buildUDP wraps a payload in a UDP header with a computed checksum.
 //
 // The checksum is optional in IPv4 and it would be less code to send zero, but a zero-checksum
 // datagram is dropped by some middleboxes — and that drop would be recorded as a fragmentation
 // failure, which is exactly the wrong conclusion.
 func buildUDP(src, dst netip.AddrPort, payload []byte) []byte {
 	out := make([]byte, 8+len(payload))
 	binary.BigEndian.PutUint16(out[0:2], src.Port())
 	binary.BigEndian.PutUint16(out[2:4], dst.Port())
 	binary.BigEndian.PutUint16(out[4:6], uint16(8+len(payload)))
 	copy(out[8:], payload)
 	// Pseudo-header + UDP header + data, per RFC 768.
 	var sum uint32
 	s4, d4 := src.Addr().Unmap().As4(), dst.Addr().Unmap().As4()
 	for _, b := range [][]byte{s4[:], d4[:]} {
 		sum += uint32(binary.BigEndian.Uint16(b[0:2]))
 		sum += uint32(binary.BigEndian.Uint16(b[2:4]))
 	}
 	sum += uint32(syscall.IPPROTO_UDP)
 	sum += uint32(len(out))
 	for i := 0; i+1 < len(out); i += 2 {
 		sum += uint32(binary.BigEndian.Uint16(out[i : i+2]))
 	}
 	if len(out)%2 == 1 {
 		sum += uint32(out[len(out)-1]) << 8
 	}
 	for sum>>16 != 0 {
 		sum = (sum & 0xFFFF) + (sum >> 16)
 	}
 	ck := ^uint16(sum)
 	if ck == 0 {
 		ck = 0xFFFF // 0 means "no checksum" in IPv4; the all-ones form is the same value
 	}
 	binary.BigEndian.PutUint16(out[6:8], ck)
 	return out
 }
 // splitIPv4 cuts a UDP datagram into IPv4 fragments of at most fragSize payload bytes each.
 //
 // Every fragment carries the same IP ID — that is what marks them as one datagram — and every one
 // but the last sets MF. The kernel fills in the header checksum and total length for us under
 // IP_HDRINCL (raw(7)); the ID it only fills when zero, which is why it is set explicitly here.
 func splitIPv4(src, dst netip.Addr, udp []byte, fragSize int, id uint16) [][]byte {
 	s4, d4 := src.Unmap().As4(), dst.Unmap().As4()
 	var out [][]byte
 	for off := 0; off < len(udp); off += fragSize {
 		end := off + fragSize
 		if end > len(udp) {
 			end = len(udp)
 		}
 		chunk := udp[off:end]
 		more := end < len(udp)
 		hdr := make([]byte, 20, 20+len(chunk))
 		hdr[0] = 0x45 // IPv4, 5 words of header
 		hdr[1] = 0    // DSCP/ECN
 		binary.BigEndian.PutUint16(hdr[2:4], uint16(20+len(chunk)))
 		binary.BigEndian.PutUint16(hdr[4:6], id)
 		flagsOff := uint16(off / 8)
 		if more {
 			flagsOff |= 0x2000 // MF
 		}
 		binary.BigEndian.PutUint16(hdr[6:8], flagsOff)
 		hdr[8] = 64 // TTL
 		hdr[9] = syscall.IPPROTO_UDP
 		// hdr[10:12] checksum left zero: the kernel computes it under IP_HDRINCL.
 		copy(hdr[12:16], s4[:])
 		copy(hdr[16:20], d4[:])
 		out = append(out, append(hdr, chunk...))
 	}
 	return out
 }
@@ -0,0 +1,160 @@
 // SPDX-FileCopyrightText: 2026 Echolot contributors
 // SPDX-License-Identifier: GPL-3.0-or-later
 //go:build linux
 package dataplane
 import (
 	"encoding/binary"
 	"net/netip"
 	"testing"
 )
 // Fragment headers are the kind of thing that is either exactly right or silently useless: a
 // wrong offset unit, a missing MF bit or a bad checksum produces packets that leave the machine
 // and are dropped by the receiver's IP stack without a word. Nothing downstream would notice —
 // the client would simply record "fragments do not get through", which is a wrong answer rather
 // than a missing one. Hence these check the bytes.
 func testAddrs() (netip.AddrPort, netip.AddrPort) {
 	return netip.MustParseAddrPort("192.0.2.1:8442"), netip.MustParseAddrPort("198.51.100.9:41000")
 }
 func TestSplitCoversThePayloadExactlyOnce(t *testing.T) {
 	src, dst := testAddrs()
 	udp := buildUDP(src, dst, make([]byte, 2000))
 	frags := splitIPv4(src.Addr(), dst.Addr(), udp, 576, 0x1234)
 	if len(frags) < 3 {
 		t.Fatalf("expected several fragments for %d bytes, got %d", len(udp), len(frags))
 	}
 	// Reassemble the way a receiver would: place each fragment's payload at its offset.
 	rebuilt := make([]byte, len(udp))
 	covered := make([]bool, len(udp))
 	for _, f := range frags {
 		flagsOff := binary.BigEndian.Uint16(f[6:8])
 		off := int(flagsOff&0x1FFF) * 8
 		body := f[20:]
 		if off+len(body) > len(udp) {
 			t.Fatalf("fragment at offset %d overruns the datagram", off)
 		}
 		for i, b := range body {
 			if covered[off+i] {
 				t.Fatalf("byte %d delivered twice", off+i)
 			}
 			covered[off+i] = true
 			rebuilt[off+i] = b
 		}
 	}
 	for i, c := range covered {
 		if !c {
 			t.Fatalf("byte %d was never sent", i)
 		}
 	}
 	for i := range udp {
 		if rebuilt[i] != udp[i] {
 			t.Fatalf("reassembled byte %d differs", i)
 		}
 	}
 }
 func TestFragmentHeadersAreWellFormed(t *testing.T) {
 	src, dst := testAddrs()
 	udp := buildUDP(src, dst, make([]byte, 3000))
 	frags := splitIPv4(src.Addr(), dst.Addr(), udp, 800, 0xBEEF)
 	for i, f := range frags {
 		if got := f[0]; got != 0x45 {
 			t.Errorf("fragment %d: version/IHL = %#x, want 0x45", i, got)
 		}
 		if got := f[9]; got != 17 {
 			t.Errorf("fragment %d: protocol = %d, want 17 (UDP)", i, got)
 		}
 		if got := binary.BigEndian.Uint16(f[4:6]); got != 0xBEEF {
 			t.Errorf("fragment %d: IP ID = %#x — all fragments of one datagram must share it", i, got)
 		}
 		if got := binary.BigEndian.Uint16(f[2:4]); int(got) != len(f) {
 			t.Errorf("fragment %d: total length = %d, actual %d", i, got, len(f))
 		}
 		flagsOff := binary.BigEndian.Uint16(f[6:8])
 		mf := flagsOff&0x2000 != 0
 		wantMF := i < len(frags)-1
 		if mf != wantMF {
 			t.Errorf("fragment %d: MF = %v, want %v", i, mf, wantMF)
 		}
 	}
 }
 // Offsets are counted in 8-byte units, so every fragment but the last must be a multiple of 8.
 // A 100-byte "fragment size" that silently becomes 100 bytes on the wire produces a datagram no
 // host will ever reassemble.
 func TestNonFinalFragmentsAreEightByteMultiples(t *testing.T) {
 	src, dst := testAddrs()
 	udp := buildUDP(src, dst, make([]byte, 2500))
 	for _, size := range []int{8, 100, 576, 999, 1400} {
 		frags := splitIPv4(src.Addr(), dst.Addr(), udp, (size/8)*8, 1)
 		for i, f := range frags[:len(frags)-1] {
 			if body := len(f) - 20; body%8 != 0 {
 				t.Errorf("size %d: non-final fragment %d carries %d bytes, not a multiple of 8",
 					size, i, body)
 			}
 		}
 	}
 }
 // The UDP checksum is optional in IPv4, and sending zero would be less code — but a
 // zero-checksum datagram is dropped by some middleboxes, and that drop would be recorded as a
 // fragmentation failure. So it must be present and correct.
 func TestUDPChecksumVerifies(t *testing.T) {
 	src, dst := testAddrs()
 	for _, n := range []int{0, 1, 7, 8, 100, 1001} { // odd lengths exercise the tail-byte path
 		udp := buildUDP(src, dst, make([]byte, n))
 		if got := binary.BigEndian.Uint16(udp[6:8]); got == 0 {
 			t.Fatalf("payload %d: checksum is zero, which means 'not computed'", n)
 		}
 		if sum := verifyUDPChecksum(src.Addr(), dst.Addr(), udp); sum != 0xFFFF {
 			t.Errorf("payload %d: checksum does not verify (one's complement sum %#x)", n, sum)
 		}
 		if got := binary.BigEndian.Uint16(udp[4:6]); int(got) != len(udp) {
 			t.Errorf("payload %d: UDP length field %d, actual %d", n, got, len(udp))
 		}
 	}
 }
 func TestUDPPortsComeFromTheSessionAddresses(t *testing.T) {
 	src, dst := testAddrs()
 	udp := buildUDP(src, dst, []byte("x"))
 	if got := binary.BigEndian.Uint16(udp[0:2]); got != src.Port() {
 		t.Errorf("source port = %d, want %d", got, src.Port())
 	}
 	// The destination port must be the client's observed source port, or the datagram arrives
 	// at the machine and is discarded before any socket sees it.
 	if got := binary.BigEndian.Uint16(udp[2:4]); got != dst.Port() {
 		t.Errorf("destination port = %d, want %d", got, dst.Port())
 	}
 }
 // Recomputes the one's complement sum over the pseudo-header and datagram; a correct checksum
 // makes the total 0xFFFF.
 func verifyUDPChecksum(src, dst netip.Addr, udp []byte) uint16 {
 	var sum uint32
 	s4, d4 := src.Unmap().As4(), dst.Unmap().As4()
 	for _, b := range [][]byte{s4[:], d4[:]} {
 		sum += uint32(binary.BigEndian.Uint16(b[0:2]))
 		sum += uint32(binary.BigEndian.Uint16(b[2:4]))
 	}
 	sum += 17
 	sum += uint32(len(udp))
 	for i := 0; i+1 < len(udp); i += 2 {
 		sum += uint32(binary.BigEndian.Uint16(udp[i : i+2]))
 	}
 	if len(udp)%2 == 1 {
 		sum += uint32(udp[len(udp)-1]) << 8
 	}
 	for sum>>16 != 0 {
 		sum = (sum & 0xFFFF) + (sum >> 16)
 	}
 	return uint16(sum)
 }
@@ -0,0 +1,41 @@
 // SPDX-FileCopyrightText: 2026 Echolot contributors
 // SPDX-License-Identifier: GPL-3.0-or-later
 //go:build !linux
 package dataplane
 import (
 	"fmt"
 	"echo-lot.app/server/internal/session"
 )
 // Crafting IP fragments needs a raw socket and Linux's IP_HDRINCL semantics. Off Linux the
 // capability is simply not advertised, so a client never asks for it — better than answering
 // with a measurement we cannot actually make.
 type FragMode string
 const (
 	FragInOrder   FragMode = "in_order"
 	FragReversed  FragMode = "reversed"
 	FragFirstLast FragMode = "first_last"
 )
 type FragResult struct {
 	Mode      FragMode `json:"mode"`
 	SizeBytes int      `json:"size_bytes"`
 	Fragments int      `json:"fragments"`
 	Sent      bool     `json:"sent"`
 	Err       string   `json:"err,omitempty"`
 }
 func RawFragSupported() bool { return false }
 func (s *Server) FragSend(
 	sess *session.Session, g *session.Grant, sizeBytes int, mode FragMode, fragSize int,
 ) (FragResult, error) {
 	return FragResult{Mode: mode, SizeBytes: sizeBytes},
 		fmt.Errorf("crafted fragmentation is only implemented on Linux")
 }
@@ -0,0 +1,182 @@
 // 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
 	res.LimitedBy = "duration"
 	var seq uint32
 	for time.Now().Before(deadline) {
 		if !g.Allow(sizeBytes) {
 			// Distinguishing these two matters: a run stopped by the byte budget has not been
 			// running long enough for its rate to mean anything.
 			if g.Sent() >= g.MaxBytes {
 				res.LimitedBy = "budget"
 			} else {
 				res.LimitedBy = "rate"
 			}
 			break
 		}
 		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
 }
@@ -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)
 	}
 }
@@ -35,6 +35,10 @@ const (
 	// Server->client under an asymmetric grant (spec §3.4/§5).
 	TypeDownTrainData = 0x06
 	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
 )
 type Server struct {
@@ -232,6 +236,17 @@ func (s *Server) send(conn *net.UDPConn, raddr netip.AddrPort, sess *session.Ses
 // EMSGSIZE means our own egress MTU refused the datagram, which is a different fact from the
 // client not receiving it.
 func (s *Server) sendErr(conn *net.UDPConn, raddr netip.AddrPort, sess *session.Session, typ byte, seq uint32, payload []byte) error {
 	pkt := s.buildPacket(sess, typ, seq, payload)
 	_, err := conn.WriteToUDPAddrPort(pkt, raddr)
 	return err
 }
 // buildPacket assembles and signs an ELT1 packet without sending it.
 //
 // Split out for the crafted-fragment path, which needs the bytes so it can cut them up itself.
 // What arrives after reassembly must be indistinguishable from an ordinary packet, or the client
 // would be measuring our sender rather than the path — so it goes through exactly this function.
 func (s *Server) buildPacket(sess *session.Session, typ byte, seq uint32, payload []byte) []byte {
 	pkt := make([]byte, HeaderSize+len(payload))
 	copy(pkt[0:4], Magic)
 	pkt[4] = typ
@@ -247,8 +262,7 @@ func (s *Server) sendErr(conn *net.UDPConn, raddr netip.AddrPort, sess *session.
 	mac.Write(pkt[0:28])
 	mac.Write(payload)
 	copy(pkt[28:32], mac.Sum(nil)[:4])
-	_, err := conn.WriteToUDPAddrPort(pkt, raddr)
+	return pkt
 	return err
 }
 func hexByte(hi, lo byte) byte {
@@ -40,6 +40,7 @@ type Session struct {
 	packetsSeen uint64
 	udpObs      []UDPObservation // ring, newest last, cap obsCap
 	connectBack []ConnectBackResult
 	throughput  []ThroughputReport
 }
 const obsCap = 4096
@@ -54,6 +55,18 @@ 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"`
 }
 // ConnectBackResult records one connect-back action outcome.
 type ConnectBackResult struct {
 	ActionID string  `json:"action_id"`
@@ -87,6 +100,27 @@ func (s *Session) Observations() (packetsSeen uint64, udp []UDPObservation, cb [
 		append([]ConnectBackResult(nil), s.connectBack...)
 }
 // 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 {
Author	SHA1	Message	Date
mrambossekandClaude Fable 5	3333788d9e	throughput: paced downstream rate, with the qualifier that makes it honest server-release / image (push) Successful in 15s Details server-test / test (push) Successful in 32s Details server-release / release (push) Successful in 32s Details A throughput number 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. So the result always carries limited_by and measures_network, and a finding is raised only when the path is actually implicated. Loss is computed against the sender's count, not the requested rate: the server reports what it put on the wire, and the gap 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. The count is stored per action, not per packet — half a million packets of structs would turn a measurement into memory exhaustion. Sending is paced rather than flat out. An unpaced burst measures the server's 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 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 halfway: 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 last behaviour came out of a test that failed honestly: 30 s at 100 Mbps needs 375 MB against a 256 MB cap. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>	2026-08-01 14:05:23 +02:00
mrambossekandClaude Fable 5	35744c609e	docs: record frag_send and the current testing state Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>	2026-08-01 13:49:43 +02:00
mrambossekandClaude Fable 5	a7dccf7da2	frag_send: crafted IP fragments, so ordering can be tested and not just delivery server-release / image (push) Successful in 15s Details server-test / test (push) Successful in 32s Details server-release / release (push) Successful in 32s Details Letting the kernel fragment an oversized datagram answers one question — do fragments get through. It cannot answer the more interesting one, because the kernel always emits them 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 drop them. That is invisible to any in-order test and shows up in the field as "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 now builds the fragments itself (raw socket, IP_HDRINCL) and controls their order: in_order as a baseline, reversed, and first-fragment-last. The datagram is assembled and signed whole before being cut up, so what the client reassembles is indistinguishable from an ordinary packet — otherwise it would be measuring our sender rather than the path. Two details that would silently produce wrong answers: - The UDP checksum is computed rather than left zero. A zero-checksum datagram is 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 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, since a permission model has more ways to say no than a capability bit has to say yes. Fragment header arithmetic is unit-tested (reassembly coverage, MF flags, shared IP ID, 8-byte offsets, checksum verification), cross-compiled and run on Linux since the code is build-tagged. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>	2026-08-01 13:45:09 +02:00
mrambossekandClaude Fable 5	4ffa6e4ae2	engine: split packet loss by direction using the server's observations "3 % loss" sends an engineer looking in both directions at once. The server records every packet it received per sequence number, so the two cases are distinguishable: sent-but-never-seen is upstream loss, seen-but-no-reply is downstream. The findings say which, and say what is not implicated. Downstream loss is measured against what reached the server, not against what was sent — the other denominator counts every upstream loss twice and overstates the return path. Per-direction jitter comes out of the same records without needing synchronised clocks: (server_rx - client_tx) carries a constant unknown offset, and differencing successive samples cancels it, so RFC 3393 variation is honestly attributable to a direction even though absolute latency is not. Correlation is by wire sequence number, not loop index — the counter is shared with every packet type on the session. ProbeSession exposes it even for a lost probe, since that is precisely the packet whose direction is in question. Live against fmr: 0.08 ms upstream jitter vs 0.85 ms downstream, an asymmetry a round-trip test cannot see. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>	2026-08-01 13:09:59 +02:00
mrambossekandClaude Fable 5	3e7e3b8d33	scripts: one command to mint an enrollment link, QR included Scanning beats pasting a 200-character string onto a phone, and with a device attached the deep link can be delivered by adb with no typing at all. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>	2026-08-01 12:12:28 +02:00
mrambossekandClaude Fable 5	199807a8c9	docs: enrollment link encoding rules in the spec, session log in build-status Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>	2026-08-01 12:11:37 +02:00