test(gateway): add Section 7 qualification harness

gateway/qualification_contract_test.go

@@ -0,0 +1,152 @@
+package gateway
+
+import (
+	"compress/gzip"
+	"io"
+	"os"
+	"path/filepath"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestQualificationCatalogMatchesSection7(t *testing.T) {
+	media := qualificationMediaProfiles()
+	if len(media) != 3 {
+		t.Fatalf("media profile count = %d, want 3", len(media))
+	}
+	wantMedia := []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},
+	}
+	if !reflect.DeepEqual(media, wantMedia) {
+		t.Fatalf("media profiles = %#v, want %#v", media, wantMedia)
+	}
+
+	impairments := qualificationImpairmentProfiles()
+	wantImpairments := []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}},
+	}
+	if !reflect.DeepEqual(impairments, wantImpairments) {
+		t.Fatalf("impairment profiles = %#v, want %#v", impairments, wantImpairments)
+	}
+}
+
+func TestQualificationOutputAndStatisticsFailClosed(t *testing.T) {
+	if err := validateQualificationOutputDir("relative/evidence"); err == nil {
+		t.Fatal("relative evidence directory was accepted")
+	}
+	if err := validateQualificationOutputDir(filepath.Join(t.TempDir(), "evidence")); err != nil {
+		t.Fatalf("absolute evidence directory rejected: %v", err)
+	}
+
+	summary, err := summarizeQualificationSamples([]time.Duration{
+		time.Millisecond, 2 * time.Millisecond, 3 * time.Millisecond,
+		4 * time.Millisecond, 5 * time.Millisecond,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if summary.Count != 5 || summary.Min != time.Millisecond || summary.Median != 3*time.Millisecond ||
+		summary.P90 != 5*time.Millisecond || summary.P95 != 5*time.Millisecond ||
+		summary.P99 != 5*time.Millisecond || summary.Max != 5*time.Millisecond ||
+		summary.Mean != 3*time.Millisecond || summary.Histogram["le_5ms"] != 5 {
+		t.Fatalf("summary = %#v", summary)
+	}
+	if err := enforceQualificationProcessingGate(summary); err != nil {
+		t.Fatalf("5 ms p95 rejected: %v", err)
+	}
+	summary.P95++
+	if err := enforceQualificationProcessingGate(summary); err == nil {
+		t.Fatal("p95 above 5 ms was accepted")
+	}
+}
+
+func TestQualificationShortProcessingWritesRawArtifact(t *testing.T) {
+	profile := qualificationMediaProfile{
+		Name: "smoke", Codec: "h264", BitrateKbps: 1000,
+		Duration: 200 * time.Millisecond, Warmup: time.Millisecond, PacketBytes: 100,
+	}
+	rawPath := filepath.Join(t.TempDir(), "processing.csv.gz")
+	summary, err := runQualificationProcessing(profile, rawPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if summary.Count < 1 || summary.RawSamplesSHA256 == "" || summary.RawSamplesBytes < 1 {
+		t.Fatalf("processing summary = %#v", summary)
+	}
+	file, err := os.Open(rawPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer file.Close()
+	reader, err := gzip.NewReader(file)
+	if err != nil {
+		t.Fatal(err)
+	}
+	raw, err := io.ReadAll(reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := reader.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if !strings.HasPrefix(string(raw), "elapsed_ns,processing_ns\n") || strings.Count(string(raw), "\n") != int(summary.Count)+1 {
+		t.Fatalf("raw sample rows do not match summary count: %q", raw)
+	}
+}
+
+func TestQualificationProcessingPreservesPayload(t *testing.T) {
+	payload := qualificationPayload(qualificationMediaProfiles()[0])
+	processed, elapsed, err := processQualificationPayload(7, payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(processed, payload) {
+		t.Fatal("encoded payload mutated")
+	}
+	if elapsed <= 0 {
+		t.Fatalf("processing duration = %s", elapsed)
+	}
+}
+
+func TestQualificationImpairmentIsDeterministicAndBounded(t *testing.T) {
+	profile := qualificationImpairmentProfiles()[3]
+	first, err := runQualificationImpairment(profile, qualificationMediaProfiles()[0], 10_000)
+	if err != nil {
+		t.Fatal(err)
+	}
+	second, err := runQualificationImpairment(profile, qualificationMediaProfiles()[0], 10_000)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(first, second) {
+		t.Fatalf("impairment run is not deterministic:\n%#v\n%#v", first, second)
+	}
+	if first.Sent != 10_000 || first.Delivered+first.Dropped != first.Sent ||
+		first.ObservedLossPercent < 4.8 || first.ObservedLossPercent > 5.2 ||
+		first.MaxQueuePackets > qualificationImpairmentQueuePackets {
+		t.Fatalf("impairment observation = %#v", first)
+	}
+	if _, err := runQualificationImpairment(profile, qualificationMediaProfiles()[0], qualificationImpairmentMaxPackets+1); err == nil {
+		t.Fatal("unbounded impairment packet count was accepted")
+	}
+}
+
+func TestQualificationUsesPublicQUICAndProductionPacer(t *testing.T) {
+	qualificationTraverseProfiles(t, qualificationMediaProfiles())
+	evidence, err := qualificationPacerEvidence()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(evidence.PerFlowBytes) != 8 || len(evidence.CapacitySteps) != 2 || evidence.JainIndex < 0.99 {
+		t.Fatalf("pacer evidence = %#v", evidence)
+	}
+}

gateway/qualification_harness_test.go

@@ -0,0 +1,723 @@
+package gateway
+
+import (
+	"bufio"
+	"bytes"
+	"compress/gzip"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"math"
+	"os"
+	"path/filepath"
+	"runtime"
+	"runtime/debug"
+	"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 {
+	info, ok := debug.ReadBuildInfo()
+	if !ok {
+		return "unknown"
+	}
+	for _, dependency := range info.Deps {
+		if dependency.Path == "git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol" {
+			return dependency.Version
+		}
+	}
+	return "unknown"
+}
+
+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)
+	}
+	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 GOWORK=off go test ./gateway -run '^TestSection7Qualification$' -count=1 -timeout 45m -v", output, commit),
+		CandidateCommit: commit, ProtocolVersion: qualificationProtocolVersion(),
+		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)
+}

openspec/changes/phase3c-gateway-qualification/.openspec.yaml

@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-07-29

openspec/changes/phase3c-gateway-qualification/design.md

@@ -0,0 +1,54 @@
+## Context
+
+P3C-029 through P3C-033 require raw, reproducible processing, payload-rate,
+fairness, cap, and impairment evidence. Existing focused tests cover the
+framer, bounded queues, native Apollo fake, and fair pacer, but do not emit the
+normative ten-minute or six-profile artifacts.
+
+## Goals / Non-Goals
+
+**Goals:**
+
+- Run the three 20/50/80 Mbps encoded-payload profiles for ten wall-clock
+  minutes each after a recorded warm-up.
+- Measure the existing gateway framing path with a monotonic clock and retain
+  compressed raw latency samples plus full summary statistics.
+- Run the exact six Section 7.2 configurations once using a deterministic,
+  bounded virtual packet discipline and retain configured and observed values.
+- Exercise one real mTLS/QUIC fake-provider traversal for every media profile
+  and reuse the real fair-pacer implementation for fairness and cap evidence.
+
+**Non-Goals:**
+
+- Live Apollo/macOS/firewall qualification, real encoder fidelity, codec
+  processing, host network mutation, or multi-host scale.
+- A production impairment framework, new gateway API, dependency, cgo, or
+  sidecar.
+
+## Decisions
+
+- Implement the harness as an opt-in `go test` in package `gateway`. This
+  keeps qualification access to the actual unexported fair pacer without
+  adding a production API. Normal suites skip the long run unless an explicit
+  absolute evidence directory is supplied.
+- Use wall-clock duration and target-rate pacing for performance profiles.
+  Measure only receive-to-framed-payload processing; pacing wait and raw-file
+  writes stay outside the measured interval and are reported separately.
+- Stream every raw sample into gzip-compressed CSV while retaining one bounded
+  duration slice per profile for exact percentiles.
+- Use a fixed-seed virtual FIFO for impairment. It records no host claim and
+  identifies its queue discipline and deterministic topology explicitly.
+- Treat any payload mutation, p95 above 5 ms, catalog mismatch, fairness error
+  above 10%, cap excess above 5%, or step convergence beyond ten seconds as a
+  hard command failure.
+
+## Risks / Trade-offs
+
+- [Local virtual impairment cannot prove deployed route behavior] → label every
+  artifact deterministic and retain live Apollo/macOS/firewall as
+  deferred-owner-e2e.
+- [Raw samples can be large] → stream gzip output and bound in-memory samples
+  to the exact profile packet budget.
+- [Host load can invalidate latency] → record OS, architecture, Go version,
+  timing overhead, actual duration, packet count, and observed bitrate; fail
+  rather than substitute configured capacity for measured egress.

openspec/changes/phase3c-gateway-qualification/proposal.md

@@ -0,0 +1,35 @@
+## Why
+
+The Phase 3C gateway candidate has deterministic transport and scheduler tests
+but no executable artifact generator for the normative ten-minute media
+measurements and six bounded impairment profiles. Without that evidence,
+P3C-029 through P3C-033 cannot be frozen truthfully.
+
+## What Changes
+
+- Add one stdlib-only qualification command for the three fixed encoded-media
+  profiles and the exact six Section 7.2 impairment profiles.
+- Emit bounded machine-readable configuration, raw observations, summaries,
+  environment, topology, direction, queue discipline, and tool version.
+- Fail the command when payload integrity, the 5 ms processing p95, impairment
+  bounds, fairness, capacity-step convergence, or aggregate cap gates fail.
+- Keep live Apollo, macOS, physical firewall, real encoder fidelity, and real
+  multi-host scale explicitly deferred-owner-e2e.
+
+## Capabilities
+
+### New Capabilities
+
+- `gateway-qualification`: Deterministic P3C-029 through P3C-033 media,
+  processing, fairness, cap, and impairment evidence generation.
+
+### Modified Capabilities
+
+None.
+
+## Impact
+
+The Data Plane gains a qualification-only Go command, focused tests, and
+documented evidence output. It adds no dependency, production transport
+abstraction, provider route, codec operation, cgo, sidecar, or Connection
+Server code.

openspec/changes/phase3c-gateway-qualification/specs/gateway-qualification/spec.md

@@ -0,0 +1,63 @@
+## ADDED Requirements
+
+### Requirement: Fixed media processing qualification
+The qualification harness SHALL run 1080p60 H.264 at 20 Mbps, 1440p120 HEVC
+at 50 Mbps, and 4K60 HEVC at 80 Mbps for ten wall-clock minutes each after a
+recorded warm-up. It SHALL preserve encoded payload bytes, record every
+monotonic processing sample, report count, min, median, p90, p95, p99, max,
+mean, standard deviation, timing overhead, and observed bitrate, and fail when
+any p95 exceeds 5 ms.
+
+#### Scenario: Healthy fixed profile
+- **WHEN** a frozen candidate runs one fixed profile for the normative duration
+- **THEN** the harness emits compressed raw samples and a summary tied to the
+  exact source commit, Protocol version, environment, and payload hash.
+
+#### Scenario: Processing gate failure
+- **WHEN** payload integrity fails or measured p95 exceeds 5 ms
+- **THEN** the qualification command exits unsuccessfully without recording a
+  passing candidate.
+
+### Requirement: Bounded impairment qualification
+The harness SHALL run exactly the baseline, latency, jitter, loss, reorder,
+and constrained Section 7.2 profiles once. Baseline SHALL cover all three
+media profiles and the other profiles SHALL cover 1080p60. Each artifact SHALL
+record tool version, exact command/configuration, direction, queue discipline,
+topology, fixed seed, and observed RTT, jitter, loss, reorder, throughput,
+drops, and capacity-step statistics.
+
+#### Scenario: Complete six-profile run
+- **WHEN** the frozen candidate runs impairment qualification
+- **THEN** one result exists for each named profile, with no Cartesian
+  expansion and with observed rather than configured statistics.
+
+#### Scenario: Unsupported or unbounded configuration
+- **WHEN** a profile name, packet count, queue bound, loss, reorder, or
+  bandwidth step falls outside the fixed catalog
+- **THEN** the harness rejects it before allocating or running the simulation.
+
+### Requirement: Fairness and cap qualification
+The harness SHALL exercise the production fair pacer with eight equal-tier
+synthetic sessions for the required 60-second virtual interval, report every
+share error and Jain's fairness index, and fail above 10% share error. It SHALL
+apply 25% and 50% capacity steps, fail convergence beyond ten virtual seconds,
+and fail aggregate egress above 105% of the cap over any rolling five-second
+window.
+
+#### Scenario: Equal-tier and capacity-step evidence
+- **WHEN** the frozen candidate runs scheduler qualification
+- **THEN** the artifact contains per-flow bytes, share errors, Jain's index,
+  step convergence, and rolling cap observations derived from the production
+  pacer.
+
+### Requirement: Honest qualification boundary
+Qualification artifacts SHALL contain no provider endpoint, credential,
+clipboard text, input payload, secret, raw media content, or claim of live
+Apollo/macOS/firewall interoperability. The harness SHALL add no codec
+operation, production dependency, cgo, sidecar, or direct provider route.
+
+#### Scenario: Deterministic evidence publication
+- **WHEN** qualification completes
+- **THEN** the manifest labels fake-provider, virtual impairment, and local
+  processing evidence separately and leaves live interoperability
+  deferred-owner-e2e.

openspec/changes/phase3c-gateway-qualification/tasks.md

@@ -0,0 +1,24 @@
+## 1. Contract and focused regressions
+
+- [x] 1.1 Add fixed catalog tests for the three media profiles, ten-minute
+  duration, exact six impairment profiles, and bounded output configuration.
+- [x] 1.2 Add summary, payload-integrity, fairness, cap, and failure-threshold
+  tests before implementing the harness.
+
+## 2. Qualification harness
+
+- [x] 2.1 Implement opt-in real-duration processing measurement with compressed
+  raw samples, full statistics, timing overhead, and environment metadata.
+- [x] 2.2 Implement deterministic bounded impairment observations and reuse the
+  production fair pacer for fairness and capacity-step evidence.
+- [x] 2.3 Add one mTLS/QUIC fake-provider traversal per fixed encoded profile
+  and prove the artifact boundary contains no provider route or secret.
+
+## 3. Freeze and evidence
+
+- [x] 3.1 Run focused red/green checks, strict OpenSpec validation, `make
+  verify`, race/fuzz/resource checks, and freeze the harness commit.
+- [ ] 3.2 Run the opt-in ten-minute and six-profile command exactly once
+  against the frozen candidate and archive raw artifacts and hashes.
+- [ ] 3.3 Sync the canonical specification, archive the completed change, and
+  revalidate strictly without claiming live Apollo/macOS/firewall evidence.