server: §3.4 asymmetric grants + downtrain and big_send actions

The grant is the keystone that makes server->client sends safe: created
only by an authenticated control-plane action, bound at creation to the
session's OBSERVED data-plane source (so it can never be aimed at a third
party), and bounded by bytes, average rate and expiry. Sends stop the
moment the budget runs out, so a buggy action cannot become a flood.

Two granted actions on top of it:
- downtrain: N packets at a given size/interval toward the client, with
  seq + send-timestamp in the payload — downstream loss/reorder/jitter,
  which an upstream-only train cannot measure.
- big_send: one datagram per requested size, echoing the intended size in
  the payload — downstream MTU / black-hole evidence the client cannot
  produce for itself (only the far end can emit a large packet toward it).

Tests cover the security properties: no grant without a verified
destination, client requests clamped to server limits, byte budget stops
sending exactly, expiry refuses, and the rate ceiling throttles a burst.
Capabilities gain downtrain + big-send.

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

server/internal/control/control.go

@@ -49,6 +49,9 @@ type Server struct {
 	TCPRecent func(ip string) any
 	// DelayedEcho schedules/sends a DELAYED_ECHO for a session (may be nil).
 	DelayedEcho func(sess *session.Session, actionID string) error
+	// Granted server->client sends (spec §5). Both consume an asymmetric grant.
+	DownTrain func(sess *session.Session, g *session.Grant, count, sizeBytes, intervalUs int) (int, error)
+	BigSend   func(sess *session.Session, g *session.Grant, sizes []int) ([]int, error)
 	// CanaryQueries returns logged canary lookups for a session prefix (may be nil).
 	CanaryQueries func(sessionPrefix string) any
 	// CanaryZone is surfaced in the profile so the app knows what to query.
@@ -70,8 +73,7 @@ func (s *Server) Handler() http.Handler {
 	mux.HandleFunc("POST /v1/sessions/{id}/actions", s.actions)
 	mux.HandleFunc("POST /v1/echo", s.httpEcho)
 	mux.HandleFunc("GET /v1/tls-reference", s.tlsReference)
-	// TODO(spec §4): TLS-echo/JA4 (tls-echo capability, needs ClientHello capture)
-	// TODO(spec §5): downtrain, big_send, frag_send, throughput
+	// TODO(spec §5): frag_send, throughput (both build on the same grant machinery)
 	return mux
 }
 
@@ -146,10 +148,14 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	var req struct {
-		Action   string `json:"action"`
-		DelayS   int    `json:"delay_s"`
-		Protocol string `json:"protocol"`
-		Port     int    `json:"port"`
+		Action     string `json:"action"`
+		DelayS     int    `json:"delay_s"`
+		Protocol   string `json:"protocol"`
+		Port       int    `json:"port"`
+		Count      int    `json:"count"`
+		SizeBytes  int    `json:"size_bytes"`
+		IntervalUs int    `json:"interval_us"`
+		SizesBytes []int  `json:"sizes_bytes"`
 	}
 	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
 		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "bad body"})
@@ -200,11 +206,81 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
 			sess.RecordConnectBack(res)
 		}()
 		writeJSON(w, http.StatusAccepted, map[string]any{"action_id": actionID, "target": target})
+	case "downtrain":
+		if s.DownTrain == nil {
+			writeJSON(w, http.StatusNotImplemented, map[string]string{"error": "downtrain not wired"})
+			return
+		}
+		// A downstream train sends far more than it receives, so it needs a grant (§3.4).
+		count := clamp(req.Count, 1, 5000)
+		size := clamp(req.SizeBytes, dataMinPacket, 1500)
+		interval := clamp(req.IntervalUs, 0, 1_000_000)
+		g := sess.NewGrant(actionID, int64(count*size), 0, session.DefaultGrantLimits)
+		if g == nil {
+			writeJSON(w, http.StatusConflict, noDataPlaneYet)
+			return
+		}
+		go func() {
+			sent, err := s.DownTrain(sess, g, count, size, interval)
+			slog.Info("downtrain finished", "action", actionID, "sent", sent, "bytes", g.Sent(), "err", err)
+		}()
+		writeJSON(w, http.StatusAccepted, map[string]any{
+			"action_id": actionID, "count": count, "size_bytes": size, "interval_us": interval,
+			"grant": map[string]any{"max_bytes": g.MaxBytes, "max_kbps": g.MaxKbps},
+		})
+
+	case "big_send":
+		if s.BigSend == nil {
+			writeJSON(w, http.StatusNotImplemented, map[string]string{"error": "big_send not wired"})
+			return
+		}
+		sizes := req.SizesBytes
+		if len(sizes) == 0 {
+			sizes = []int{1200, 1400, 1472, 1500, 2000, 4000}
+		}
+		if len(sizes) > 32 {
+			sizes = sizes[:32]
+		}
+		total := 0
+		for _, x := range sizes {
+			total += clamp(x, dataMinPacket, 9000)
+		}
+		g := sess.NewGrant(actionID, int64(total), 0, session.DefaultGrantLimits)
+		if g == nil {
+			writeJSON(w, http.StatusConflict, noDataPlaneYet)
+			return
+		}
+		go func() {
+			attempted, err := s.BigSend(sess, g, sizes)
+			slog.Info("big_send finished", "action", actionID, "attempted", attempted, "err", err)
+		}()
+		writeJSON(w, http.StatusAccepted, map[string]any{
+			"action_id": actionID, "sizes_bytes": sizes,
+			"grant": map[string]any{"max_bytes": g.MaxBytes, "max_kbps": g.MaxKbps},
+		})
+
 	default:
 		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "unknown or unimplemented action"})
 	}
 }
 
+// dataMinPacket is the smallest datagram that still carries a header + a little payload.
+const dataMinPacket = 40
+
+var noDataPlaneYet = map[string]string{
+	"error": "no data-plane traffic seen yet — send an ECHO first so the destination is verified",
+}
+
+func clamp(v, lo, hi int) int {
+	if v < lo {
+		return lo
+	}
+	if v > hi {
+		return hi
+	}
+	return v
+}
+
 func isTimeout(err error) bool {
 	var ne net.Error
 	return errors.As(err, &ne) && ne.Timeout()