grant: replace the rate check with a token bucket

The live throughput test found it: a 3-second run delivered 104 packets and
stopped after 50 milliseconds.

The rate check exempted the first 50 ms entirely, meaning to be lenient at
startup. The effect was the opposite. A sender could dump an unbounded burst
into that free window, and the instant the check switched on it compared those
bytes against 50 ms worth of allowance and refused everything until real time
caught up. Every short test passed — downtrain sends 50 packets, big_send seven
— and every sustained send died about fifty milliseconds in.

A token bucket (allowance = burst + rate x elapsed) has no such cliff; it is
smooth from t=0. The burst is 100 ms of the allowed rate, floored at one
ordinary datagram so a single packet is never refused outright. The floor is
deliberately one datagram: at 8 kbps a 64 KB floor would be sixty-four seconds'
worth, which is precisely the instant dump the ceiling exists to prevent. The
existing rate test caught that when I first tried it, and it was right.

Second half of the same bug: callers treated any refusal as terminal. TryAllow
now says why, so a sender can pace through a transient "too fast just now" and
still stop dead on a spent budget or an expired grant.

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

docs/build-status.md

@@ -728,3 +728,49 @@ 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.

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

@@ -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,
+)

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

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

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

@@ -0,0 +1,68 @@
+// SPDX-FileCopyrightText: 2026 Echolot contributors
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package app.echo_lot.engine
+
+import app.echo_lot.measurement.TestStatus
+import app.echo_lot.protocol.ControlClient
+import app.echo_lot.protocol.ProbeSession
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
+
+/**
+ * Downstream throughput against a LIVE server. Self-skips without ECHOLOT_LIVE_*.
+ *
+ * The assertions are about *honesty* rather than speed: a rate is only a measurement if the run
+ * was ended by the clock and the sender's own count backs it up. A test that just asserted "some
+ * Mbps arrived" would pass equally well against a broken implementation.
+ */
+class LiveThroughputTest {
+
+    private val url = System.getenv("ECHOLOT_LIVE_URL")
+    private val pin = System.getenv("ECHOLOT_LIVE_PIN")
+    private val cred = System.getenv("ECHOLOT_LIVE_CRED")
+    private val udp = System.getenv("ECHOLOT_LIVE_UDP")
+    private val target = System.getenv("ECHOLOT_LIVE_TARGET") ?: "fmr"
+
+    @Test
+    fun measuresDownstreamRateAndSaysWhatLimitedIt() {
+        if (url == null || pin == null || cred == null || udp == null) {
+            println("LiveThroughputTest skipped (no ECHOLOT_LIVE_* env)"); return
+        }
+        val control = ControlClient(url, setOf(pin), "0.2.0")
+        val session = control.createSession(cred, target)
+        val (host, port) = udp.split(":").let { it[0] to it[1].toInt() }
+
+        val (test, findings) = ProbeSession(cred, session, host, port).use { ps ->
+            ps.echo() // prime: the grant binds to the observed source
+            ThroughputMeasurement(SystemIdSource()).run(
+                cred, session.sessionId, control, ps, sessionRef = "sess-1",
+                durationS = 3, kbps = 20_000,
+            )
+        }
+        control.deleteSession(cred, session.sessionId)
+
+        val m = assertNotNull(test.metrics).toString()
+        println("throughput: ${test.status} $m")
+        for (f in findings) println("finding ${f.code} [${f.severity}] ${f.title}")
+
+        assertEquals(TestStatus.OK, test.status, "no throughput traffic arrived: $m")
+
+        // The sender's own count must be present — without it, loss cannot be attributed and the
+        // number is not a measurement.
+        assertTrue(m.contains("sender_packets"), "no sender report to compare against: $m")
+        assertTrue(m.contains("limited_by"), "the result must say what ended the run: $m")
+
+        val received = Regex(""""received_kbps":(\d+)""").find(m)?.groupValues?.get(1)?.toInt()
+        assertNotNull(received)
+        assertTrue(received > 0, "measured 0 kbps: $m")
+        println("received ${received / 1000} Mbit/s")
+
+        // A run this short and this far below the ceiling should end on the clock. Anything else
+        // means the grant was the constraint, and then the rate says nothing about the path.
+        assertTrue(m.contains(""""limited_by":"duration""""),
+            "the run did not end on the clock, so the rate measures the server, not the path: $m")
+    }
+}

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

@@ -38,6 +38,9 @@ object Wire {
      */
     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" }

server/cmd/echolot-server/main.go

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

server/internal/control/control.go

@@ -62,6 +62,8 @@ type Server struct {
 	// FragSend emits one datagram as hand-built IP fragments in a chosen order (may be nil:
 	// needs a raw socket, so it is unavailable to an unprivileged server).
 	FragSend func(sess *session.Session, g *session.Grant, sizeBytes int, mode dataplane.FragMode, fragSize int) (dataplane.FragResult, error)
+	// DownThroughput sends paced traffic toward the client for a bounded time (may be nil).
+	DownThroughput func(sess *session.Session, g *session.Grant, durationMs, kbps, sizeBytes int) (dataplane.ThroughputResult, error)
 	// EgressMTU reports the server's own measured egress path MTU (0 = unknown). With DF set
 	// we cannot emit a datagram larger than this, so requested sizes above it are refused up
 	// front and reported as such — the client must not read that as a downstream path limit.
@@ -221,7 +223,9 @@ func (s *Server) observations(w http.ResponseWriter, r *http.Request) {
 		"udp":          map[string]any{"packets_seen": packetsSeen, "packets": udp},
 		"tcp":          tcp,
 		"connect_back": cb,
-		"dns_canary":   dnsCanary,
+		// The sender's own count, which is what makes the receiver's count mean something.
+		"throughput": sess.ThroughputReports(),
+		"dns_canary": dnsCanary,
 		// TODO(spec §6): http echo records
 	})
 }
@@ -246,6 +250,10 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
 		DF         *bool  `json:"df"`
 		Mode       string `json:"mode"`
 		FragBytes  int    `json:"frag_bytes"`
+		Direction  string `json:"direction"`
+		DurationS  int    `json:"duration_s"`
+		Kbps       int    `json:"kbps"`
+		Streams    int    `json:"streams"`
 	}
 	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
 		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "bad body"})
@@ -417,6 +425,55 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
 			"grant": map[string]any{"max_bytes": g.MaxBytes, "max_kbps": g.MaxKbps},
 		})
 
+	case "throughput":
+		if s.DownThroughput == nil {
+			writeJSON(w, http.StatusNotImplemented, map[string]string{"error": "throughput not wired"})
+			return
+		}
+		// Only the downstream direction needs the server to send. Upstream is the client
+		// sending and the server counting, which needs no action at all — so asking for it here
+		// is a client bug worth naming rather than silently doing the other thing.
+		if req.Direction != "" && 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"})
 	}

server/internal/dataplane/throughput.go

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

server/internal/dataplane/throughput_test.go

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

server/internal/dataplane/udp.go

@@ -37,6 +37,8 @@ const (
 	TypeBigSend       = 0x0C
 	// TypeFragData is delivered only after IP reassembly, so its arrival IS the measurement.
 	TypeFragData = 0x0D
+	// TypeThroughputData is one packet of a sustained-rate downstream run.
+	TypeThroughputData = 0x0E
 )
 
 type Server struct {

server/internal/session/grant.go

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

server/internal/session/grant_test.go

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

server/internal/session/session.go

@@ -40,6 +40,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 {