VerseVDI-Data-Plane/gateway/qualification_harness_test.go

package gateway

import (
	"bufio"
	"bytes"
	"compress/gzip"
	"context"
	"crypto/aes"
	"crypto/cipher"
	"crypto/sha256"
	"crypto/tls"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math"
	"net"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"reflect"
	"regexp"
	"runtime"
	"runtime/debug"
	runtimemetrics "runtime/metrics"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	protocol "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol/gen/go/protocol"
)

const (
	qualificationToolVersion                   = "versevdi-gateway-qualification/v3"
	qualificationImpairmentQueuePackets        = 64
	qualificationImpairmentMaxPackets          = 100_000
	qualificationImpairmentPacketCount         = 10_000
	qualificationProcessingLimit               = 5 * time.Millisecond
	qualificationImpairmentSeed         uint64 = 0x3c6a11ce
)

type qualificationMediaProfile struct {
	Name        string
	Codec       string
	BitrateKbps int64
	Duration    time.Duration
	Warmup      time.Duration
	PacketBytes int
}

type qualificationPathTrace struct {
	NativeSetup         bool
	NativeOpen          bool
	NativeUDPIngress    bool
	ApolloRecovered     bool
	ProductionQueue     bool
	ProductionMediaLoop bool
	ProductionPacer     bool
	VerseQUIC           bool
	PublicClientDecode  bool
	PayloadPreserved    bool
}

type qualificationPath struct {
	client    *Client
	server    *Server
	session   *nativeApolloSession
	fixture   *qualificationApolloFixture
	backend   *qualificationTracingBackend
	key       []byte
	flow      string
	frame     uint32
	bootTrace atomic.Bool
	closeOnce sync.Once
	shutdown  func()
}

type qualificationImpairmentProfile struct {
	Name          string
	RTT           time.Duration
	Jitter        time.Duration
	LossPercent   float64
	Reorder       bool
	CapacitySteps []int
}

type qualificationProcessingSummary struct {
	Profile               string         `json:"profile"`
	Codec                 string         `json:"codec"`
	ConfiguredBitrateKbps int64          `json:"configured_bitrate_kbps"`
	ObservedBitrateKbps   float64        `json:"observed_bitrate_kbps"`
	Warmup                time.Duration  `json:"warmup_ns"`
	ConfiguredDuration    time.Duration  `json:"configured_duration_ns"`
	ActualDuration        time.Duration  `json:"actual_duration_ns"`
	Count                 int64          `json:"count"`
	Min                   time.Duration  `json:"min_ns"`
	Median                time.Duration  `json:"median_ns"`
	P90                   time.Duration  `json:"p90_ns"`
	P95                   time.Duration  `json:"p95_ns"`
	P99                   time.Duration  `json:"p99_ns"`
	Max                   time.Duration  `json:"max_ns"`
	Mean                  time.Duration  `json:"mean_ns"`
	StandardDeviation     time.Duration  `json:"standard_deviation_ns"`
	ClockOverhead         time.Duration  `json:"clock_overhead_ns"`
	Histogram             map[string]int `json:"histogram"`
	PayloadSHA256         string         `json:"payload_sha256"`
	RawSamples            string         `json:"raw_samples"`
	RawSamplesSHA256      string         `json:"raw_samples_sha256"`
	RawSamplesBytes       int64          `json:"raw_samples_bytes"`
	ResourceSamples       int            `json:"resource_samples"`
	CPUSeconds            float64        `json:"cpu_seconds"`
	PeakHeapBytes         uint64         `json:"peak_heap_bytes"`
	PeakGoroutines        int            `json:"peak_goroutines"`
	Mallocs               uint64         `json:"mallocs"`
	AllocatedBytes        uint64         `json:"allocated_bytes"`
	RawResources          string         `json:"raw_resources"`
	RawResourcesSHA256    string         `json:"raw_resources_sha256"`
	RawResourcesBytes     int64          `json:"raw_resources_bytes"`
}

type qualificationImpairmentObservation struct {
	Profile                  string                      `json:"profile"`
	MediaProfile             string                      `json:"media_profile"`
	Seed                     uint64                      `json:"seed"`
	Sent                     int                         `json:"sent"`
	Delivered                int                         `json:"delivered"`
	Dropped                  int                         `json:"dropped"`
	InjectedReordered        int                         `json:"injected_reordered"`
	ObservedOutOfOrder       int                         `json:"observed_out_of_order"`
	ObservedRTT              time.Duration               `json:"observed_rtt_ns"`
	ObservedJitter           time.Duration               `json:"observed_jitter_ns"`
	ObservedLossPercent      float64                     `json:"observed_loss_percent"`
	ObservedReorderPercent   float64                     `json:"observed_reorder_percent"`
	ObservedThroughputKbps   float64                     `json:"observed_throughput_kbps"`
	MaxQueuePackets          int                         `json:"max_queue_packets"`
	ConfiguredRTT            time.Duration               `json:"configured_rtt_ns"`
	ConfiguredJitter         time.Duration               `json:"configured_jitter_ns"`
	ConfiguredLossPercent    float64                     `json:"configured_loss_percent"`
	ConfiguredReorder        bool                        `json:"configured_reorder"`
	ConfiguredCapacitySteps  []int                       `json:"configured_capacity_steps_percent"`
	CapacityStepObservations []qualificationCapacityStep `json:"capacity_step_observations,omitempty"`
	RawSamples               string                      `json:"raw_samples"`
	RawSamplesSHA256         string                      `json:"raw_samples_sha256"`
	RawSamplesBytes          int64                       `json:"raw_samples_bytes"`
}

type qualificationCapacityStep struct {
	ReductionPercent  int           `json:"reduction_percent"`
	Convergence       time.Duration `json:"convergence_ns"`
	MaximumFiveSecond int64         `json:"maximum_five_second_bytes"`
	FiveSecondCap     int64         `json:"five_second_cap_bytes"`
}

type qualificationResourceSample struct {
	Elapsed    time.Duration
	CPUSeconds float64
	HeapBytes  uint64
	Goroutines int
	Mallocs    uint64
	Allocated  uint64
}

type qualificationDeliverySample struct {
	At    time.Time
	Bytes int64
}

type qualificationFairnessEvidence struct {
	Evaluation       time.Duration                 `json:"evaluation_ns"`
	PerFlowBytes     map[string]int64              `json:"per_flow_bytes"`
	ShareError       map[string]float64            `json:"share_error"`
	JainIndex        float64                       `json:"jain_index"`
	CapacitySteps    []qualificationCapacityStep   `json:"capacity_steps"`
	Series           []qualificationFairnessSeries `json:"per_flow_aggregate_series"`
	RawSamples       string                        `json:"raw_samples"`
	RawSamplesSHA256 string                        `json:"raw_samples_sha256"`
	RawSamplesBytes  int64                         `json:"raw_samples_bytes"`
}

type qualificationFairnessSeries struct {
	Elapsed        time.Duration    `json:"elapsed_ns"`
	PerFlowBytes   map[string]int64 `json:"per_flow_bytes"`
	AggregateBytes int64            `json:"aggregate_bytes"`
}

type qualificationManifest struct {
	Status          string                               `json:"status"`
	ToolVersion     string                               `json:"tool_version"`
	Command         string                               `json:"command"`
	CandidateCommit string                               `json:"candidate_commit"`
	ProtocolVersion string                               `json:"protocol_version"`
	StartedAt       string                               `json:"started_at"`
	CompletedAt     string                               `json:"completed_at"`
	GoVersion       string                               `json:"go_version"`
	ToolVersions    map[string]string                    `json:"tool_versions"`
	OS              string                               `json:"os"`
	Architecture    string                               `json:"architecture"`
	Topology        string                               `json:"topology"`
	Direction       string                               `json:"direction"`
	QueueDiscipline string                               `json:"queue_discipline"`
	Evidence        []string                             `json:"evidence_classification"`
	Deferred        []string                             `json:"deferred"`
	Processing      []qualificationProcessingSummary     `json:"processing"`
	Impairments     []qualificationImpairmentObservation `json:"impairments"`
	Fairness        qualificationFairnessEvidence        `json:"fairness"`
}

func qualificationMediaProfiles() []qualificationMediaProfile {
	return []qualificationMediaProfile{
		{Name: "1080p60-h264", Codec: "h264", BitrateKbps: 20000, Duration: 10 * time.Minute, Warmup: time.Second, PacketBytes: 1179},
		{Name: "1440p120-hevc", Codec: "hevc", BitrateKbps: 50000, Duration: 10 * time.Minute, Warmup: time.Second, PacketBytes: 1179},
		{Name: "4k60-hevc", Codec: "hevc", BitrateKbps: 80000, Duration: 10 * time.Minute, Warmup: time.Second, PacketBytes: 1179},
	}
}

func qualificationImpairmentProfiles() []qualificationImpairmentProfile {
	return []qualificationImpairmentProfile{
		{Name: "baseline", RTT: 20 * time.Millisecond},
		{Name: "latency", RTT: 150 * time.Millisecond},
		{Name: "jitter", RTT: 50 * time.Millisecond, Jitter: 30 * time.Millisecond},
		{Name: "loss", RTT: 50 * time.Millisecond, Jitter: 10 * time.Millisecond, LossPercent: 5},
		{Name: "reorder", RTT: 100 * time.Millisecond, Jitter: 10 * time.Millisecond, LossPercent: 1, Reorder: true},
		{Name: "constrained", RTT: 50 * time.Millisecond, Jitter: 10 * time.Millisecond, LossPercent: 2, Reorder: true, CapacitySteps: []int{25, 50}},
	}
}

func validateQualificationOutputDir(path string) error {
	if path == "" || !filepath.IsAbs(path) || filepath.Clean(path) == string(filepath.Separator) {
		return errors.New("qualification evidence directory must be a non-root absolute path")
	}
	return nil
}

func qualificationPayload(profile qualificationMediaProfile) []byte {
	payload := make([]byte, profile.PacketBytes)
	if profile.Codec == "h264" {
		copy(payload, []byte{0, 0, 1, 0x65})
	} else {
		copy(payload, []byte{0, 0, 1, 0x26})
	}
	for index := 4; index < len(payload); index++ {
		payload[index] = byte(index*31 + len(profile.Name))
	}
	return payload
}

type qualificationTracingBackend struct {
	native   *NativeApolloBackend
	setups   atomic.Uint64
	opens    atomic.Uint64
	sessions sync.Map
}

func (b *qualificationTracingBackend) Management(ctx context.Context, request LaunchRequest) ([]byte, error) {
	return b.native.Management(ctx, request)
}

func (b *qualificationTracingBackend) Setup(ctx context.Context, request LaunchRequest, management []byte) ([]byte, error) {
	response, err := b.native.Setup(ctx, request, management)
	if err == nil {
		b.setups.Add(1)
	}
	return response, err
}

func (b *qualificationTracingBackend) Open(ctx context.Context, request LaunchRequest, response RTSPResponse) (ProviderSession, error) {
	session, err := b.native.Open(ctx, request, response)
	if err == nil {
		native, ok := session.(*nativeApolloSession)
		if !ok {
			return nil, ErrProviderMalformed
		}
		b.opens.Add(1)
		b.sessions.Store(request.SessionID, native)
	}
	return session, err
}

func (b *qualificationTracingBackend) session(sessionID string) *nativeApolloSession {
	value, _ := b.sessions.Load(sessionID)
	session, _ := value.(*nativeApolloSession)
	return session
}

type qualificationApolloFixture struct {
	sessionID   string
	management  *httptest.Server
	stream      net.Listener
	control     *net.UDPConn
	audio       *net.UDPConn
	video       *net.UDPConn
	videoRemote atomic.Pointer[net.UDPAddr]
	key         atomic.Pointer[[]byte]
	keyReady    chan []byte
	failures    chan error
	closed      atomic.Bool
	closeOnce   sync.Once
	sentPackets atomic.Uint64
	work        protocol.ProviderSessionWork
}

func newQualificationApolloFixture(t *testing.T, serverTLS, clientTLS *tls.Config, sessionID string, profile qualificationMediaProfile) *qualificationApolloFixture {
	t.Helper()
	fixture := &qualificationApolloFixture{sessionID: sessionID, keyReady: make(chan []byte, 1), failures: make(chan error, 8)}
	var err error
	fixture.stream, err = net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	fixture.control, err = net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
	if err != nil {
		fixture.Close()
		t.Fatal(err)
	}
	fixture.audio, err = net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
	if err != nil {
		fixture.Close()
		t.Fatal(err)
	}
	fixture.video, err = net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
	if err != nil {
		fixture.Close()
		t.Fatal(err)
	}
	streamHost, streamPortText, err := net.SplitHostPort(fixture.stream.Addr().String())
	if err != nil {
		fixture.Close()
		t.Fatal(err)
	}
	streamPort, err := strconv.ParseInt(streamPortText, 10, 64)
	if err != nil {
		fixture.Close()
		t.Fatal(err)
	}
	fixture.management = httptest.NewUnstartedServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
		if request.TLS == nil || len(request.TLS.PeerCertificates) != 1 {
			http.Error(response, "mTLS required", http.StatusUnauthorized)
			return
		}
		switch request.URL.Path {
		case "/serverinfo":
			_, _ = response.Write([]byte("<root><uniqueid>qualification-apollo</uniqueid><ServerCodecModeSupport>257</ServerCodecModeSupport><MaxLumaPixelsHEVC>1869449984</MaxLumaPixelsHEVC></root>"))
		case "/applist":
			_, _ = response.Write([]byte("<root><App><ID>42</ID></App></root>"))
		case "/launch":
			key, decodeErr := hex.DecodeString(request.URL.Query().Get("rikey"))
			if decodeErr != nil || len(key) != 16 {
				http.Error(response, "invalid RIK", http.StatusBadRequest)
				return
			}
			copyKey := append([]byte(nil), key...)
			fixture.key.Store(&copyKey)
			select {
			case fixture.keyReady <- copyKey:
			default:
			}
			_, _ = response.Write([]byte("<root status_code=\"200\"><sessionUrl0>rtspenc://" + fixture.stream.Addr().String() + "</sessionUrl0></root>"))
		default:
			http.NotFound(response, request)
		}
	}))
	fixture.management.TLS = serverTLS
	fixture.management.StartTLS()
	managementHost, managementPortText, err := net.SplitHostPort(fixture.management.Listener.Addr().String())
	if err != nil {
		fixture.Close()
		t.Fatal(err)
	}
	managementPort, err := strconv.ParseInt(managementPortText, 10, 64)
	if err != nil {
		fixture.Close()
		t.Fatal(err)
	}
	codec, width, height, fps := "H264", int64(1920), int64(1080), int64(60)
	if profile.Codec == "hevc" {
		codec = "HEVC"
		if strings.Contains(profile.Name, "1440") {
			width, height, fps = 2560, 1440, 120
		} else {
			width, height, fps = 3840, 2160, 60
		}
	}
	pinned := sha256.Sum256(serverTLS.Certificates[0].Certificate[0])
	fixture.work = protocol.ProviderSessionWork{
		Version: "1", SessionID: sessionID, GatewayID: "gateway-1",
		ExpiresAt: "2099-01-01T00:00:00Z", ProviderProfile: ProviderProfileApollo,
		ProviderIdentity: "qualification-apollo#sha256:" + hex.EncodeToString(pinned[:]),
		PolicyVersionID:  "qualification-policy", ApplicationID: "42", ClientID: "qualification-client",
		StreamPolicy: protocol.ProviderStreamPolicy{
			ResolutionWidth: width, ResolutionHeight: height, Fps: fps,
			Codec: codec, BitrateKbps: profile.BitrateKbps, AudioEnabled: true,
		},
		ManagementHost: managementHost, ManagementPort: managementPort,
		StreamHost: streamHost, StreamPort: streamPort,
		ClientCertificatePem: certificatePEM(t, clientTLS.Certificates[0]),
		ClientPrivateKeyPem:  privateKeyPEM(t, clientTLS.Certificates[0]),
		ServerCertificatePem: certificatePEM(t, tls.Certificate{Certificate: [][]byte{serverTLS.Certificates[0].Certificate[1]}}),
		ClipboardPolicy:      protocol.ClipboardPolicy{MaxTextBytes: 65536, MaxUpdatesPerMinute: 30},
	}
	go fixture.serveRTSP()
	go fixture.serveControl()
	go fixture.serveMedia(fixture.audio, false)
	go fixture.serveMedia(fixture.video, true)
	t.Cleanup(fixture.Close)
	return fixture
}

func (f *qualificationApolloFixture) fail(err error) {
	if err == nil || f.closed.Load() {
		return
	}
	select {
	case f.failures <- err:
	default:
	}
}

func (f *qualificationApolloFixture) serveRTSP() {
	key := <-f.keyReady
	codec, err := newEncryptedRTSPCodec(key)
	if err != nil {
		f.fail(err)
		return
	}
	methods := []string{"OPTIONS", "DESCRIBE", "SETUP", "SETUP", "SETUP", "ANNOUNCE", "PLAY"}
	targets := []string{"rtspenc://" + f.stream.Addr().String(), "rtspenc://" + f.stream.Addr().String(), "streamid=audio/0/0", "streamid=video/0/0", "streamid=control/13/0", "streamid=control/13/0", "/"}
	describe := "a=x-ss-general.featureFlags:1\r\na=x-ss-general.encryptionSupported:7\r\na=x-ss-general.encryptionRequested:1\r\na=fmtp:97 surround-params=21101\r\n"
	responses := []string{
		"RTSP/1.0 200 OK\r\nCSeq: 1\r\n\r\n",
		fmt.Sprintf("RTSP/1.0 200 OK\r\nCSeq: 2\r\nContent-Type: application/sdp\r\nContent-Length: %d\r\n\r\n%s", len(describe), describe),
		fmt.Sprintf("RTSP/1.0 200 OK\r\nCSeq: 3\r\nSession: %s;timeout=90\r\nTransport: unicast;server_port=%d\r\nX-SS-Ping-Payload: 0123456789abcdef\r\n\r\n", f.sessionID, f.audio.LocalAddr().(*net.UDPAddr).Port),
		fmt.Sprintf("RTSP/1.0 200 OK\r\nCSeq: 4\r\nSession: %s\r\nTransport: unicast;server_port=%d\r\nX-SS-Ping-Payload: fedcba9876543210\r\n\r\n", f.sessionID, f.video.LocalAddr().(*net.UDPAddr).Port),
		fmt.Sprintf("RTSP/1.0 200 OK\r\nCSeq: 5\r\nSession: %s\r\nTransport: unicast;server_port=%d\r\nX-SS-Connect-Data: 305419896\r\n\r\n", f.sessionID, f.control.LocalAddr().(*net.UDPAddr).Port),
		fmt.Sprintf("RTSP/1.0 200 OK\r\nCSeq: 6\r\nSession: %s\r\n\r\n", f.sessionID),
		fmt.Sprintf("RTSP/1.0 200 OK\r\nCSeq: 7\r\nSession: %s\r\n\r\n", f.sessionID),
	}
	for index, method := range methods {
		connection, acceptErr := f.stream.Accept()
		if acceptErr != nil {
			f.fail(acceptErr)
			return
		}
		header := make([]byte, encryptedRTSPHeaderSize)
		if _, err = io.ReadFull(connection, header); err != nil {
			_ = connection.Close()
			f.fail(err)
			return
		}
		length := binary.BigEndian.Uint32(header[:4]) & 0x7fffffff
		if length > maxApolloRTSPHeaders+maxApolloRTSPBody {
			_ = connection.Close()
			f.fail(ErrProviderMalformed)
			return
		}
		frame := make([]byte, encryptedRTSPHeaderSize+int(length))
		copy(frame, header)
		if _, err = io.ReadFull(connection, frame[encryptedRTSPHeaderSize:]); err != nil {
			_ = connection.Close()
			f.fail(err)
			return
		}
		plaintext, openErr := codec.OpenClient(frame)
		firstLine, _, _ := strings.Cut(string(plaintext), "\r\n")
		if openErr != nil || firstLine != method+" "+targets[index]+" RTSP/1.0" {
			_ = connection.Close()
			f.fail(ErrProviderMalformed)
			return
		}
		responseFrame, sealErr := hostEncryptedRTSPFrameNoTest(key, uint32(index+1), []byte(responses[index]))
		if sealErr != nil {
			_ = connection.Close()
			f.fail(sealErr)
			return
		}
		_, err = connection.Write(responseFrame)
		_ = connection.Close()
		if err != nil {
			f.fail(err)
			return
		}
	}
}

func hostEncryptedRTSPFrameNoTest(key []byte, sequence uint32, plaintext []byte) ([]byte, error) {
	block, err := aes.NewCipher(key)
	if err != nil {
		return nil, err
	}
	aead, err := cipher.NewGCM(block)
	if err != nil {
		return nil, err
	}
	nonce := encryptedRTSPNonce(sequence, 'H', 'R')
	sealed := aead.Seal(nil, nonce[:], plaintext, nil)
	frame := make([]byte, encryptedRTSPHeaderSize+len(plaintext))
	binary.BigEndian.PutUint32(frame[:4], uint32(len(plaintext))|0x80000000)
	binary.BigEndian.PutUint32(frame[4:8], sequence)
	copy(frame[8:24], sealed[len(plaintext):])
	copy(frame[24:], sealed[:len(plaintext)])
	return frame, nil
}

func (f *qualificationApolloFixture) serveControl() {
	buffer := make([]byte, apolloENetMaximumPacket)
	count, remote, err := f.control.ReadFromUDP(buffer)
	if err != nil {
		f.fail(err)
		return
	}
	connect := buffer[:count]
	if count != 52 || connect[4] != apolloENetConnect|apolloENetAcknowledged ||
		binary.BigEndian.Uint32(connect[20:24]) != apolloENetChannels {
		f.fail(ErrProviderMalformed)
		return
	}
	verify := make([]byte, 48)
	apolloENetHeader(verify, 0, 0, time.Now())
	verify[4] = apolloENetVerifyConnect | apolloENetAcknowledged
	verify[5] = 0xff
	binary.BigEndian.PutUint16(verify[6:8], 1)
	binary.BigEndian.PutUint16(verify[8:10], 7)
	verify[10], verify[11] = 2, 3
	binary.BigEndian.PutUint32(verify[12:16], 1400)
	binary.BigEndian.PutUint32(verify[16:20], 32768)
	binary.BigEndian.PutUint32(verify[20:24], apolloENetChannels)
	binary.BigEndian.PutUint32(verify[44:48], binary.BigEndian.Uint32(connect[44:48]))
	if _, err = f.control.WriteToUDP(verify, remote); err != nil {
		f.fail(err)
		return
	}
	for {
		count, _, err = f.control.ReadFromUDP(buffer)
		if err != nil {
			f.fail(err)
			return
		}
		packet := buffer[:count]
		if len(packet) < 8 {
			f.fail(ErrProviderMalformed)
			return
		}
		command, channel := packet[4]&apolloENetCommandMask, packet[5]
		sequence := binary.BigEndian.Uint16(packet[6:8])
		switch command {
		case 1:
		case apolloENetSendReliable, apolloENetPing, apolloENetDisconnect:
			if _, err = f.control.WriteToUDP(sourceShapedENetAcknowledgePacket(7, 2, channel, sequence), remote); err != nil {
				f.fail(err)
				return
			}
			if command == apolloENetDisconnect {
				return
			}
		case apolloENetSendUnsequenced:
		default:
			f.fail(ErrProviderMalformed)
			return
		}
	}
}

func (f *qualificationApolloFixture) serveMedia(socket *net.UDPConn, video bool) {
	buffer := make([]byte, apolloMediaMaximumPacket)
	for {
		_, remote, err := socket.ReadFromUDP(buffer)
		if err != nil {
			f.fail(err)
			return
		}
		if video && f.videoRemote.Load() == nil {
			copyRemote := *remote
			f.videoRemote.Store(&copyRemote)
		}
	}
}

func (f *qualificationApolloFixture) sendVideo(ctx context.Context, packets [][]byte) error {
	for f.videoRemote.Load() == nil {
		select {
		case err := <-f.failures:
			return err
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(time.Millisecond):
		}
	}
	remote := f.videoRemote.Load()
	for _, packet := range packets {
		if _, err := f.video.WriteToUDP(packet, remote); err != nil {
			return err
		}
		f.sentPackets.Add(1)
	}
	return nil
}

func (f *qualificationApolloFixture) streamKey() ([]byte, error) {
	key := f.key.Load()
	if key == nil || len(*key) != 16 {
		return nil, ErrProviderMalformed
	}
	return append([]byte(nil), (*key)...), nil
}

func (f *qualificationApolloFixture) Close() {
	if f == nil {
		return
	}
	f.closeOnce.Do(func() {
		f.closed.Store(true)
		if f.management != nil {
			f.management.Close()
		}
		if f.stream != nil {
			_ = f.stream.Close()
		}
		for _, socket := range []*net.UDPConn{f.control, f.audio, f.video} {
			if socket != nil {
				_ = socket.Close()
			}
		}
	})
}

type qualificationAdmissionRecord struct {
	authority protocol.SessionAuthority
	work      protocol.ProviderSessionWork
	used      bool
}

type qualificationAdmission struct {
	mu      sync.Mutex
	records map[string]*qualificationAdmissionRecord
}

func (a *qualificationAdmission) Admit(_ context.Context, request protocol.TunnelAdmissionRequest) (protocol.SessionAuthority, error) {
	a.mu.Lock()
	defer a.mu.Unlock()
	record := a.records[request.SessionID]
	if record == nil || record.used || request.GatewayID != record.authority.GatewayID ||
		request.Audience != record.authority.Audience || !reflect.DeepEqual(request.Capabilities, record.authority.Capabilities) {
		return protocol.SessionAuthority{}, ErrAdmissionRejected
	}
	record.used = true
	return record.authority, nil
}

func (a *qualificationAdmission) ProviderWork(_ context.Context, authority protocol.SessionAuthority) (protocol.ProviderSessionWork, error) {
	a.mu.Lock()
	defer a.mu.Unlock()
	record := a.records[authority.SessionID]
	if record == nil || !record.used || !reflect.DeepEqual(authority, record.authority) {
		return protocol.ProviderSessionWork{}, ErrAdmissionRejected
	}
	return record.work, nil
}

func (*qualificationAdmission) Release(context.Context, protocol.SessionAuthority) error { return nil }

type qualificationFleet struct {
	t         *testing.T
	server    *Server
	paths     []*qualificationPath
	clients   []*Client
	cancel    context.CancelFunc
	serveDone chan error
	closeOnce sync.Once
}

func newQualificationFleet(t *testing.T, count int, profile qualificationMediaProfile, pacerKbps int64) *qualificationFleet {
	t.Helper()
	serverTLS, clientTLS := testTLS(t)
	backend := &qualificationTracingBackend{native: NewNativeApolloBackend()}
	admission := &qualificationAdmission{records: make(map[string]*qualificationAdmissionRecord, count)}
	fixtures := make([]*qualificationApolloFixture, 0, count)
	for index := 0; index < count; index++ {
		sessionID := fmt.Sprintf("qualification-flow-%d", index+1)
		fixture := newQualificationApolloFixture(t, serverTLS, clientTLS, sessionID, profile)
		authority := protocol.SessionAuthority{
			Version: "1", SessionID: sessionID, GatewayID: "gateway-1", Audience: "versevdi-gateway",
			ExpiresAt:    time.Now().Add(2 * time.Minute).UTC().Format(time.RFC3339Nano),
			Capabilities: DefaultCapabilities(), ProviderProfile: ProviderProfileApollo,
			ProviderIdentity: fixture.work.ProviderIdentity,
		}
		work := fixture.work
		work.ExpiresAt = authority.ExpiresAt
		admission.records[sessionID] = &qualificationAdmissionRecord{authority: authority, work: work}
		fixtures = append(fixtures, fixture)
	}
	server, err := NewServer(ServerConfig{
		ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: "gateway-1",
		Capabilities: DefaultCapabilities(), ProviderCapabilities: DefaultCapabilities(),
		Admission: admission, ProviderStateReporter: &recordingProviderStateReporter{},
		Provider: NewApolloAdapter(backend, ProviderIdentity{}), PacerKbps: pacerKbps,
	})
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	fleet := &qualificationFleet{t: t, server: server, cancel: cancel, serveDone: make(chan error, 1)}
	go func() { fleet.serveDone <- server.Serve(ctx) }()
	for index, fixture := range fixtures {
		sessionID := fmt.Sprintf("qualification-flow-%d", index+1)
		request := protocol.TunnelAdmissionRequest{
			Version: "1", SessionID: sessionID, GatewayID: "gateway-1", Audience: "versevdi-gateway",
			Grant: strings.Repeat(string(rune('a'+index)), 64), ClientNonce: fmt.Sprintf("nonce-fleet-%06d", index),
			DeviceSignature: strings.Repeat("s", 86), Capabilities: DefaultCapabilities(),
		}
		client, err := Dial(context.Background(), server.Addr().String(), clientTLS, request)
		if err != nil {
			fleet.Close()
			t.Fatal(err)
		}
		session := backend.session(sessionID)
		key, keyErr := fixture.streamKey()
		if keyErr != nil || session == nil {
			fleet.Close()
			t.Fatalf("qualification fleet session %s unavailable: %v", sessionID, keyErr)
		}
		fleet.clients = append(fleet.clients, client)
		fleet.paths = append(fleet.paths, &qualificationPath{
			client: client, server: server, session: session, fixture: fixture,
			backend: backend, key: key, shutdown: func() {},
		})
	}
	t.Cleanup(fleet.Close)
	return fleet
}

func (f *qualificationFleet) Close() {
	if f == nil {
		return
	}
	f.closeOnce.Do(func() {
		for _, client := range f.clients {
			_ = client.Close()
		}
		f.cancel()
		_ = f.server.Close()
		if err := <-f.serveDone; err != nil {
			f.t.Errorf("serve qualification fleet: %v", err)
		}
	})
}

func newQualificationPath(t *testing.T, profile qualificationMediaProfile, pacerKbps int64) *qualificationPath {
	t.Helper()
	serverTLS, clientTLS := testTLS(t)
	fixture := newQualificationApolloFixture(t, serverTLS, clientTLS, "qualification-session", profile)
	backend := &qualificationTracingBackend{native: NewNativeApolloBackend()}
	provider := NewApolloAdapter(backend, ProviderIdentity{})
	authority := protocol.SessionAuthority{
		Version: "1", SessionID: "qualification-session", GatewayID: "gateway-1",
		Audience: "versevdi-gateway", ReconnectSequence: 0,
		ExpiresAt:    time.Now().Add(45 * time.Minute).UTC().Format(time.RFC3339Nano),
		Capabilities: DefaultCapabilities(), ProviderProfile: ProviderProfileApollo,
		ProviderIdentity: fixture.work.ProviderIdentity,
	}
	work := fixture.work
	work.ExpiresAt = authority.ExpiresAt
	admission := &oneTimeAdmission{authority: authority, released: make(chan struct{}), disableClipboard: true, providerWork: &work}
	server, err := NewServer(ServerConfig{
		ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: authority.GatewayID,
		Capabilities: DefaultCapabilities(), ProviderCapabilities: DefaultCapabilities(),
		Admission: admission, ProviderStateReporter: &recordingProviderStateReporter{},
		Provider: provider, PacerKbps: pacerKbps,
	})
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	serveDone := make(chan error, 1)
	go func() { serveDone <- server.Serve(ctx) }()
	request := protocol.TunnelAdmissionRequest{
		Version: "1", SessionID: authority.SessionID, GatewayID: authority.GatewayID,
		Audience: authority.Audience, Grant: strings.Repeat("g", 64),
		ClientNonce: "nonce-qualification", DeviceSignature: strings.Repeat("s", 86),
		Capabilities: DefaultCapabilities(),
	}
	client, err := Dial(context.Background(), server.Addr().String(), clientTLS, request)
	if err != nil {
		cancel()
		_ = server.Close()
		t.Fatal(err)
	}
	session := backend.session(authority.SessionID)
	key, err := fixture.streamKey()
	if err != nil || session == nil {
		_ = client.Close()
		cancel()
		_ = server.Close()
		t.Fatalf("native qualification session unavailable: %v", err)
	}
	path := &qualificationPath{client: client, server: server, session: session, fixture: fixture, backend: backend, key: key}
	path.shutdown = func() {
		_ = client.Close()
		cancel()
		_ = server.Close()
		if err := <-serveDone; err != nil {
			t.Errorf("serve qualification path: %v", err)
		}
	}
	t.Cleanup(path.Close)
	return path
}

func (p *qualificationPath) Close() {
	if p != nil {
		p.closeOnce.Do(p.shutdown)
	}
}

func (p *qualificationPath) traverse(t *testing.T, payload []byte) (qualificationPathTrace, time.Duration, error) {
	t.Helper()
	beforeMetrics := p.server.Metrics()
	beforeIngress := p.session.mediaIngress.Load()
	beforeRecovered := p.session.mediaRecovered.Load()
	beforeEnqueued := p.session.mediaEnqueued.Load()
	beforePacer := p.server.pacer.reservations.Load()
	trace, err := p.emit(t, payload)
	if err != nil {
		return trace, 0, err
	}
	recovered, err := p.receivePayload(context.Background())
	if err != nil {
		return trace, 0, err
	}
	afterMetrics := p.server.Metrics()
	deadline := time.Now().Add(2 * time.Second)
	for (afterMetrics.ProcessingSamples < beforeMetrics.ProcessingSamples+1 ||
		afterMetrics.MediaPackets <= beforeMetrics.MediaPackets) && time.Now().Before(deadline) {
		runtime.Gosched()
		afterMetrics = p.server.Metrics()
	}
	trace.NativeUDPIngress = p.session.mediaIngress.Load() > beforeIngress
	trace.ApolloRecovered = p.session.mediaRecovered.Load() > beforeRecovered
	trace.ProductionQueue = p.session.mediaEnqueued.Load() > beforeEnqueued
	trace.ProductionMediaLoop = afterMetrics.ProcessingSamples > beforeMetrics.ProcessingSamples
	trace.ProductionPacer = p.server.pacer.reservations.Load() > beforePacer
	trace.VerseQUIC = afterMetrics.MediaPackets > beforeMetrics.MediaPackets
	trace.PublicClientDecode = true
	trace.PayloadPreserved = bytes.Equal(recovered, payload)
	if p.bootTrace.CompareAndSwap(false, true) {
		trace.NativeSetup = p.backend.setups.Load() == 1
		trace.NativeOpen = p.backend.opens.Load() == 1
	}
	return trace, time.Duration(afterMetrics.ProcessingDelayNanos - beforeMetrics.ProcessingDelayNanos), nil
}

func (p *qualificationPath) emit(t *testing.T, payload []byte) (qualificationPathTrace, error) {
	t.Helper()
	if p == nil || p.session == nil || len(payload) == 0 || len(payload) > 2*apolloVideoShardPayloadSize-8 {
		return qualificationPathTrace{}, ErrProviderMalformed
	}
	p.frame++
	packets := qualificationSourceVideoPackets(t, p.key, p.frame, payload)
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	if err := p.fixture.sendVideo(ctx, packets); err != nil {
		return qualificationPathTrace{}, err
	}
	return qualificationPathTrace{}, nil
}

func (p *qualificationPath) receivePayload(parent context.Context) ([]byte, error) {
	ctx, cancel := context.WithTimeout(parent, 2*time.Second)
	defer cancel()
	var recovered []byte
	var sequence uint32
	var fragmentCount byte
	for {
		frame, err := p.client.ReceiveFrame(ctx)
		if err != nil {
			return nil, err
		}
		if frame.Channel != ChannelVideo {
			continue
		}
		if fragmentCount == 0 {
			sequence, fragmentCount = frame.Sequence, frame.FragmentCount
		}
		if frame.Sequence != sequence || frame.FragmentIndex != byte(len(recovered)/1179) {
			return nil, errors.New("qualification QUIC fragments reordered")
		}
		recovered = append(recovered, frame.Payload...)
		if frame.FragmentIndex+1 == fragmentCount {
			break
		}
	}
	return recovered, nil
}

func qualificationSourceVideoPackets(t *testing.T, key []byte, frame uint32, encoded []byte) [][]byte {
	t.Helper()
	if len(encoded) <= apolloVideoShardPayloadSize-8 {
		payload := make([]byte, apolloVideoShardPayloadSize)
		payload[0], payload[3] = 0x01, 0x01
		binary.LittleEndian.PutUint16(payload[4:6], uint16(8+len(encoded)))
		copy(payload[8:], encoded)
		raw := sourceShapedVideoRaw(frame, uint16(frame), frame, 0x07, 1, 0, 0, payload)
		return [][]byte{sourceEncryptVideoRaw(t, key, raw, qualificationVideoIV(frame, 0))}
	}
	combined := make([]byte, 2*apolloVideoShardPayloadSize)
	combined[0], combined[3] = 0x01, 0x01
	binary.LittleEndian.PutUint16(combined[4:6], uint16(8+len(encoded)-apolloVideoShardPayloadSize))
	copy(combined[8:], encoded)
	first := sourceShapedVideoRaw(frame, uint16(frame*3), frame*3, 0x05, 2, 50, 0, combined[:apolloVideoShardPayloadSize])
	second := sourceShapedVideoRaw(frame, uint16(frame*3+1), frame*3+1, 0x03, 2, 50, 1, combined[apolloVideoShardPayloadSize:])
	parity := make([]byte, len(first))
	for index := range parity {
		parity[index] = first[index] ^ sourceGFMultiply(second[index], 142)
	}
	sourceConfigureVideoShard(parity, frame, uint16(frame*3+2), frame*3+2, 2, 50, 2)
	return [][]byte{
		sourceEncryptVideoRaw(t, key, second, qualificationVideoIV(frame, 1)),
		sourceEncryptVideoRaw(t, key, parity, qualificationVideoIV(frame, 2)),
	}
}

func qualificationVideoIV(frame uint32, shard byte) string {
	return fmt.Sprintf("%09x%01xQV", frame, shard)
}

func qualificationProductionPathSmoke(t *testing.T, profile qualificationMediaProfile) qualificationPathTrace {
	t.Helper()
	path := newQualificationPath(t, profile, profile.BitrateKbps)
	defer path.Close()
	trace, _, err := path.traverse(t, qualificationPayload(profile))
	if err != nil {
		t.Fatal(err)
	}
	return trace
}

func TestQualificationTraversesNativePublicGatewayPath(t *testing.T) {
	profile := qualificationMediaProfiles()[0]
	profile.BitrateKbps = 100000
	trace := qualificationProductionPathSmoke(t, profile)
	if !trace.NativeSetup || !trace.NativeOpen || !trace.NativeUDPIngress || !trace.ApolloRecovered ||
		!trace.ProductionQueue || !trace.ProductionMediaLoop || !trace.ProductionPacer ||
		!trace.VerseQUIC || !trace.PublicClientDecode || !trace.PayloadPreserved {
		t.Fatalf("qualification path skipped production stages: %#v", trace)
	}
}

func summarizeQualificationSamples(samples []time.Duration) (qualificationProcessingSummary, error) {
	if len(samples) == 0 {
		return qualificationProcessingSummary{}, errors.New("qualification has no processing samples")
	}
	var mean, m2 float64
	histogram := newQualificationHistogram()
	for index, sample := range samples {
		value := float64(sample)
		delta := value - mean
		mean += delta / float64(index+1)
		m2 += delta * (value - mean)
		observeQualificationHistogram(histogram, sample)
	}
	sort.Slice(samples, func(first, second int) bool { return samples[first] < samples[second] })
	return qualificationProcessingSummary{
		Count:             int64(len(samples)),
		Min:               samples[0],
		Median:            qualificationPercentile(samples, 0.50),
		P90:               qualificationPercentile(samples, 0.90),
		P95:               qualificationPercentile(samples, 0.95),
		P99:               qualificationPercentile(samples, 0.99),
		Max:               samples[len(samples)-1],
		Mean:              time.Duration(mean),
		StandardDeviation: time.Duration(math.Sqrt(m2 / float64(len(samples)))),
		Histogram:         histogram,
	}, nil
}

func qualificationPercentile(samples []time.Duration, percentile float64) time.Duration {
	index := int(math.Ceil(percentile*float64(len(samples)))) - 1
	if index < 0 {
		index = 0
	}
	return samples[index]
}

func enforceQualificationProcessingGate(summary qualificationProcessingSummary) error {
	if summary.Count < 1 || summary.P95 > qualificationProcessingLimit {
		return fmt.Errorf("processing p95 %s exceeds %s", summary.P95, qualificationProcessingLimit)
	}
	return nil
}

func newQualificationHistogram() map[string]int {
	return map[string]int{
		"le_1us": 0, "le_5us": 0, "le_10us": 0, "le_25us": 0,
		"le_50us": 0, "le_100us": 0, "le_250us": 0, "le_500us": 0,
		"le_1ms": 0, "le_2ms": 0, "le_5ms": 0, "gt_5ms": 0,
	}
}

func observeQualificationHistogram(histogram map[string]int, sample time.Duration) {
	buckets := []struct {
		name  string
		limit time.Duration
	}{
		{"le_1us", time.Microsecond}, {"le_5us", 5 * time.Microsecond},
		{"le_10us", 10 * time.Microsecond}, {"le_25us", 25 * time.Microsecond},
		{"le_50us", 50 * time.Microsecond}, {"le_100us", 100 * time.Microsecond},
		{"le_250us", 250 * time.Microsecond}, {"le_500us", 500 * time.Microsecond},
		{"le_1ms", time.Millisecond}, {"le_2ms", 2 * time.Millisecond},
		{"le_5ms", 5 * time.Millisecond},
	}
	observed := false
	for _, bucket := range buckets {
		if sample <= bucket.limit {
			histogram[bucket.name]++
			observed = true
		}
	}
	if !observed {
		histogram["gt_5ms"]++
	}
}

func runQualificationImpairment(t *testing.T, profile qualificationImpairmentProfile, media qualificationMediaProfile, packetCount int, rawPath string) (qualificationImpairmentObservation, error) {
	t.Helper()
	if packetCount < 1 || packetCount > qualificationImpairmentMaxPackets || media.PacketBytes < 1 ||
		media.PacketBytes > 1179 || !qualificationKnownImpairment(profile) {
		return qualificationImpairmentObservation{}, errors.New("qualification impairment bounds invalid")
	}
	file, err := os.OpenFile(rawPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640)
	if err != nil {
		return qualificationImpairmentObservation{}, err
	}
	compressed := gzip.NewWriter(file)
	buffered := bufio.NewWriter(compressed)
	closed := false
	defer func() {
		if !closed {
			_ = buffered.Flush()
			_ = compressed.Close()
			_ = file.Close()
		}
	}()
	if _, err := buffered.WriteString("source_sequence,sent_ns,delivered_ns,processing_ns,outcome,delivery_order,bytes,queue_packets\n"); err != nil {
		return qualificationImpairmentObservation{}, err
	}
	state := qualificationImpairmentSeed
	random := func() uint64 {
		state ^= state << 13
		state ^= state >> 7
		state ^= state << 17
		return state
	}
	spacing := time.Duration(int64(time.Second) * int64(media.PacketBytes) * 8 / (media.BitrateKbps * 1000))
	if spacing < time.Nanosecond {
		spacing = time.Nanosecond
	}
	observation := qualificationImpairmentObservation{
		Profile: profile.Name, MediaProfile: media.Name, Seed: qualificationImpairmentSeed,
		Sent: packetCount, ConfiguredRTT: profile.RTT, ConfiguredJitter: profile.Jitter,
		ConfiguredLossPercent: profile.LossPercent, ConfiguredReorder: profile.Reorder,
		ConfiguredCapacitySteps: append([]int(nil), profile.CapacitySteps...),
	}
	path := newQualificationPath(t, media, media.BitrateKbps)
	defer path.Close()
	started := time.Now()
	payload := qualificationPayload(media)
	var deliveries []qualificationDeliverySample
	var totalRTT, totalJitter, previousRTT time.Duration
	previousDelivered := -1
	deliveryOrder := 0
	type pendingPacket struct {
		index  int
		jitter time.Duration
	}
	pending := pendingPacket{index: -1}
	stepAt := make(map[int]time.Time, len(profile.CapacitySteps))
	deliver := func(packet pendingPacket) error {
		if len(profile.CapacitySteps) == 2 {
			switch {
			case packet.index >= packetCount*2/3 && stepAt[profile.CapacitySteps[1]].IsZero():
				path.server.pacer.setKbps(media.BitrateKbps * int64(100-profile.CapacitySteps[1]) / 100)
				stepAt[profile.CapacitySteps[1]] = time.Now()
			case packet.index >= packetCount/3 && stepAt[profile.CapacitySteps[0]].IsZero():
				path.server.pacer.setKbps(media.BitrateKbps * int64(100-profile.CapacitySteps[0]) / 100)
				stepAt[profile.CapacitySteps[0]] = time.Now()
			}
		}
		sentAt := started.Add(time.Duration(packet.index) * spacing)
		target := sentAt.Add(profile.RTT/2 + packet.jitter)
		if delay := time.Until(target); delay > 0 {
			time.Sleep(delay)
		}
		current := append([]byte(nil), payload...)
		binary.BigEndian.PutUint32(current[len(current)-4:], uint32(packet.index))
		trace, processing, err := path.traverse(t, current)
		if err != nil || !trace.PayloadPreserved || !trace.ProductionPacer {
			if err == nil {
				err = errors.New("impaired packet bypassed production gateway path")
			}
			return err
		}
		deliveredAt := time.Now()
		rtt := 2 * deliveredAt.Sub(sentAt)
		totalRTT += rtt
		if previousRTT != 0 {
			delta := rtt - previousRTT
			if delta < 0 {
				delta = -delta
			}
			totalJitter += delta
		}
		previousRTT = rtt
		if previousDelivered >= 0 && packet.index < previousDelivered {
			observation.ObservedOutOfOrder++
		}
		previousDelivered = packet.index
		observation.Delivered++
		deliveryOrder++
		wireBytes := int64(len(current) + frameHeaderSize)
		deliveries = append(deliveries, qualificationDeliverySample{At: deliveredAt, Bytes: wireBytes})
		queuePackets := int(path.session.mediaQueueMaximum.Load())
		if _, err := fmt.Fprintf(buffered, "%d,%d,%d,%d,delivered,%d,%d,%d\n", packet.index,
			sentAt.Sub(started).Nanoseconds(), deliveredAt.Sub(started).Nanoseconds(),
			processing.Nanoseconds(), deliveryOrder, len(current), queuePackets); err != nil {
			return err
		}
		observation.MaxQueuePackets = max(observation.MaxQueuePackets, queuePackets)
		return nil
	}
	for index := 0; index < packetCount; index++ {
		jitter := time.Duration(0)
		if profile.Jitter > 0 {
			width := uint64(profile.Jitter*2 + 1)
			jitter = time.Duration(random()%width) - profile.Jitter
		}
		if float64(random()%10_000) < profile.LossPercent*100 {
			observation.Dropped++
			if _, err := fmt.Fprintf(buffered, "%d,%d,0,0,dropped,0,0,%d\n", index, time.Duration(index)*spacing, path.session.mediaQueueMaximum.Load()); err != nil {
				return qualificationImpairmentObservation{}, err
			}
			continue
		}
		packet := pendingPacket{index: index, jitter: jitter}
		if profile.Reorder && index%20 == 18 {
			pending = packet
			continue
		}
		if err := deliver(packet); err != nil {
			return qualificationImpairmentObservation{}, err
		}
		if pending.index >= 0 {
			if err := deliver(pending); err != nil {
				return qualificationImpairmentObservation{}, err
			}
			pending.index = -1
			observation.InjectedReordered++
		}
	}
	if pending.index >= 0 {
		if err := deliver(pending); err != nil {
			return qualificationImpairmentObservation{}, err
		}
	}
	if err := buffered.Flush(); err != nil {
		return qualificationImpairmentObservation{}, err
	}
	if err := compressed.Close(); err != nil {
		return qualificationImpairmentObservation{}, err
	}
	if err := file.Close(); err != nil {
		return qualificationImpairmentObservation{}, err
	}
	closed = true
	if observation.Delivered > 0 {
		observation.ObservedRTT = totalRTT / time.Duration(observation.Delivered)
		if observation.Delivered > 1 {
			observation.ObservedJitter = totalJitter / time.Duration(observation.Delivered-1)
		}
		observation.ObservedThroughputKbps = float64(observation.Delivered*media.PacketBytes*8) / time.Since(started).Seconds() / 1000
	}
	observation.ObservedLossPercent = float64(observation.Dropped) * 100 / float64(packetCount)
	observation.ObservedReorderPercent = float64(observation.ObservedOutOfOrder) * 100 / float64(packetCount)
	if packetCount >= qualificationImpairmentPacketCount {
		lowerThroughput := float64(media.BitrateKbps) * (1 - profile.LossPercent/100) * 0.90
		upperThroughput := float64(media.BitrateKbps) * 1.05
		if observation.ObservedThroughputKbps < lowerThroughput || observation.ObservedThroughputKbps > upperThroughput {
			return qualificationImpairmentObservation{}, fmt.Errorf("observed throughput %.2f outside [%.2f,%.2f]", observation.ObservedThroughputKbps, lowerThroughput, upperThroughput)
		}
	}
	observation.RawSamples = filepath.Base(rawPath)
	observation.RawSamplesSHA256, observation.RawSamplesBytes, err = qualificationFileSHA256(rawPath)
	if err != nil {
		return qualificationImpairmentObservation{}, err
	}
	for _, reduction := range profile.CapacitySteps {
		bytesPerSecond := media.BitrateKbps * int64(100-reduction) * 1000 / 100 / 8
		stepDeliveries := qualificationDeliveriesAfter(deliveries, stepAt[reduction])
		convergence := qualificationMeasuredConvergence(stepDeliveries, stepAt[reduction], bytesPerSecond)
		maximum := qualificationMaximumDeliveryBytes(stepDeliveries, 5*time.Second)
		observation.CapacityStepObservations = append(observation.CapacityStepObservations, qualificationCapacityStep{
			ReductionPercent: reduction, Convergence: convergence,
			MaximumFiveSecond: maximum, FiveSecondCap: bytesPerSecond * 5,
		})
		if packetCount >= qualificationImpairmentPacketCount &&
			(convergence > 10*time.Second || maximum > bytesPerSecond*5*105/100) {
			return qualificationImpairmentObservation{}, fmt.Errorf("capacity step %d failed measured convergence=%s five-second=%d", reduction, convergence, maximum)
		}
	}
	if observation.Delivered+observation.Dropped != observation.Sent ||
		observation.MaxQueuePackets > qualificationImpairmentQueuePackets {
		return qualificationImpairmentObservation{}, errors.New("qualification impairment accounting invalid")
	}
	return observation, nil
}

func qualificationKnownImpairment(profile qualificationImpairmentProfile) bool {
	for _, known := range qualificationImpairmentProfiles() {
		if profile.Name != known.Name || profile.RTT != known.RTT || profile.Jitter != known.Jitter ||
			profile.LossPercent != known.LossPercent || profile.Reorder != known.Reorder ||
			len(profile.CapacitySteps) != len(known.CapacitySteps) {
			continue
		}
		match := true
		for index := range known.CapacitySteps {
			match = match && profile.CapacitySteps[index] == known.CapacitySteps[index]
		}
		if match {
			return true
		}
	}
	return false
}

func qualificationDeliveriesAfter(deliveries []qualificationDeliverySample, start time.Time) []qualificationDeliverySample {
	index := sort.Search(len(deliveries), func(index int) bool { return !deliveries[index].At.Before(start) })
	return deliveries[index:]
}

func qualificationMeasuredConvergence(deliveries []qualificationDeliverySample, start time.Time, targetBytesPerSecond int64) time.Duration {
	const window = 250 * time.Millisecond
	const requiredWindows = 4
	consecutive := 0
	for offset := time.Duration(0); offset <= 10*time.Second; offset += window {
		windowStart := start.Add(offset)
		var total int64
		for _, delivery := range deliveries {
			if !delivery.At.Before(windowStart) && delivery.At.Before(windowStart.Add(window)) {
				total += delivery.Bytes
			}
		}
		rate := total * int64(time.Second) / int64(window)
		if rate >= targetBytesPerSecond*90/100 && rate <= targetBytesPerSecond*105/100 {
			consecutive++
			if consecutive == requiredWindows {
				return offset + window
			}
		} else {
			consecutive = 0
		}
		if len(deliveries) > 0 && windowStart.After(deliveries[len(deliveries)-1].At) {
			break
		}
	}
	return 11 * time.Second
}

func qualificationMaximumDeliveryBytes(deliveries []qualificationDeliverySample, window time.Duration) int64 {
	var maximum, total int64
	for first, last := 0, 0; first < len(deliveries); first++ {
		for last < len(deliveries) && deliveries[last].At.Sub(deliveries[first].At) <= window {
			total += deliveries[last].Bytes
			last++
		}
		if total > maximum {
			maximum = total
		}
		total -= deliveries[first].Bytes
	}
	return maximum
}

func runQualificationProcessing(t *testing.T, profile qualificationMediaProfile, rawPath string) (qualificationProcessingSummary, error) {
	t.Helper()
	payload := qualificationPayload(profile)
	if len(payload) < 4 {
		return qualificationProcessingSummary{}, errors.New("qualification payload too small")
	}
	pacerKbps := (profile.BitrateKbps*int64(profile.PacketBytes+frameHeaderSize) + int64(profile.PacketBytes) - 1) / int64(profile.PacketBytes)
	path := newQualificationPath(t, profile, pacerKbps)
	defer path.Close()
	if err := runQualificationWarmup(t, path, profile, payload); err != nil {
		return qualificationProcessingSummary{}, err
	}
	file, err := os.OpenFile(rawPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640)
	if err != nil {
		return qualificationProcessingSummary{}, err
	}
	compressed, err := gzip.NewWriterLevel(file, gzip.BestSpeed)
	if err != nil {
		_ = file.Close()
		return qualificationProcessingSummary{}, err
	}
	buffered := bufio.NewWriterSize(compressed, 1<<20)
	closed := false
	defer func() {
		if !closed {
			_ = buffered.Flush()
			_ = compressed.Close()
			_ = file.Close()
		}
	}()
	if _, err := buffered.WriteString("elapsed_ns,processing_ns\n"); err != nil {
		return qualificationProcessingSummary{}, err
	}
	bytesPerSecond := profile.BitrateKbps * 1000 / 8
	targetBytes := bytesPerSecond * profile.Duration.Nanoseconds() / int64(time.Second)
	targetPackets := (targetBytes + int64(profile.PacketBytes) - 1) / int64(profile.PacketBytes)
	samples := make([]time.Duration, 0, int(targetPackets))
	started := time.Now()
	resources := []qualificationResourceSample{qualificationRuntimeSample(started)}
	lastResourceSample := started
	var processed int64
	for processed < targetPackets || time.Since(started) < profile.Duration {
		current := append([]byte(nil), payload...)
		binary.BigEndian.PutUint32(current[len(current)-4:], uint32(processed))
		trace, sample, traverseErr := path.traverse(t, current)
		if traverseErr != nil {
			return qualificationProcessingSummary{}, traverseErr
		}
		if !trace.NativeUDPIngress || !trace.ApolloRecovered || !trace.ProductionQueue ||
			!trace.ProductionMediaLoop || !trace.ProductionPacer || !trace.VerseQUIC ||
			!trace.PublicClientDecode || !trace.PayloadPreserved {
			return qualificationProcessingSummary{}, fmt.Errorf("qualification bypassed the production provider-to-client path: %#v", trace)
		}
		samples = append(samples, sample)
		if _, writeErr := fmt.Fprintf(buffered, "%d,%d\n", time.Since(started).Nanoseconds(), sample.Nanoseconds()); writeErr != nil {
			return qualificationProcessingSummary{}, writeErr
		}
		processed++
		if time.Since(lastResourceSample) >= time.Second {
			resources = append(resources, qualificationRuntimeSample(started))
			lastResourceSample = time.Now()
		}
	}
	actualDuration := time.Since(started)
	resources = append(resources, qualificationRuntimeSample(started))
	if err := buffered.Flush(); err != nil {
		return qualificationProcessingSummary{}, err
	}
	if err := compressed.Close(); err != nil {
		return qualificationProcessingSummary{}, err
	}
	if err := file.Close(); err != nil {
		return qualificationProcessingSummary{}, err
	}
	closed = true
	summary, err := summarizeQualificationSamples(samples)
	if err != nil {
		return qualificationProcessingSummary{}, err
	}
	sum, size, err := qualificationFileSHA256(rawPath)
	if err != nil {
		return qualificationProcessingSummary{}, err
	}
	summary.Profile = profile.Name
	summary.Codec = profile.Codec
	summary.ConfiguredBitrateKbps = profile.BitrateKbps
	summary.ObservedBitrateKbps = float64(processed*int64(profile.PacketBytes)*8) / actualDuration.Seconds() / 1000
	summary.Warmup = profile.Warmup
	summary.ConfiguredDuration = profile.Duration
	summary.ActualDuration = actualDuration
	summary.ClockOverhead = qualificationClockOverhead()
	summary.PayloadSHA256 = fmt.Sprintf("%x", sha256.Sum256(payload))
	summary.RawSamples = filepath.Base(rawPath)
	summary.RawSamplesSHA256 = sum
	summary.RawSamplesBytes = size
	resourcePath := strings.TrimSuffix(rawPath, ".csv.gz") + "-resources.csv.gz"
	if err := writeQualificationResourceSamples(resourcePath, resources); err != nil {
		return qualificationProcessingSummary{}, err
	}
	resourceSum, resourceSize, err := qualificationFileSHA256(resourcePath)
	if err != nil {
		return qualificationProcessingSummary{}, err
	}
	summary.RawResources = filepath.Base(resourcePath)
	summary.RawResourcesSHA256 = resourceSum
	summary.RawResourcesBytes = resourceSize
	summary.ResourceSamples = len(resources)
	firstResource, lastResource := resources[0], resources[len(resources)-1]
	summary.CPUSeconds = math.Max(0, lastResource.CPUSeconds-firstResource.CPUSeconds)
	summary.Mallocs = lastResource.Mallocs - firstResource.Mallocs
	summary.AllocatedBytes = lastResource.Allocated - firstResource.Allocated
	for _, resource := range resources {
		if resource.HeapBytes > summary.PeakHeapBytes {
			summary.PeakHeapBytes = resource.HeapBytes
		}
		if resource.Goroutines > summary.PeakGoroutines {
			summary.PeakGoroutines = resource.Goroutines
		}
	}
	if actualDuration < profile.Duration || actualDuration > profile.Duration*105/100+250*time.Millisecond {
		return qualificationProcessingSummary{}, fmt.Errorf("wall-clock duration %s outside [%s,%s]", actualDuration, profile.Duration, profile.Duration*105/100+250*time.Millisecond)
	}
	if summary.ObservedBitrateKbps < float64(profile.BitrateKbps)*0.95 || summary.ObservedBitrateKbps > float64(profile.BitrateKbps)*1.05 {
		return qualificationProcessingSummary{}, fmt.Errorf("observed bitrate %.2f outside profile bounds for %d", summary.ObservedBitrateKbps, profile.BitrateKbps)
	}
	if err := enforceQualificationProcessingGate(summary); err != nil {
		return qualificationProcessingSummary{}, err
	}
	return summary, nil
}

func runQualificationWarmup(t *testing.T, path *qualificationPath, profile qualificationMediaProfile, payload []byte) error {
	t.Helper()
	started := time.Now()
	for time.Since(started) < profile.Warmup {
		trace, _, err := path.traverse(t, payload)
		if err != nil {
			return err
		}
		if !trace.PayloadPreserved {
			return errors.New("qualification warmup payload integrity failure")
		}
	}
	return nil
}

func qualificationRuntimeSample(started time.Time) qualificationResourceSample {
	cpu := []runtimemetrics.Sample{{Name: "/cpu/classes/total:cpu-seconds"}}
	runtimemetrics.Read(cpu)
	var memory runtime.MemStats
	runtime.ReadMemStats(&memory)
	return qualificationResourceSample{
		Elapsed: time.Since(started), CPUSeconds: cpu[0].Value.Float64(),
		HeapBytes: memory.HeapAlloc, Goroutines: runtime.NumGoroutine(),
		Mallocs: memory.Mallocs, Allocated: memory.TotalAlloc,
	}
}

func writeQualificationResourceSamples(path string, samples []qualificationResourceSample) error {
	file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640)
	if err != nil {
		return err
	}
	compressed := gzip.NewWriter(file)
	buffered := bufio.NewWriter(compressed)
	if _, err = buffered.WriteString("elapsed_ns,cpu_seconds,heap_bytes,goroutines,mallocs,allocated_bytes\n"); err == nil {
		for _, sample := range samples {
			if _, err = fmt.Fprintf(buffered, "%d,%.9f,%d,%d,%d,%d\n", sample.Elapsed.Nanoseconds(),
				sample.CPUSeconds, sample.HeapBytes, sample.Goroutines, sample.Mallocs, sample.Allocated); err != nil {
				break
			}
		}
	}
	if flushErr := buffered.Flush(); err == nil {
		err = flushErr
	}
	if closeErr := compressed.Close(); err == nil {
		err = closeErr
	}
	if closeErr := file.Close(); err == nil {
		err = closeErr
	}
	return err
}

func qualificationClockOverhead() time.Duration {
	samples := make([]time.Duration, 10_000)
	for index := range samples {
		started := time.Now()
		samples[index] = time.Since(started)
	}
	summary, _ := summarizeQualificationSamples(samples)
	return summary.Median
}

func qualificationFileSHA256(path string) (string, int64, error) {
	file, err := os.Open(path)
	if err != nil {
		return "", 0, err
	}
	defer file.Close()
	hash := sha256.New()
	size, err := io.Copy(hash, file)
	if err != nil {
		return "", 0, err
	}
	return hex.EncodeToString(hash.Sum(nil)), size, nil
}

type qualificationFlowDelivery struct {
	at    time.Time
	flow  string
	bytes int64
}

func runQualificationFleetStage(t *testing.T, fleet *qualificationFleet, profile qualificationMediaProfile, duration time.Duration) ([]qualificationFlowDelivery, error) {
	t.Helper()
	if duration <= 0 {
		return nil, errors.New("qualification fleet duration must be positive")
	}
	end := time.Now().Add(duration)
	payload := qualificationPayload(profile)
	deliveries := make([]qualificationFlowDelivery, 0, int(duration/time.Millisecond))
	var sequence uint32
	for time.Now().Before(end) {
		for _, path := range fleet.paths {
			if !time.Now().Before(end) {
				break
			}
			current := append([]byte(nil), payload...)
			binary.BigEndian.PutUint32(current[len(current)-4:], sequence)
			sequence++
			trace, _, err := path.traverse(t, current)
			if err != nil {
				return nil, err
			}
			if !trace.NativeUDPIngress || !trace.ApolloRecovered || !trace.ProductionQueue ||
				!trace.ProductionMediaLoop || !trace.ProductionPacer || !trace.VerseQUIC ||
				!trace.PublicClientDecode || !trace.PayloadPreserved {
				return nil, errors.New("fairness traffic bypassed production gateway traversal")
			}
			deliveries = append(deliveries, qualificationFlowDelivery{
				at: time.Now(), flow: path.flow, bytes: int64(len(current) + frameHeaderSize),
			})
		}
	}
	return deliveries, nil
}

func qualificationPacerEvidence(t *testing.T, rawPath string, baselineDuration, stepDuration time.Duration) (qualificationFairnessEvidence, error) {
	t.Helper()
	flows := []string{"one", "two", "three", "four", "five", "six", "seven", "eight"}
	profile := qualificationMediaProfile{Name: "fairness-h264", Codec: "h264", BitrateKbps: 8000, PacketBytes: 1000}
	fleet := newQualificationFleet(t, len(flows), profile, 8000)
	defer fleet.Close()
	for index, path := range fleet.paths {
		path.flow = flows[index]
		if _, _, err := path.traverse(t, qualificationPayload(profile)); err != nil {
			return qualificationFairnessEvidence{}, err
		}
	}
	start := time.Now()
	baseline, err := runQualificationFleetStage(t, fleet, profile, baselineDuration)
	if err != nil {
		return qualificationFairnessEvidence{}, err
	}
	allDeliveries := append([]qualificationFlowDelivery(nil), baseline...)
	evidence := qualificationFairnessEvidence{
		Evaluation: baselineDuration, PerFlowBytes: make(map[string]int64, len(flows)),
		ShareError: make(map[string]float64, len(flows)),
	}
	var total, squares float64
	for _, delivery := range baseline {
		evidence.PerFlowBytes[delivery.flow] += delivery.bytes
	}
	for _, flow := range flows {
		total += float64(evidence.PerFlowBytes[flow])
		squares += float64(evidence.PerFlowBytes[flow]) * float64(evidence.PerFlowBytes[flow])
	}
	target := total / float64(len(flows))
	for _, flow := range flows {
		evidence.ShareError[flow] = math.Abs(float64(evidence.PerFlowBytes[flow])-target) / target
		if evidence.ShareError[flow] > 0.10 {
			return qualificationFairnessEvidence{}, fmt.Errorf("flow %s share error %.4f", flow, evidence.ShareError[flow])
		}
	}
	evidence.JainIndex = total * total / (float64(len(flows)) * squares)
	for _, step := range []struct {
		reduction int
		kbps      int64
		cap       int64
	}{
		{25, 6000, 750_000},
		{50, 4000, 500_000},
	} {
		fleet.server.pacer.setKbps(step.kbps)
		stepStart := time.Now()
		deliveries, runErr := runQualificationFleetStage(t, fleet, profile, stepDuration)
		if runErr != nil {
			return qualificationFairnessEvidence{}, runErr
		}
		allDeliveries = append(allDeliveries, deliveries...)
		convergence := qualificationPacerConvergence(deliveries, stepStart, flows, step.cap)
		maximum := qualificationMaximumFiveSecondBytes(deliveries)
		if stepDuration >= 10*time.Second && (convergence > 10*time.Second || maximum > step.cap*5*105/100) {
			return qualificationFairnessEvidence{}, fmt.Errorf("capacity step %d failed convergence=%s five-second=%d", step.reduction, convergence, maximum)
		}
		evidence.CapacitySteps = append(evidence.CapacitySteps, qualificationCapacityStep{
			ReductionPercent: step.reduction, Convergence: convergence,
			MaximumFiveSecond: maximum, FiveSecondCap: step.cap * 5,
		})
	}
	end := start.Add(baselineDuration + 2*stepDuration)
	evidence.Series = qualificationFairnessSeriesFor(allDeliveries, start, end, flows)
	if err := writeQualificationPacerSamples(rawPath, allDeliveries, start); err != nil {
		return qualificationFairnessEvidence{}, err
	}
	evidence.RawSamples = filepath.Base(rawPath)
	evidence.RawSamplesSHA256, evidence.RawSamplesBytes, err = qualificationFileSHA256(rawPath)
	if err != nil {
		return qualificationFairnessEvidence{}, err
	}
	return evidence, nil
}

func qualificationPacerConvergence(deliveries []qualificationFlowDelivery, start time.Time, flows []string, targetBytesPerSecond int64) time.Duration {
	consecutive := 0
	for second := time.Duration(0); second < 10*time.Second; second += time.Second {
		windowStart := start.Add(second)
		perFlow := make(map[string]int64, len(flows))
		var aggregate int64
		for _, delivery := range deliveries {
			if !delivery.at.Before(windowStart) && delivery.at.Before(windowStart.Add(time.Second)) {
				perFlow[delivery.flow] += delivery.bytes
				aggregate += delivery.bytes
			}
		}
		if aggregate < targetBytesPerSecond*90/100 || aggregate > targetBytesPerSecond*105/100 {
			consecutive = 0
			continue
		}
		targetFlow := targetBytesPerSecond / int64(len(flows))
		converged := true
		for _, flow := range flows {
			converged = converged && perFlow[flow] >= targetFlow*90/100 && perFlow[flow] <= targetFlow*110/100
		}
		if converged {
			consecutive++
			if consecutive == 2 {
				return second + time.Second
			}
		} else {
			consecutive = 0
		}
	}
	return 11 * time.Second
}

func qualificationMaximumFiveSecondBytes(deliveries []qualificationFlowDelivery) int64 {
	sort.Slice(deliveries, func(first, second int) bool { return deliveries[first].at.Before(deliveries[second].at) })
	var maximum, total int64
	for first, last := 0, 0; first < len(deliveries); first++ {
		for last < len(deliveries) && deliveries[last].at.Sub(deliveries[first].at) <= 5*time.Second {
			total += deliveries[last].bytes
			last++
		}
		if total > maximum {
			maximum = total
		}
		total -= deliveries[first].bytes
	}
	return maximum
}

func qualificationFairnessSeriesFor(deliveries []qualificationFlowDelivery, start, end time.Time, flows []string) []qualificationFairnessSeries {
	series := make([]qualificationFairnessSeries, 0, int(end.Sub(start)/time.Second))
	for windowStart := start; windowStart.Before(end); windowStart = windowStart.Add(time.Second) {
		sample := qualificationFairnessSeries{
			Elapsed: windowStart.Sub(start), PerFlowBytes: make(map[string]int64, len(flows)),
		}
		for _, delivery := range deliveries {
			if !delivery.at.Before(windowStart) && delivery.at.Before(windowStart.Add(time.Second)) {
				sample.PerFlowBytes[delivery.flow] += delivery.bytes
				sample.AggregateBytes += delivery.bytes
			}
		}
		series = append(series, sample)
	}
	return series
}

func writeQualificationPacerSamples(path string, deliveries []qualificationFlowDelivery, start time.Time) error {
	file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640)
	if err != nil {
		return err
	}
	compressed := gzip.NewWriter(file)
	buffered := bufio.NewWriter(compressed)
	if _, err = buffered.WriteString("elapsed_ns,flow,bytes\n"); err == nil {
		for _, delivery := range deliveries {
			if _, err = fmt.Fprintf(buffered, "%d,%s,%d\n", delivery.at.Sub(start).Nanoseconds(), delivery.flow, delivery.bytes); err != nil {
				break
			}
		}
	}
	if flushErr := buffered.Flush(); err == nil {
		err = flushErr
	}
	if closeErr := compressed.Close(); err == nil {
		err = closeErr
	}
	if closeErr := file.Close(); err == nil {
		err = closeErr
	}
	return err
}

func qualificationProtocolVersion() (string, error) {
	version := os.Getenv("VERSEVDI_QUALIFICATION_PROTOCOL_VERSION")
	valid, err := regexp.MatchString(`^v[0-9]+\.[0-9]+\.[0-9]+-[0-9A-Za-z]+(?:[.-][0-9A-Za-z]+)*$`, version)
	if err != nil || !valid {
		return "", errors.New("VERSEVDI_QUALIFICATION_PROTOCOL_VERSION must be an immutable prerelease tag")
	}
	return version, nil
}

func qualificationCandidateCommit() (string, error) {
	commit := os.Getenv("VERSEVDI_QUALIFICATION_COMMIT")
	decoded, err := hex.DecodeString(commit)
	if err != nil || len(decoded) != 20 || strings.ToLower(commit) != commit {
		return "", errors.New("VERSEVDI_QUALIFICATION_COMMIT must be a lowercase full SHA-1")
	}
	return commit, nil
}

func qualificationToolVersions() (map[string]string, error) {
	versions := map[string]string{"qualification": qualificationToolVersion, "go": runtime.Version()}
	info, ok := debug.ReadBuildInfo()
	if ok {
		for _, dependency := range info.Deps {
			if dependency.Path == "github.com/quic-go/quic-go" {
				versions["quic-go"] = dependency.Version
				break
			}
		}
	}
	if versions["quic-go"] == "" {
		module, err := os.ReadFile(filepath.Join("..", "go.mod"))
		if err != nil {
			return nil, errors.New("qualification QUIC implementation version unavailable")
		}
		match := regexp.MustCompile(`(?m)^\s*github\.com/quic-go/quic-go\s+(v[^\s]+)`).FindSubmatch(module)
		if len(match) != 2 {
			return nil, errors.New("qualification QUIC implementation version unavailable")
		}
		versions["quic-go"] = string(match[1])
	}
	return versions, nil
}

func writeQualificationJSON(path string, value any) error {
	file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640)
	if err != nil {
		return err
	}
	encoder := json.NewEncoder(file)
	encoder.SetIndent("", "  ")
	if err := encoder.Encode(value); err != nil {
		_ = file.Close()
		return err
	}
	return file.Close()
}

func TestSection7Qualification(t *testing.T) {
	output := os.Getenv("VERSEVDI_QUALIFICATION_DIR")
	if output == "" {
		t.Skip("set VERSEVDI_QUALIFICATION_DIR to run the 30-minute frozen-candidate qualification")
	}
	if err := validateQualificationOutputDir(output); err != nil {
		t.Fatal(err)
	}
	commit, err := qualificationCandidateCommit()
	if err != nil {
		t.Fatal(err)
	}
	protocolVersion, err := qualificationProtocolVersion()
	if err != nil {
		t.Fatal(err)
	}
	toolVersions, err := qualificationToolVersions()
	if err != nil {
		t.Fatal(err)
	}
	if err := os.Mkdir(output, 0o750); err != nil {
		t.Fatalf("create new qualification evidence directory: %v", err)
	}
	started := time.Now().UTC()
	media := qualificationMediaProfiles()
	qualificationTraverseProfiles(t, media)
	manifest := qualificationManifest{
		Status: "running", ToolVersion: qualificationToolVersion,
		Command: fmt.Sprintf(
			"VERSEVDI_QUALIFICATION_DIR=%s VERSEVDI_QUALIFICATION_COMMIT=%s VERSEVDI_QUALIFICATION_PROTOCOL_VERSION=%s GOWORK=off go test ./gateway -run '^TestSection7Qualification$' -count=1 -timeout 45m -v",
			output, commit, protocolVersion,
		),
		CandidateCommit: commit, ProtocolVersion: protocolVersion,
		StartedAt: started.Format(time.RFC3339Nano), GoVersion: runtime.Version(),
		ToolVersions: toolVersions,
		OS:           runtime.GOOS, Architecture: runtime.GOARCH,
		Topology:        "source-shaped encrypted Apollo fixture -> native recovery/FEC -> bounded provider queue -> production fair pacer -> Verse framing over mTLS/QUIC -> public Verse client decoder",
		Direction:       "provider_to_client",
		QueueDiscipline: "bounded 16-packet native provider queue, production equal-tier fair pacer, deterministic fixed-seed source impairment",
		Evidence:        []string{"deterministic source-shaped Apollo recovery", "local real-time production path", "mTLS/QUIC fixture transport", "path impairment", "production fair pacer"},
		Deferred:        []string{"live Apollo", "macOS client", "physical firewall and packet route", "real encoder fidelity", "multi-host scale"},
	}
	for _, profile := range media {
		raw := filepath.Join(output, "processing-"+profile.Name+".csv.gz")
		summary, runErr := runQualificationProcessing(t, profile, raw)
		if runErr != nil {
			t.Fatal(runErr)
		}
		manifest.Processing = append(manifest.Processing, summary)
		t.Logf("%s count=%d p95=%s observed=%.2f kbps", profile.Name, summary.Count, summary.P95, summary.ObservedBitrateKbps)
	}
	fairness, err := qualificationPacerEvidence(t, filepath.Join(output, "fairness.csv.gz"), 60*time.Second, 10*time.Second)
	if err != nil {
		t.Fatal(err)
	}
	manifest.Fairness = fairness
	for _, impairment := range qualificationImpairmentProfiles() {
		profiles := media[:1]
		if impairment.Name == "baseline" {
			profiles = media
		}
		for _, profile := range profiles {
			raw := filepath.Join(output, "impairment-"+impairment.Name+"-"+profile.Name+".csv.gz")
			observation, runErr := runQualificationImpairment(t, impairment, profile, qualificationImpairmentPacketCount, raw)
			if runErr != nil {
				t.Fatal(runErr)
			}
			manifest.Impairments = append(manifest.Impairments, observation)
		}
	}
	manifest.Status = "passed"
	manifest.CompletedAt = time.Now().UTC().Format(time.RFC3339Nano)
	manifestPath := filepath.Join(output, "manifest.json")
	if err := writeQualificationJSON(manifestPath, manifest); err != nil {
		t.Fatal(err)
	}
	manifestBytes, err := os.ReadFile(manifestPath)
	if err != nil {
		t.Fatal(err)
	}
	for _, forbidden := range []string{"private_key", "client_private", "clipboard text", "apollo.test", "provider endpoint", "live interoperability passed"} {
		if bytes.Contains(bytes.ToLower(manifestBytes), []byte(forbidden)) {
			t.Fatalf("qualification manifest contains forbidden boundary text %q", forbidden)
		}
	}
	t.Logf("qualification manifest: %s", manifestPath)
}

func qualificationTraverseProfiles(t *testing.T, profiles []qualificationMediaProfile) {
	t.Helper()
	for _, profile := range profiles {
		trace := qualificationProductionPathSmoke(t, profile)
		if !trace.NativeSetup || !trace.NativeOpen || !trace.NativeUDPIngress || !trace.ApolloRecovered ||
			!trace.ProductionQueue || !trace.ProductionMediaLoop || !trace.ProductionPacer ||
			!trace.VerseQUIC || !trace.PublicClientDecode || !trace.PayloadPreserved {
			t.Fatalf("%s production path: %#v", profile.Name, trace)
		}
	}
}