VerseVDI-Data-Plane/gateway/qualification_harness_test.go

package gateway

import (
	"bufio"
	"bytes"
	"compress/gzip"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math"
	"os"
	"path/filepath"
	"regexp"
	"runtime"
	"sort"
	"strings"
	"testing"
	"time"
)

const (
	qualificationToolVersion                   = "versevdi-gateway-qualification/v1"
	qualificationImpairmentQueuePackets        = 64
	qualificationImpairmentMaxPackets          = 100_000
	qualificationImpairmentPacketCount         = 10_000
	qualificationMaximumCatchupPackets         = 16_384
	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 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"`
}

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"`
}

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 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"`
}

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"`
	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
}

func processQualificationPayload(sequence uint32, payload []byte) ([]byte, time.Duration, error) {
	started := time.Now()
	frames, err := FragmentPayload(ChannelVideo, sequence, uint64(started.UnixMilli()), payload)
	if err != nil {
		return nil, 0, err
	}
	recovered := make([]byte, 0, len(payload))
	for _, frame := range frames {
		encoded, encodeErr := EncodeFrame(frame)
		if encodeErr != nil {
			return nil, 0, encodeErr
		}
		decoded, decodeErr := DecodeFrame(encoded)
		if decodeErr != nil {
			return nil, 0, decodeErr
		}
		recovered = append(recovered, decoded.Payload...)
	}
	elapsed := time.Since(started)
	if !bytes.Equal(recovered, payload) {
		return nil, elapsed, errors.New("qualification payload integrity failure")
	}
	return recovered, elapsed, nil
}

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(profile qualificationImpairmentProfile, media qualificationMediaProfile, packetCount int) (qualificationImpairmentObservation, error) {
	if packetCount < 1 || packetCount > qualificationImpairmentMaxPackets || media.PacketBytes < 1 || media.PacketBytes > 1179 {
		return qualificationImpairmentObservation{}, errors.New("qualification impairment bounds invalid")
	}
	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...),
	}
	payload := qualificationPayload(media)
	var nextService, virtualEnd, previousArrival time.Duration
	var totalRTT, totalJitter time.Duration
	for index := 0; index < packetCount; index++ {
		sentAt := time.Duration(index) * spacing
		jitter := time.Duration(0)
		if profile.Jitter > 0 {
			width := uint64(profile.Jitter*2 + 1)
			jitter = time.Duration(random()%width) - profile.Jitter
		}
		arrival := sentAt + profile.RTT + jitter
		if profile.Reorder && index%20 == 19 {
			arrival = previousArrival - spacing
			observation.InjectedReordered++
		}
		if index > 0 && arrival < previousArrival {
			observation.ObservedOutOfOrder++
		}
		previousArrival = arrival
		totalRTT += arrival - sentAt
		if jitter < 0 {
			totalJitter -= jitter
		} else {
			totalJitter += jitter
		}
		if float64(random()%10_000) < profile.LossPercent*100 {
			observation.Dropped++
			continue
		}
		capacityPercent := 100
		if len(profile.CapacitySteps) == 2 {
			if index < packetCount/2 {
				capacityPercent -= profile.CapacitySteps[0]
			} else {
				capacityPercent -= profile.CapacitySteps[1]
			}
		}
		serviceInterval := spacing * 100 / time.Duration(capacityPercent)
		queuePackets := 0
		if nextService > arrival {
			queuePackets = int((nextService - arrival + serviceInterval - 1) / serviceInterval)
		}
		if queuePackets >= qualificationImpairmentQueuePackets {
			observation.Dropped++
			continue
		}
		if queuePackets > observation.MaxQueuePackets {
			observation.MaxQueuePackets = queuePackets
		}
		if arrival > nextService {
			nextService = arrival
		}
		nextService += serviceInterval
		virtualEnd = nextService
		if _, _, err := processQualificationPayload(uint32(index), payload); err != nil {
			return qualificationImpairmentObservation{}, err
		}
		observation.Delivered++
	}
	observation.ObservedRTT = totalRTT / time.Duration(packetCount)
	observation.ObservedJitter = totalJitter / time.Duration(packetCount)
	observation.ObservedLossPercent = float64(observation.Dropped) * 100 / float64(packetCount)
	observation.ObservedReorderPercent = float64(observation.ObservedOutOfOrder) * 100 / float64(packetCount)
	if virtualEnd > 0 {
		observation.ObservedThroughputKbps = float64(observation.Delivered*media.PacketBytes*8) / virtualEnd.Seconds() / 1000
	}
	if observation.Delivered+observation.Dropped != observation.Sent || observation.MaxQueuePackets > qualificationImpairmentQueuePackets {
		return qualificationImpairmentObservation{}, errors.New("qualification impairment accounting invalid")
	}
	return observation, nil
}

func runQualificationProcessing(profile qualificationMediaProfile, rawPath string) (qualificationProcessingSummary, error) {
	payload := qualificationPayload(profile)
	if err := runQualificationWarmup(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
	targetPackets := bytesPerSecond * profile.Duration.Nanoseconds() / int64(time.Second) / int64(profile.PacketBytes)
	samples := make([]time.Duration, 0, int(targetPackets))
	started := time.Now()
	ticker := time.NewTicker(time.Millisecond)
	defer ticker.Stop()
	var processed int64
	for processed < targetPackets {
		elapsed := time.Since(started)
		expected := targetPackets
		if elapsed < profile.Duration {
			expected = targetPackets * elapsed.Nanoseconds() / profile.Duration.Nanoseconds()
		}
		if backlog := expected - processed; backlog > qualificationMaximumCatchupPackets {
			return qualificationProcessingSummary{}, fmt.Errorf("qualification host fell behind by %d packets", backlog)
		}
		for processed < expected {
			_, sample, processErr := processQualificationPayload(uint32(processed), payload)
			if processErr != nil {
				return qualificationProcessingSummary{}, processErr
			}
			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 processed < targetPackets {
			<-ticker.C
		}
	}
	actualDuration := time.Since(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
	if summary.ObservedBitrateKbps < float64(profile.BitrateKbps)*0.95 {
		return qualificationProcessingSummary{}, fmt.Errorf("observed bitrate %.2f below profile %d", summary.ObservedBitrateKbps, profile.BitrateKbps)
	}
	if err := enforceQualificationProcessingGate(summary); err != nil {
		return qualificationProcessingSummary{}, err
	}
	return summary, nil
}

func runQualificationWarmup(profile qualificationMediaProfile, payload []byte) error {
	started := time.Now()
	var sequence uint32
	for time.Since(started) < profile.Warmup {
		if _, _, err := processQualificationPayload(sequence, payload); err != nil {
			return err
		}
		sequence++
	}
	return nil
}

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
}

func qualificationPacerEvidence() (qualificationFairnessEvidence, error) {
	start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
	flows := []string{"one", "two", "three", "four", "five", "six", "seven", "eight"}
	pacer := newFairPacer(8000)
	next := make(map[string]time.Time, len(flows))
	baseline := runSyntheticPacer(pacer, start, start.Add(60*time.Second), flows, next)
	evidence := qualificationFairnessEvidence{
		Evaluation: 60 * time.Second, 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
		start     time.Time
		end       time.Time
		cap       int64
	}{
		{25, 6000, start.Add(60 * time.Second), start.Add(70 * time.Second), 750_000},
		{50, 4000, start.Add(70 * time.Second), start.Add(80 * time.Second), 500_000},
	} {
		pacer.setKbps(step.kbps)
		deliveries := runSyntheticPacer(pacer, step.start, step.end, flows, next)
		convergence := qualificationPacerConvergence(deliveries, step.start, flows)
		maximum := qualificationMaximumFiveSecondBytes(deliveries)
		if 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,
		})
	}
	return evidence, nil
}

func qualificationPacerConvergence(deliveries []syntheticPacerDelivery, start time.Time, flows []string) time.Duration {
	first := make(map[string]time.Time, len(flows))
	for _, delivery := range deliveries {
		if first[delivery.flow].IsZero() {
			first[delivery.flow] = delivery.at
		}
	}
	var convergence time.Duration
	for _, flow := range flows {
		if first[flow].IsZero() {
			return 11 * time.Second
		}
		if delay := first[flow].Sub(start); delay > convergence {
			convergence = delay
		}
	}
	return convergence
}

func qualificationMaximumFiveSecondBytes(deliveries []syntheticPacerDelivery) 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 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 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)
	}
	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(),
		OS: runtime.GOOS, Architecture: runtime.GOARCH,
		Topology:        "bounded fixture provider -> gateway framing and mTLS/QUIC transport -> fixture client",
		Direction:       "provider_to_client",
		QueueDiscipline: "deterministic virtual FIFO, 64 packets, fixed seed",
		Evidence:        []string{"local real-time processing", "mTLS/QUIC fixture transport", "virtual impairment", "production fair pacer", "source-shaped Apollo covered by separate frozen test"},
		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(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()
	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 {
			observation, runErr := runQualificationImpairment(impairment, profile, qualificationImpairmentPacketCount)
			if runErr != nil {
				t.Fatal(runErr)
			}
			if impairment.Name == "constrained" {
				observation.CapacityStepObservations = append([]qualificationCapacityStep(nil), fairness.CapacitySteps...)
			}
			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()
	harness := newGatewayTransportHarness(t)
	for _, profile := range profiles {
		payload := qualificationPayload(profile)
		harness.session.EmitVideo(payload)
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		for {
			frame, err := harness.client.ReceiveFrame(ctx)
			if err != nil {
				cancel()
				t.Fatalf("%s fixture transport: %v", profile.Name, err)
			}
			if frame.Channel == ChannelVideo && bytes.Equal(frame.Payload, payload) {
				break
			}
		}
		cancel()
	}
	_ = harness.client.Close()
	harness.waitReleased(t)
}