VerseVDI-Data-Plane/gateway/qualification_contract_test.go

package gateway

import (
	"bufio"
	"bytes"
	"compress/gzip"
	"context"
	"crypto/cipher"
	"encoding/binary"
	"encoding/csv"
	"errors"
	"fmt"
	"io"
	"net"
	"os"
	"path/filepath"
	"reflect"
	"runtime"
	"sort"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"
)

type qualificationBlockingAEAD struct {
	cipher.AEAD
	ctx     context.Context
	blocked chan struct{}
	release chan struct{}
	once    sync.Once
	waitErr error
}

func (a *qualificationBlockingAEAD) Open(dst, nonce, ciphertext, additionalData []byte) ([]byte, error) {
	a.once.Do(func() {
		close(a.blocked)
		select {
		case <-a.release:
		case <-a.ctx.Done():
			a.waitErr = a.ctx.Err()
		}
	})
	if a.waitErr != nil {
		return nil, a.waitErr
	}
	return a.AEAD.Open(dst, nonce, ciphertext, additionalData)
}

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, FPS: 60, Duration: 10 * time.Minute, Warmup: time.Second, PacketBytes: 1179},
		{Name: "1440p120-hevc", Codec: "hevc", BitrateKbps: 50000, FPS: 120, Duration: 10 * time.Minute, Warmup: time.Second, PacketBytes: 1179},
		{Name: "4k60-hevc", Codec: "hevc", BitrateKbps: 80000, FPS: 60, 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 TestQualificationProtocolVersionIsExplicitAndImmutable(t *testing.T) {
	const version = "v1.0.0-phase3c-gateway-rc.8"
	t.Setenv("VERSEVDI_QUALIFICATION_PROTOCOL_VERSION", version)
	got, err := qualificationProtocolVersion()
	if err != nil || got != version {
		t.Fatalf("qualificationProtocolVersion() = %q, want %q", got, version)
	}
	for _, invalid := range []string{"", "unknown", "v1", " v1.0.0", "v1.0.0+mutable"} {
		t.Setenv("VERSEVDI_QUALIFICATION_PROTOCOL_VERSION", invalid)
		if _, err := qualificationProtocolVersion(); err == nil {
			t.Fatalf("qualificationProtocolVersion() accepted %q", invalid)
		}
	}
}

func TestQualificationRecordsLinkedToolVersions(t *testing.T) {
	versions, err := qualificationToolVersions()
	if err != nil || versions["qualification"] != qualificationToolVersion ||
		versions["go"] == "" || versions["quic-go"] == "" {
		t.Fatalf("qualification tool versions = %#v, %v", versions, err)
	}
}

func TestQualificationApolloFixturePacesSourceShapedVideo(t *testing.T) {
	key := bytes.Repeat([]byte{0x3c}, 16)
	encoded := make([]byte, 1000*apolloVideoShardPayloadSize-8)
	packets := qualificationSourceVideoPackets(t, key, 1, encoded)
	if len(packets) != 1000 || len(packets[0]) != 1072 {
		t.Fatalf("source vector = %d packets of %d bytes, want 1000 packets of 1072 bytes", len(packets), len(packets[0]))
	}
	packetsPerMillisecond, batchSize := qualificationApolloVideoPacing(apolloVideoRawPacketSize)
	if apolloVideoRawPacketSize != 1040 || packetsPerMillisecond != 96 || batchSize != 63 {
		t.Fatalf("Apollo raw pacing vector = %d bytes, %d packets/ms, batch %d; want 1040, 96, and 63", apolloVideoRawPacketSize, packetsPerMillisecond, batchSize)
	}
	wantOffsets := []time.Duration{0, 656250 * time.Nanosecond, 1312500 * time.Nanosecond, 10416666 * time.Nanosecond}
	for index, sent := range []int{0, 63, 126, 1000} {
		if got := qualificationApolloVideoOffset(sent, packetsPerMillisecond); got != wantOffsets[index] {
			t.Fatalf("Apollo pacing offset after %d packets = %s, want %s", sent, got, wantOffsets[index])
		}
	}

	receiver, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
	if err != nil {
		t.Fatal(err)
	}
	defer receiver.Close()
	sender, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
	if err != nil {
		t.Fatal(err)
	}
	defer sender.Close()

	fixture := &qualificationApolloFixture{video: sender, failures: make(chan error, 1)}
	remote := *receiver.LocalAddr().(*net.UDPAddr)
	fixture.videoRemote.Store(&remote)
	var batchIndices []int
	var batchStarts []time.Duration
	var started time.Time
	fixture.observeVideoBatch = func(index int, at time.Time) {
		if len(batchIndices) < 3 {
			batchIndices = append(batchIndices, index)
			batchStarts = append(batchStarts, at.Sub(started))
		}
	}
	drained := make(chan struct{})
	go func() {
		defer close(drained)
		buffer := make([]byte, 2048)
		for range len(packets) + 1 {
			if _, _, readErr := receiver.ReadFromUDP(buffer); readErr != nil {
				return
			}
		}
	}()

	started = time.Now()
	if err := fixture.sendVideo(context.Background(), 1, packets); err != nil {
		t.Fatal(err)
	}
	if err := fixture.sendVideo(context.Background(), 2, packets[:1]); err != nil {
		t.Fatal(err)
	}
	elapsed := time.Since(started)
	wantCarry := qualificationApolloVideoOffset(len(packets), packetsPerMillisecond)
	if wantCarry != 10416666*time.Nanosecond {
		t.Fatalf("1000-packet carry = %s, want 10.416666 ms", wantCarry)
	}
	if elapsed < wantCarry {
		t.Fatalf("source fixture sent the next frame after %s, before Apollo pacing carry %s", elapsed, wantCarry)
	}
	wantIndices := []int{0, 63, 126}
	wantStarts := []time.Duration{0, 656250 * time.Nanosecond, 1312500 * time.Nanosecond}
	if len(batchIndices) != len(wantIndices) {
		t.Fatalf("observed %d batch starts, want %d", len(batchIndices), len(wantIndices))
	}
	for index := range wantIndices {
		if batchIndices[index] != wantIndices[index] || batchStarts[index] < wantStarts[index] {
			t.Fatalf("batch starts = indices %v at %v; want indices %v no earlier than %v", batchIndices, batchStarts, wantIndices, wantStarts)
		}
	}
	select {
	case <-drained:
	case <-time.After(time.Second):
		t.Fatal("source-shaped UDP receiver did not drain the fixed vector")
	}
	fixture.videoNext = time.Now().Add(time.Second)
	beforeCancel := fixture.sentPackets.Load()
	cancelled, cancel := context.WithCancel(context.Background())
	cancel()
	if err := fixture.sendVideo(cancelled, 3, packets[:1]); !errors.Is(err, context.Canceled) {
		t.Fatalf("cancelled pacing wait returned %v, want context.Canceled", err)
	}
	if fixture.sentPackets.Load() != beforeCancel {
		t.Fatal("cancelled pacing wait emitted a UDP shard")
	}
}

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: 20000,
		Duration: time.Second, Warmup: time.Millisecond, PacketBytes: 1000,
	}
	rawPath := filepath.Join(t.TempDir(), "processing.csv.gz")
	summary, err := runQualificationProcessing(t, profile, rawPath)
	if err != nil {
		t.Fatal(err)
	}
	if summary.Count < 1 || summary.RawSamplesSHA256 == "" || summary.RawSamplesBytes < 1 ||
		summary.ResourceSamples < 2 || summary.RawResourcesSHA256 == "" || summary.RawResourcesBytes < 1 ||
		summary.CPUScope != qualificationGatewayCPUScope || summary.ResourceMethod != qualificationResourceMethod ||
		summary.ClockOverhead <= 0 || summary.ClockMethod == "" {
		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,queue_ns,processing_ns,pacing_ns\n") ||
		strings.Count(string(raw), "\n") != int(summary.Count)+1 {
		t.Fatalf("raw sample rows do not match summary count: %q", raw)
	}
}

func TestQualificationShortProcessingSubprocessCoversFixedProfiles(t *testing.T) {
	for _, profile := range qualificationMediaProfiles() {
		profile.Duration = time.Second
		profile.Warmup = 10 * time.Millisecond
		summary, err := runQualificationProcessing(t, profile, filepath.Join(t.TempDir(), profile.Name+".csv.gz"))
		if err != nil {
			t.Fatalf("%s: %v", profile.Name, err)
		}
		if summary.Count < 1 || summary.CPUScope != qualificationGatewayCPUScope ||
			summary.ClockOverhead <= 0 || summary.ClockMethod != qualificationClockOverheadMethod ||
			summary.Count != int64(profile.FPS) ||
			summary.ConfiguredFPS != profile.FPS ||
			summary.ObservedFPS < float64(profile.FPS)*0.95 ||
			summary.ObservedFPS > float64(profile.FPS)*1.05 ||
			summary.PayloadBytes != profile.BitrateKbps*1000/8 ||
			summary.MaximumFrameBytes <= summary.MinimumFrameBytes ||
			summary.MaximumFrameBytes <= 18_864 ||
			summary.PayloadSHA256 == "" ||
			summary.ObservedBitrateKbps < float64(profile.BitrateKbps)*0.95 ||
			summary.ObservedBitrateKbps > float64(profile.BitrateKbps)*1.05 {
			t.Fatalf("%s subprocess summary = %#v", profile.Name, summary)
		}
	}
}

func TestQualificationShortProcessingUsesCompleteFrameCadence(t *testing.T) {
	profile := qualificationMediaProfiles()[0]
	profile.Duration = 100 * time.Millisecond
	profile.Warmup = time.Millisecond
	summary, err := runQualificationProcessing(t, profile, filepath.Join(t.TempDir(), "frames.csv.gz"))
	if err != nil {
		t.Fatal(err)
	}
	if summary.Count != 6 {
		t.Fatalf("100 ms of 1080p60 processed %d units, want 6 complete encoded frames", summary.Count)
	}
}

func TestQualificationSustainedProcessingKeepsCleanPathBounded(t *testing.T) {
	profile := qualificationMediaProfiles()[2]
	profile.Duration = 2 * time.Minute
	profile.Warmup = 100 * time.Millisecond
	summary, err := runQualificationProcessing(t, profile, filepath.Join(t.TempDir(), "4k60-hevc.csv.gz"))
	if err != nil {
		t.Fatal(err)
	}
	if summary.Count < 1 || summary.ObservedBitrateKbps < float64(profile.BitrateKbps)*0.95 ||
		summary.ObservedBitrateKbps > float64(profile.BitrateKbps)*1.05 {
		t.Fatalf("sustained processing summary = %#v", summary)
	}
}

func TestQualificationProcessingResourcesExcludeParentDriverCPU(t *testing.T) {
	profile := qualificationMediaProfile{
		Name: "resource-isolation", Codec: "h264", BitrateKbps: 1000,
		Duration: 300 * time.Millisecond, Warmup: time.Millisecond, PacketBytes: 1000,
	}
	baseline, err := runQualificationProcessing(t, profile, filepath.Join(t.TempDir(), "baseline.csv.gz"))
	if err != nil {
		t.Fatal(err)
	}
	stop := make(chan struct{})
	done := make(chan struct{})
	go func() {
		defer close(done)
		var value uint64 = 1
		for {
			select {
			case <-stop:
				if value == 0 {
					panic("bounded parent work was optimized away")
				}
				return
			default:
				value = value*2862933555777941757 + 3037000493
			}
		}
	}()
	busy, err := runQualificationProcessing(t, profile, filepath.Join(t.TempDir(), "busy.csv.gz"))
	close(stop)
	<-done
	if err != nil {
		t.Fatal(err)
	}
	if busy.CPUSeconds > baseline.CPUSeconds+50*time.Millisecond.Seconds() {
		t.Fatalf("parent CPU leaked into gateway sample: baseline=%.6fs busy=%.6fs", baseline.CPUSeconds, busy.CPUSeconds)
	}
}

func TestQualificationProcessingPreservesPayload(t *testing.T) {
	profile := qualificationMediaProfiles()[0]
	payload := qualificationPayload(profile)
	trace, elapsed, err := newQualificationPath(t, profile, profile.BitrateKbps).traverse(t, payload)
	if err != nil {
		t.Fatal(err)
	}
	if !trace.PayloadPreserved || !trace.ApolloRecovered || !trace.VerseQUIC {
		t.Fatalf("production path trace = %#v", trace)
	}
	if elapsed <= 0 {
		t.Fatalf("processing duration = %s", elapsed)
	}
}

func TestQualificationRepeatedTraversalTracksEveryProductionStage(t *testing.T) {
	profile := qualificationMediaProfiles()[1]
	pacerKbps := (profile.BitrateKbps*int64(profile.PacketBytes+frameHeaderSize) + int64(profile.PacketBytes) - 1) / int64(profile.PacketBytes)
	path := newQualificationPath(t, profile, pacerKbps)
	payload := qualificationPayload(profile)
	if err := runQualificationWarmup(t, path, profile, payload); err != nil {
		t.Fatal(err)
	}
	for index := 0; index < 25_000; index++ {
		current := append([]byte(nil), payload...)
		binary.BigEndian.PutUint32(current[len(current)-4:], uint32(index))
		trace, _, err := path.traverse(t, current)
		if err != nil {
			t.Fatal(err)
		}
		if !trace.NativeUDPIngress || !trace.ApolloRecovered || !trace.ProductionQueue ||
			!trace.ProductionMediaLoop || !trace.ProductionPacer || !trace.VerseQUIC ||
			!trace.PublicClientDecode || !trace.PayloadPreserved {
			t.Fatalf("traversal %d missed a production stage: %#v", index, trace)
		}
	}
}

func TestQualificationZeroLossBaselinesAreAttributedAtNormativeScale(t *testing.T) {
	for _, media := range qualificationMediaProfiles() {
		observation, err := runQualificationImpairment(t, qualificationImpairmentProfiles()[0], media,
			qualificationImpairmentPacketCount, filepath.Join(t.TempDir(), media.Name+".csv.gz"))
		if err != nil {
			t.Fatalf("%s: %v", media.Name, err)
		}
		if observation.InjectedDropped != 0 || observation.Dropped != 0 ||
			observation.ProviderFECDropped != 0 || observation.ProviderEnqueueDropped != 0 ||
			observation.ProviderQueueReplaced != 0 || observation.GatewayDropped != 0 ||
			observation.QUICSendDropped != 0 || observation.ClientDeliveryDropped != 0 ||
			observation.UnexplainedDropped != 0 {
			t.Fatalf("%s clean-path loss attribution = %#v", media.Name, observation)
		}
		if observation.SourceEmitted != qualificationImpairmentPacketCount ||
			observation.ProviderRecovered != qualificationImpairmentPacketCount ||
			observation.ProviderEnqueued != qualificationImpairmentPacketCount ||
			observation.GatewayForwarded != qualificationImpairmentPacketCount ||
			observation.QUICSent != qualificationImpairmentPacketCount ||
			observation.Delivered != qualificationImpairmentPacketCount {
			t.Fatalf("%s stage counts = %#v", media.Name, observation)
		}
	}
}

func TestQualificationImpairmentIsDeterministicAndBounded(t *testing.T) {
	profile := qualificationImpairmentProfiles()[3]
	first, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], 1000, filepath.Join(t.TempDir(), "first.csv.gz"))
	if err != nil {
		t.Fatal(err)
	}
	second, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], 1000, filepath.Join(t.TempDir(), "second.csv.gz"))
	if err != nil {
		t.Fatal(err)
	}
	if first.Dropped != second.Dropped || first.InjectedReordered != second.InjectedReordered {
		t.Fatalf("deterministic impairment selection differs:\n%#v\n%#v", first, second)
	}
	if first.Sent != 1000 || first.Delivered+first.Dropped != first.Sent ||
		first.ObservedLossPercent < 3.5 || first.ObservedLossPercent > 6.5 ||
		first.MaxQueuePackets > qualificationImpairmentQueuePackets || first.RawSamplesSHA256 == "" {
		t.Fatalf("impairment observation = %#v", first)
	}
	if _, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], qualificationImpairmentMaxPackets+1, filepath.Join(t.TempDir(), "invalid.csv.gz")); err == nil {
		t.Fatal("unbounded impairment packet count was accepted")
	}
	unknown := profile
	unknown.Name = "private-simulator"
	if _, err := runQualificationImpairment(t, unknown, qualificationMediaProfiles()[0], 1, filepath.Join(t.TempDir(), "unknown.csv.gz")); err == nil {
		t.Fatal("unregistered impairment profile was accepted")
	}
}

func TestQualificationRTTIsNotSyntheticDoubleOneWayCompletion(t *testing.T) {
	rawPath := filepath.Join(t.TempDir(), "latency.csv.gz")
	observation, err := runQualificationImpairment(
		t,
		qualificationImpairmentProfiles()[1],
		qualificationMediaProfiles()[0],
		40,
		rawPath,
	)
	if err != nil {
		t.Fatal(err)
	}
	meanLatency := qualificationRawMeanLatency(t, rawPath)
	delta := observation.ObservedRTT - 2*meanLatency
	if delta < 0 {
		delta = -delta
	}
	if delta < 5*time.Millisecond {
		t.Fatalf("RTT %s was synthesized as twice one-way completion %s", observation.ObservedRTT, meanLatency)
	}
}

func TestQualificationFixedSeedJitterIsObservableOnTraversedTraffic(t *testing.T) {
	profile := qualificationImpairmentProfiles()[2]
	observation, err := runQualificationImpairment(
		t,
		profile,
		qualificationMediaProfiles()[0],
		200,
		filepath.Join(t.TempDir(), "jitter.csv.gz"),
	)
	if err != nil {
		t.Fatal(err)
	}
	if observation.RTTSource != "apollo_enet_acknowledge" || observation.ObservedRTT <= 0 {
		t.Fatalf("RTT observation is not transport-acknowledged: %#v", observation)
	}
	if observation.ObservedLatency < 5*time.Millisecond || observation.ObservedLatency > 100*time.Millisecond {
		t.Fatalf("observed one-way latency %s does not reflect configured traversal", observation.ObservedLatency)
	}
	if observation.AppliedJitter < 10*time.Millisecond || observation.AppliedJitter > 25*time.Millisecond {
		t.Fatalf("applied fixed-seed jitter %s is outside the uniform-delay tolerance", observation.AppliedJitter)
	}
	if observation.ObservedJitter <= 0 || observation.ObservedJitter > observation.AppliedJitter {
		t.Fatalf("ordered traversal jitter %s is not bounded by applied jitter %s", observation.ObservedJitter, observation.AppliedJitter)
	}
	if observation.InjectedReordered != 0 || observation.ObservedOutOfOrder != 0 {
		t.Fatalf("reorder-off jitter changed source order: %#v", observation)
	}
}

func TestQualificationLossOnlyDoesNotImplicitlyReorder(t *testing.T) {
	observation, err := runQualificationImpairment(
		t,
		qualificationImpairmentProfiles()[3],
		qualificationMediaProfiles()[0],
		400,
		filepath.Join(t.TempDir(), "loss.csv.gz"),
	)
	if err != nil {
		t.Fatal(err)
	}
	if observation.InjectedReordered != 0 || observation.ObservedOutOfOrder != 0 {
		t.Fatalf("loss-only profile changed source order: %#v", observation)
	}
}

func TestQualificationSourceShaperCarriesAtMostOnePacketOfCatchup(t *testing.T) {
	spacing := time.Millisecond
	started := time.Unix(0, 0)
	now := started.Add(100 * spacing)
	first := qualificationBoundedRelease(started, time.Time{}, now, spacing)
	if first != now.Add(-spacing) {
		t.Fatalf("first catch-up release = %s, want %s", first, now.Add(-spacing))
	}
	second := qualificationBoundedRelease(started.Add(spacing), first.Add(spacing), now, spacing)
	if second != now {
		t.Fatalf("second catch-up release = %s, want %s", second, now)
	}
	third := qualificationBoundedRelease(started.Add(2*spacing), second.Add(spacing), now, spacing)
	if third != now.Add(spacing) {
		t.Fatalf("catch-up debt was reset: third release = %s, want %s", third, now.Add(spacing))
	}
}

func TestQualificationExplicitReorderIsBoundedAndAttributed(t *testing.T) {
	observation, err := runQualificationImpairment(
		t,
		qualificationImpairmentProfiles()[4],
		qualificationMediaProfiles()[0],
		400,
		filepath.Join(t.TempDir(), "reorder.csv.gz"),
	)
	if err != nil {
		t.Fatal(err)
	}
	if observation.InjectedReordered == 0 || observation.ObservedOutOfOrder != observation.InjectedReordered {
		t.Fatalf("explicit reorder attribution = %#v", observation)
	}
}

func TestQualificationGatewaySubprocessResourcesResetAndTrackWork(t *testing.T) {
	profile := qualificationMediaProfile{
		Name: "resource-process", Codec: "h264", BitrateKbps: 100000,
		Duration: time.Second, Warmup: time.Millisecond, PacketBytes: 1000,
	}
	path := newQualificationProcessingPath(t, profile, qualificationFramePacerKbps(profile))
	defer path.Close()
	output := t.TempDir()
	record := func(name string, work func() error) qualificationProcessRecordResult {
		t.Helper()
		if err := path.process.startRecording(
			filepath.Join(output, name+".csv.gz"),
			filepath.Join(output, name+"-resources.csv.gz"),
		); err != nil {
			t.Fatal(err)
		}
		if err := work(); err != nil {
			t.Fatal(err)
		}
		result, err := path.process.stopRecording()
		if err != nil {
			t.Fatal(err)
		}
		return result
	}
	idle := record("idle", func() error {
		time.Sleep(150 * time.Millisecond)
		return nil
	})
	payload := qualificationPayload(profile)
	work := record("work", func() error {
		for index := 0; index < 500; index++ {
			if _, err := path.emit(t, payload); err != nil {
				return err
			}
			recovered, err := path.receivePayload(context.Background())
			if err != nil || !bytes.Equal(recovered, payload) {
				return errors.New("gateway subprocess work payload mismatch")
			}
		}
		return nil
	})
	secondIdle := record("idle-again", func() error {
		time.Sleep(150 * time.Millisecond)
		return nil
	})
	if idle.CPUSeconds > 50*time.Millisecond.Seconds() || secondIdle.CPUSeconds > 50*time.Millisecond.Seconds() {
		t.Fatalf("idle gateway CPU was reported as consumed work: first=%.6fs second=%.6fs", idle.CPUSeconds, secondIdle.CPUSeconds)
	}
	if work.CPUSeconds <= idle.CPUSeconds || work.Count != 500 || work.AllocatedObjects == 0 ||
		work.AllocatedBytes == 0 || work.PeakHeapBytes == 0 || work.PeakGoroutines == 0 {
		t.Fatalf("gateway work resource sample = %#v idle=%#v", work, idle)
	}
	if secondIdle.AllocatedObjects >= work.AllocatedObjects || secondIdle.AllocatedBytes >= work.AllocatedBytes {
		t.Fatalf("successive recording inherited counters: work=%#v second=%#v", work, secondIdle)
	}
	if work.ClockOverhead <= 0 || work.ClockMethod != qualificationClockOverheadMethod {
		t.Fatalf("clock overhead evidence = %#v", work)
	}
}

func qualificationRawMeanLatency(t *testing.T, path string) time.Duration {
	t.Helper()
	file, err := os.Open(path)
	if err != nil {
		t.Fatal(err)
	}
	defer file.Close()
	compressed, err := gzip.NewReader(file)
	if err != nil {
		t.Fatal(err)
	}
	raw, err := io.ReadAll(compressed)
	if err != nil {
		t.Fatal(err)
	}
	if err := compressed.Close(); err != nil {
		t.Fatal(err)
	}
	var total time.Duration
	var count int
	for _, line := range strings.Split(string(raw), "\n")[1:] {
		fields := strings.Split(line, ",")
		if len(fields) < 5 || fields[4] != "delivered" {
			continue
		}
		sent, sentErr := strconv.ParseInt(fields[1], 10, 64)
		delivered, deliveredErr := strconv.ParseInt(fields[2], 10, 64)
		if sentErr != nil || deliveredErr != nil || delivered < sent {
			t.Fatalf("invalid raw latency row %q", line)
		}
		total += time.Duration(delivered - sent)
		count++
	}
	if count == 0 {
		t.Fatal("no delivered raw latency rows")
	}
	return total / time.Duration(count)
}

func TestQualificationSixImpairmentProfilesTraverseProductionPath(t *testing.T) {
	profiles := qualificationImpairmentProfiles()
	if len(profiles) != 6 {
		t.Fatalf("impairment profile count = %d, want exactly 6", len(profiles))
	}
	for _, profile := range profiles {
		observation, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], 40,
			filepath.Join(t.TempDir(), profile.Name+".csv.gz"))
		if err != nil {
			t.Fatalf("%s: %v", profile.Name, err)
		}
		if observation.Profile != profile.Name || observation.Delivered+observation.Dropped != 40 ||
			observation.RawSamplesSHA256 == "" || observation.MaxQueuePackets > qualificationImpairmentQueuePackets {
			t.Fatalf("%s observation = %#v", profile.Name, observation)
		}
	}
}

func TestQualificationLossAndSteppedThroughputBounds(t *testing.T) {
	profiles := qualificationImpairmentProfiles()
	for _, profile := range profiles[3:] {
		directory := t.TempDir()
		observation, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0],
			qualificationImpairmentPacketCount, filepath.Join(directory, profile.Name+".csv.gz"))
		if err != nil {
			t.Fatalf("%s: %v", profile.Name, err)
		}
		if observation.ObservedThroughputKbps <= 0 ||
			profile.Name == "constrained" && len(observation.CapacityStepObservations) != 2 {
			t.Fatalf("%s observation = %#v", profile.Name, observation)
		}
		for _, step := range observation.CapacityStepObservations {
			t.Logf("%s %d%%: convergence=%s wire_max_5s=%d wire_cap=%d", profile.Name, step.ReductionPercent, step.Convergence, step.MaximumFiveSecond, step.FiveSecondCap)
		}
		if profile.Name == "constrained" {
			if err := validateQualificationWireBundle(directory, observation, qualificationMediaProfiles()[0], qualificationEvidenceNormative); err != nil {
				t.Fatal(err)
			}
		}
	}
}

func TestQualificationBundleRetainsIndependentlyRecomputableWireEvidence(t *testing.T) {
	profile := qualificationImpairmentProfiles()[5]
	directory := t.TempDir()
	rawPath := filepath.Join(directory, "impairment-constrained-1080p60-h264.csv.gz")
	observation, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], 40, rawPath)
	if err != nil {
		t.Fatal(err)
	}
	if err := validateQualificationWireBundle(directory, observation, qualificationMediaProfiles()[0], qualificationEvidenceSmoke); err != nil {
		t.Fatal(err)
	}
}

func TestQualificationV9AggregateOnlyBundleIsRejectedAsIncomplete(t *testing.T) {
	observation := qualificationImpairmentObservation{
		ConfiguredCapacitySteps: []int{25, 50},
		CapacityStepObservations: []qualificationCapacityStep{
			{ReductionPercent: 25, Convergence: 2 * time.Second, MaximumFiveSecond: 1, FiveSecondCap: 1},
			{ReductionPercent: 50, Convergence: 2 * time.Second, MaximumFiveSecond: 1, FiveSecondCap: 1},
		},
		RawSamples: "impairment-constrained-1080p60-h264.csv.gz",
	}
	if err := validateQualificationWireBundle(t.TempDir(), observation, qualificationMediaProfiles()[0], qualificationEvidenceSmoke); err == nil || !strings.Contains(err.Error(), "missing retained public-wire samples") {
		t.Fatalf("v9 aggregate-only bundle rejection = %v", err)
	}
}

type qualificationEvidenceMode uint8

const (
	qualificationEvidenceSmoke qualificationEvidenceMode = iota
	qualificationEvidenceNormative
)

const qualificationCanonicalCapacityStepCount = 2

var qualificationCanonicalCapacitySteps = [qualificationCanonicalCapacityStepCount]int{25, 50}

type qualificationCSVLimits struct {
	compressedBytes   int64
	decompressedBytes int64
	headerBytes       []int
	fieldBytes        []int
}

type qualificationBoundedCSVReader struct {
	file        *os.File
	compressed  *gzip.Reader
	limited     *io.LimitedReader
	reader      *csv.Reader
	headerBytes []int
	fieldBytes  []int
	first       bool
}

func qualificationMaximumWireOffset(media qualificationMediaProfile) time.Duration {
	minimumKbps := qualificationMediaPacerKbps(media, 50)
	spacing := time.Duration(int64(time.Second) * int64(media.PacketBytes) * 8 / (minimumKbps * 1000))
	maximumNetworkDelay := time.Duration(0)
	for _, profile := range qualificationImpairmentProfiles() {
		maximumNetworkDelay = max(maximumNetworkDelay, profile.RTT+profile.Jitter)
	}
	return time.Duration(qualificationImpairmentMaxPackets)*spacing + maximumNetworkDelay + nativeApolloVideoQueueLatency + 10*time.Second
}

func qualificationMaximumFairnessOffset() time.Duration {
	return 60*time.Second + 2*10*time.Second + nativeApolloVideoQueueLatency
}

func qualificationGzipMaximumBytes(decompressedBytes int64) int64 {
	const maximumStoredBlockBytes = 16_383
	return decompressedBytes + ((decompressedBytes+maximumStoredBlockBytes-1)/maximumStoredBlockBytes)*5 + 64
}

func qualificationCSVLimitsFor(header []string, maximumRows int, maximumFieldBytes []int) qualificationCSVLimits {
	rowBytes := int64(len(maximumFieldBytes))
	fieldBytes := make([]int, len(maximumFieldBytes))
	headerBytes := make([]int, len(header))
	for index, size := range maximumFieldBytes {
		headerBytes[index] = len(header[index])
		fieldBytes[index] = size
		rowBytes += int64(max(size, headerBytes[index]))
	}
	headerLineBytes := int64(len(strings.Join(header, ",")) + 1)
	decompressedBytes := headerLineBytes + int64(maximumRows)*rowBytes
	return qualificationCSVLimits{
		compressedBytes: qualificationGzipMaximumBytes(decompressedBytes), decompressedBytes: decompressedBytes,
		headerBytes: headerBytes, fieldBytes: fieldBytes,
	}
}

func qualificationWireCSVLimits(media qualificationMediaProfile, maximumRows int) qualificationCSVLimits {
	maxOffsetBytes := len(strconv.FormatInt(qualificationMaximumWireOffset(media).Nanoseconds(), 10))
	return qualificationCSVLimitsFor(
		[]string{"record_type", "reduction_percent", "transition_after_ns", "received_after_ns", "encoded_bytes"},
		maximumRows,
		[]int{len("transition"), len("100"), maxOffsetBytes, maxOffsetBytes, len(strconv.Itoa(frameV2HeaderSize + frameV2PayloadSize))},
	)
}

func qualificationFairnessCSVLimits() qualificationCSVLimits {
	maxFlowBytes := 0
	for _, flow := range qualificationFairnessFlows() {
		maxFlowBytes = max(maxFlowBytes, len(flow))
	}
	return qualificationCSVLimitsFor(
		[]string{"elapsed_ns", "flow", "bytes"}, qualificationImpairmentMaxPackets,
		[]int{
			len(strconv.FormatInt(qualificationMaximumFairnessOffset().Nanoseconds(), 10)),
			maxFlowBytes,
			len(strconv.Itoa(1000 + frameHeaderSize)),
		},
	)
}

func openQualificationBoundedCSV(path string, limits qualificationCSVLimits, fieldsPerRecord int) (*qualificationBoundedCSVReader, error) {
	info, err := os.Stat(path)
	if err != nil {
		return nil, err
	}
	if info.Size() > limits.compressedBytes {
		return nil, errors.New("qualification CSV compressed byte bound exceeded")
	}
	file, err := os.Open(path)
	if err != nil {
		return nil, err
	}
	compressed, err := gzip.NewReader(file)
	if err != nil {
		_ = file.Close()
		return nil, err
	}
	limited := &io.LimitedReader{R: compressed, N: limits.decompressedBytes + 1}
	reader := csv.NewReader(limited)
	reader.FieldsPerRecord = fieldsPerRecord
	return &qualificationBoundedCSVReader{
		file: file, compressed: compressed, limited: limited, reader: reader,
		headerBytes: limits.headerBytes, fieldBytes: limits.fieldBytes, first: true,
	}, nil
}

func (reader *qualificationBoundedCSVReader) Read() ([]string, error) {
	row, err := reader.reader.Read()
	if reader.limited.N == 0 {
		return nil, errors.New("qualification CSV decompressed byte bound exceeded")
	}
	if err != nil {
		return nil, err
	}
	bounds := reader.fieldBytes
	if reader.first {
		reader.first = false
		bounds = reader.headerBytes
	}
	for index, field := range row {
		if index >= len(bounds) || len(field) > bounds[index] {
			return nil, errors.New("qualification CSV field bound exceeded")
		}
	}
	return row, nil
}

func (reader *qualificationBoundedCSVReader) Close() {
	_ = reader.compressed.Close()
	_ = reader.file.Close()
}

func validateQualificationWireBundle(directory string, observation qualificationImpairmentObservation, media qualificationMediaProfile, mode qualificationEvidenceMode) error {
	if mode != qualificationEvidenceSmoke && mode != qualificationEvidenceNormative {
		return errors.New("qualification public-wire validation mode invalid")
	}
	if mode == qualificationEvidenceNormative && observation.Sent != qualificationImpairmentPacketCount {
		return fmt.Errorf("qualification normative sent=%d want=%d", observation.Sent, qualificationImpairmentPacketCount)
	}
	if len(observation.ConfiguredCapacitySteps) != len(qualificationCanonicalCapacitySteps) {
		return errors.New("qualification public-wire capacity steps invalid")
	}
	for index, reduction := range qualificationCanonicalCapacitySteps {
		if observation.ConfiguredCapacitySteps[index] != reduction {
			return errors.New("qualification public-wire capacity steps invalid")
		}
	}
	if observation.RawWireSamples == "" || filepath.Base(observation.RawWireSamples) != observation.RawWireSamples {
		return errors.New("qualification bundle missing retained public-wire samples")
	}
	if observation.RawWireTimebase != qualificationWireTimebase {
		return errors.New("qualification public-wire timebase invalid")
	}
	fragmentsPerUnit := (media.PacketBytes + frameV2PayloadSize - 1) / frameV2PayloadSize
	maximumRows := qualificationImpairmentMaxPackets*fragmentsPerUnit + qualificationCanonicalCapacityStepCount
	limits := qualificationWireCSVLimits(media, maximumRows)
	if observation.RawWireDeliveryRows <= 0 || observation.RawWireTransitionRows != qualificationCanonicalCapacityStepCount ||
		observation.RawWireRows < 1 || observation.RawWireRows > maximumRows ||
		observation.RawWireDeliveryRows > maximumRows-qualificationCanonicalCapacityStepCount ||
		observation.RawWireRows != observation.RawWireDeliveryRows+qualificationCanonicalCapacityStepCount {
		return errors.New("qualification public-wire row counts invalid")
	}
	path := filepath.Join(directory, observation.RawWireSamples)
	info, err := os.Stat(path)
	if err != nil {
		return err
	}
	if info.Size() > limits.compressedBytes {
		return errors.New("qualification public-wire compressed byte bound exceeded")
	}
	digest, size, err := qualificationFileSHA256(path)
	if err != nil {
		return err
	}
	if digest != observation.RawWireSamplesSHA256 || size != observation.RawWireSamplesBytes {
		return errors.New("qualification public-wire hash or size mismatch")
	}
	reader, err := openQualificationBoundedCSV(path, limits, 5)
	if err != nil {
		return err
	}
	defer reader.Close()
	header, err := reader.Read()
	if err != nil || !reflect.DeepEqual(header, []string{"record_type", "reduction_percent", "transition_after_ns", "received_after_ns", "encoded_bytes"}) {
		return errors.New("qualification public-wire schema invalid")
	}

	epoch := time.Unix(0, 0)
	transitions := make(map[int]time.Duration, qualificationCanonicalCapacityStepCount)
	deliveries := make([]qualificationDeliverySample, 0, observation.RawWireDeliveryRows)
	expectedLengths := make([]int64, 0, fragmentsPerUnit)
	remaining := media.PacketBytes
	for remaining > 0 {
		payloadBytes := min(remaining, frameV2PayloadSize)
		expectedLengths = append(expectedLengths, int64(payloadBytes+frameV2HeaderSize))
		remaining -= payloadBytes
	}
	lastAfter := int64(-1)
	rows, deliveryRows, transitionRows := 0, 0, 0
	for {
		row, readErr := reader.Read()
		if errors.Is(readErr, io.EOF) {
			break
		}
		if readErr != nil {
			return readErr
		}
		rows++
		if rows > maximumRows {
			return errors.New("qualification public-wire row bound exceeded")
		}
		switch row[0] {
		case "transition":
			if row[1] == "" || row[2] == "" || row[3] != "" || row[4] != "" {
				return errors.New("qualification public-wire transition fields invalid")
			}
			reduction, reductionErr := strconv.Atoi(row[1])
			after, afterErr := strconv.ParseInt(row[2], 10, 64)
			if reductionErr != nil || afterErr != nil || after < 0 || time.Duration(after) > qualificationMaximumWireOffset(media) {
				return errors.New("qualification public-wire transition invalid")
			}
			if _, duplicate := transitions[reduction]; duplicate {
				return errors.New("qualification public-wire transition duplicated")
			}
			transitions[reduction] = time.Duration(after)
			transitionRows++
			if after < lastAfter {
				return errors.New("qualification public-wire records not monotonic")
			}
			lastAfter = after
		case "delivery":
			if row[1] != "" || row[2] != "" || row[3] == "" || row[4] == "" {
				return errors.New("qualification public-wire delivery fields invalid")
			}
			after, afterErr := strconv.ParseInt(row[3], 10, 64)
			encodedBytes, bytesErr := strconv.ParseInt(row[4], 10, 64)
			if afterErr != nil || bytesErr != nil || after < 0 || time.Duration(after) > qualificationMaximumWireOffset(media) || encodedBytes <= 0 {
				return errors.New("qualification public-wire delivery invalid")
			}
			if after < lastAfter {
				return errors.New("qualification public-wire records not monotonic")
			}
			if encodedBytes != expectedLengths[deliveryRows%len(expectedLengths)] {
				return errors.New("qualification public-wire encoded length or logical-frame substitution invalid")
			}
			lastAfter = after
			deliveries = append(deliveries, qualificationDeliverySample{At: epoch.Add(time.Duration(after)), Bytes: encodedBytes})
			deliveryRows++
		default:
			return errors.New("qualification public-wire record type invalid")
		}
	}
	if rows != observation.RawWireRows || deliveryRows != observation.RawWireDeliveryRows ||
		transitionRows != observation.RawWireTransitionRows || deliveryRows != observation.QUICDatagramsSent ||
		transitionRows != qualificationCanonicalCapacityStepCount || deliveryRows%len(expectedLengths) != 0 {
		return errors.New("qualification public-wire retained counts mismatch")
	}
	if len(observation.CapacityStepObservations) != qualificationCanonicalCapacityStepCount {
		return errors.New("qualification public-wire capacity summary count mismatch")
	}
	if _, ok := transitions[25]; !ok {
		return errors.New("qualification public-wire transition missing")
	}
	if _, ok := transitions[50]; !ok {
		return errors.New("qualification public-wire transition missing")
	}
	if transitions[25] >= transitions[50] {
		return errors.New("qualification public-wire transition order invalid")
	}
	for index, reduction := range qualificationCanonicalCapacitySteps {
		transition, ok := transitions[reduction]
		if !ok {
			return errors.New("qualification public-wire transition missing")
		}
		step := observation.CapacityStepObservations[index]
		if step.ReductionPercent != reduction || step.TransitionAfter != transition ||
			step.RecomputationSource != observation.RawWireSamples {
			return errors.New("qualification public-wire manifest transition mismatch")
		}
		start := epoch.Add(transition)
		first := sort.Search(len(deliveries), func(index int) bool { return !deliveries[index].At.Before(start) })
		afterStep := deliveries[first:]
		target := qualificationMediaPacerKbps(media, reduction) * 1000 / 8
		convergence := qualificationMeasuredConvergence(afterStep, start, target)
		maximum := qualificationMaximumDeliveryBytes(afterStep, 5*time.Second)
		if step.Convergence != convergence || step.MaximumFiveSecond != maximum || step.FiveSecondCap != target*5 {
			return fmt.Errorf(
				"qualification public-wire capacity summary mismatch reduction=%d convergence=%s/%s maximum=%d/%d cap=%d/%d",
				reduction, step.Convergence, convergence, step.MaximumFiveSecond, maximum, step.FiveSecondCap, target*5,
			)
		}
		if mode == qualificationEvidenceNormative &&
			(step.Convergence > 10*time.Second || step.MaximumFiveSecond > step.FiveSecondCap*105/100) {
			return fmt.Errorf("qualification normative capacity gate failed reduction=%d convergence=%s maximum=%d cap=%d",
				reduction, step.Convergence, step.MaximumFiveSecond, step.FiveSecondCap)
		}
	}
	return nil
}

func validateQualificationFairnessBundle(directory string, evidence qualificationFairnessEvidence) error {
	if evidence.RawSamples == "" || filepath.Base(evidence.RawSamples) != evidence.RawSamples {
		return errors.New("qualification fairness raw samples missing")
	}
	path := filepath.Join(directory, evidence.RawSamples)
	limits := qualificationFairnessCSVLimits()
	info, err := os.Stat(path)
	if err != nil {
		return err
	}
	if info.Size() > limits.compressedBytes {
		return errors.New("qualification fairness compressed byte bound exceeded")
	}
	digest, size, err := qualificationFileSHA256(path)
	if err != nil {
		return err
	}
	if digest != evidence.RawSamplesSHA256 || size != evidence.RawSamplesBytes {
		return errors.New("qualification fairness hash or size mismatch")
	}
	reader, err := openQualificationBoundedCSV(path, limits, 3)
	if err != nil {
		return err
	}
	defer reader.Close()
	header, err := reader.Read()
	if err != nil || !reflect.DeepEqual(header, []string{"elapsed_ns", "flow", "bytes"}) {
		return errors.New("qualification fairness schema invalid")
	}
	epoch := time.Unix(0, 0)
	deliveries := make([]qualificationFlowDelivery, 0, qualificationImpairmentMaxPackets)
	flowSet := make(map[string]struct{}, 8)
	allowedFlows := make(map[string]struct{}, 8)
	for _, flow := range qualificationFairnessFlows() {
		allowedFlows[flow] = struct{}{}
	}
	lastElapsed := int64(-1)
	for {
		row, readErr := reader.Read()
		if errors.Is(readErr, io.EOF) {
			break
		}
		if readErr != nil {
			return readErr
		}
		if len(deliveries) >= qualificationImpairmentMaxPackets {
			return errors.New("qualification fairness row bound exceeded")
		}
		elapsed, elapsedErr := strconv.ParseInt(row[0], 10, 64)
		encodedBytes, bytesErr := strconv.ParseInt(row[2], 10, 64)
		_, knownFlow := allowedFlows[row[1]]
		if elapsedErr != nil || bytesErr != nil || elapsed < 0 || time.Duration(elapsed) > qualificationMaximumFairnessOffset() ||
			elapsed < lastElapsed || !knownFlow || encodedBytes != int64(1000+frameHeaderSize) {
			return errors.New("qualification fairness row invalid")
		}
		lastElapsed = elapsed
		flowSet[row[1]] = struct{}{}
		deliveries = append(deliveries, qualificationFlowDelivery{at: epoch.Add(time.Duration(elapsed)), flow: row[1], bytes: encodedBytes})
	}
	if len(deliveries) == 0 || len(flowSet) != 8 || len(evidence.CapacitySteps) != 2 {
		return errors.New("qualification fairness retained counts invalid")
	}
	flows := make([]string, 0, len(flowSet))
	for flow := range flowSet {
		flows = append(flows, flow)
	}
	sort.Strings(flows)
	previousTransition := time.Duration(-1)
	for index, reduction := range []int{25, 50} {
		step := evidence.CapacitySteps[index]
		if step.ReductionPercent != reduction || step.TransitionAfter < 0 || step.TransitionAfter <= previousTransition ||
			step.RecomputationSource != evidence.RawSamples {
			return errors.New("qualification fairness transition manifest invalid")
		}
		previousTransition = step.TransitionAfter
		start := epoch.Add(step.TransitionAfter)
		first := sort.Search(len(deliveries), func(index int) bool { return !deliveries[index].at.Before(start) })
		last := len(deliveries)
		if index+1 < len(evidence.CapacitySteps) {
			next := epoch.Add(evidence.CapacitySteps[index+1].TransitionAfter)
			last = sort.Search(len(deliveries), func(index int) bool { return !deliveries[index].at.Before(next) })
		}
		afterStep := append([]qualificationFlowDelivery(nil), deliveries[first:last]...)
		target := int64(8_000*1000/8) * int64(100-reduction) / 100
		convergence := qualificationPacerConvergence(afterStep, start, flows, target)
		maximum := qualificationMaximumFiveSecondBytes(append([]qualificationFlowDelivery(nil), afterStep...))
		if convergence != 2*time.Second || step.Convergence != convergence ||
			step.MaximumFiveSecond != maximum || step.FiveSecondCap != target*5 ||
			step.MaximumFiveSecond > step.FiveSecondCap*105/100 {
			return fmt.Errorf(
				"qualification fairness capacity summary mismatch reduction=%d transition=%s convergence=%s/%s maximum=%d/%d cap=%d/%d",
				reduction, step.TransitionAfter, step.Convergence, convergence,
				step.MaximumFiveSecond, maximum, step.FiveSecondCap, target*5,
			)
		}
	}
	return nil
}

func TestQualificationFairnessTransitionsAreIndependentlyRecomputable(t *testing.T) {
	epoch := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
	flows := []string{"one", "two", "three", "four", "five", "six", "seven", "eight"}
	deliveries := make([]qualificationFlowDelivery, 0, 5_520)
	for _, stage := range []struct {
		start, end time.Duration
		packets    int
	}{
		{0, 2 * time.Second, 12},
		{2 * time.Second, 5 * time.Second, 9},
		{5 * time.Second, 8 * time.Second, 6},
	} {
		for elapsed := stage.start; elapsed < stage.end; elapsed += 100 * time.Millisecond {
			for packet := 0; packet < stage.packets; packet++ {
				for _, flow := range flows {
					deliveries = append(deliveries, qualificationFlowDelivery{
						at: epoch.Add(elapsed + time.Duration(packet)*time.Millisecond), flow: flow, bytes: int64(1000 + frameHeaderSize),
					})
				}
			}
		}
	}
	directory := t.TempDir()
	path := filepath.Join(directory, "fairness.csv.gz")
	if err := writeQualificationPacerSamples(path, deliveries, epoch); err != nil {
		t.Fatal(err)
	}
	evidence := qualificationFairnessEvidence{RawSamples: filepath.Base(path)}
	steps := []struct {
		reduction  int
		transition time.Duration
	}{
		{25, 2 * time.Second},
		{50, 5 * time.Second},
	}
	for stepIndex, step := range steps {
		start := epoch.Add(step.transition)
		first := sort.Search(len(deliveries), func(index int) bool { return !deliveries[index].at.Before(start) })
		last := len(deliveries)
		if stepIndex+1 < len(steps) {
			next := epoch.Add(steps[stepIndex+1].transition)
			last = sort.Search(len(deliveries), func(index int) bool { return !deliveries[index].at.Before(next) })
		}
		afterStep := append([]qualificationFlowDelivery(nil), deliveries[first:last]...)
		target := int64(8_000*1000/8) * int64(100-step.reduction) / 100
		evidence.CapacitySteps = append(evidence.CapacitySteps, qualificationCapacityStep{
			ReductionPercent: step.reduction, TransitionAfter: step.transition,
			Convergence:       qualificationPacerConvergence(afterStep, start, flows, target),
			MaximumFiveSecond: qualificationMaximumFiveSecondBytes(append([]qualificationFlowDelivery(nil), afterStep...)),
			FiveSecondCap:     target * 5, RecomputationSource: filepath.Base(path),
		})
	}
	evidence.RawSamplesSHA256, evidence.RawSamplesBytes, _ = qualificationFileSHA256(path)
	if err := validateQualificationFairnessBundle(directory, evidence); err != nil {
		t.Fatal(err)
	}
}

func readQualificationTestCSV(t *testing.T, path string) [][]string {
	t.Helper()
	file, err := os.Open(path)
	if err != nil {
		t.Fatal(err)
	}
	defer file.Close()
	compressed, err := gzip.NewReader(file)
	if err != nil {
		t.Fatal(err)
	}
	defer compressed.Close()
	rows, err := csv.NewReader(compressed).ReadAll()
	if err != nil {
		t.Fatal(err)
	}
	return rows
}

func writeQualificationTestCSV(t *testing.T, path string, rows [][]string) {
	t.Helper()
	file, err := os.Create(path)
	if err != nil {
		t.Fatal(err)
	}
	compressed := gzip.NewWriter(file)
	buffered := bufio.NewWriter(compressed)
	writer := csv.NewWriter(buffered)
	writer.WriteAll(rows)
	if err := writer.Error(); err != nil {
		t.Fatal(err)
	}
	if err := buffered.Flush(); err != nil {
		t.Fatal(err)
	}
	if err := compressed.Close(); err != nil {
		t.Fatal(err)
	}
	if err := file.Close(); err != nil {
		t.Fatal(err)
	}
}

func qualificationTestWireBundle(t *testing.T, sent int) (string, qualificationImpairmentObservation, qualificationMediaProfile, []qualificationDeliverySample, map[int]time.Time) {
	t.Helper()
	media := qualificationMediaProfiles()[0]
	epoch := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
	transitions := map[int]time.Time{25: epoch.Add(time.Second), 50: epoch.Add(2 * time.Second)}
	deliveries := []qualificationDeliverySample{
		{At: transitions[25], Bytes: 1200},
		{At: transitions[25].Add(time.Nanosecond), Bytes: 25},
		{At: transitions[50], Bytes: 1200},
		{At: transitions[50].Add(time.Nanosecond), Bytes: 25},
	}
	directory := t.TempDir()
	path := filepath.Join(directory, "impairment-constrained-1080p60-h264-wire.csv.gz")
	evidence, err := writeQualificationWireSamples(path, epoch, []int{25, 50}, transitions, deliveries, len(deliveries)+2)
	if err != nil {
		t.Fatal(err)
	}
	observation := qualificationImpairmentObservation{
		Sent: sent, ConfiguredCapacitySteps: []int{25, 50}, QUICDatagramsSent: len(deliveries),
		RawWireSamples: evidence.Name, RawWireSamplesSHA256: evidence.SHA256, RawWireSamplesBytes: evidence.Bytes,
		RawWireRows: evidence.Rows, RawWireDeliveryRows: evidence.DeliveryRows,
		RawWireTransitionRows: evidence.TransitionRows, RawWireTimebase: qualificationWireTimebase,
	}
	for _, reduction := range observation.ConfiguredCapacitySteps {
		step := qualificationCapacityStepObservation(deliveries, transitions[reduction], media, reduction)
		step.TransitionAfter = transitions[reduction].Sub(epoch)
		step.RecomputationSource = evidence.Name
		observation.CapacityStepObservations = append(observation.CapacityStepObservations, step)
	}
	return directory, observation, media, deliveries, transitions
}

func TestQualificationCapacityStepRetainsFullPostTransitionTail(t *testing.T) {
	media := qualificationMediaProfiles()[0]
	epoch := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
	quarter := epoch.Add(time.Second)
	half := epoch.Add(3 * time.Second)
	deliveries := make([]qualificationDeliverySample, 0, 120)
	for at := quarter; at.Before(epoch.Add(7 * time.Second)); at = at.Add(100 * time.Millisecond) {
		deliveries = append(deliveries,
			qualificationDeliverySample{At: at, Bytes: 1200},
			qualificationDeliverySample{At: at.Add(time.Nanosecond), Bytes: 25},
		)
	}
	got := qualificationCapacityStepObservation(deliveries, quarter, media, 25)
	fullTail := qualificationDeliveriesAfter(deliveries, quarter)
	wantMaximum := qualificationMaximumDeliveryBytes(fullTail, 5*time.Second)
	wantConvergence := qualificationMeasuredConvergence(fullTail, quarter, qualificationMediaPacerKbps(media, 25)*1000/8)
	stageEnd := sort.Search(len(deliveries), func(index int) bool { return !deliveries[index].At.Before(half) })
	stageMaximum := qualificationMaximumDeliveryBytes(deliveries[:stageEnd], 5*time.Second)
	if wantMaximum <= stageMaximum {
		t.Fatalf("test fixture full-tail maximum=%d stage-only=%d", wantMaximum, stageMaximum)
	}
	if got.MaximumFiveSecond != wantMaximum || got.Convergence != wantConvergence {
		t.Fatalf("25%% summary used shortened stage: got maximum=%d convergence=%s, full-tail maximum=%d convergence=%s, stage-only=%d",
			got.MaximumFiveSecond, got.Convergence, wantMaximum, wantConvergence, stageMaximum)
	}
}

func TestQualificationNormativeValidationCannotTrustPersistedSent(t *testing.T) {
	directory, observation, media, _, _ := qualificationTestWireBundle(t, qualificationImpairmentPacketCount-1)
	if err := validateQualificationWireBundle(directory, observation, media, qualificationEvidenceNormative); err == nil {
		t.Fatal("normative validation accepted persisted sent=9999 and 11-second convergence sentinel")
	}
}

func TestQualificationWireBundleRejectsReversedCapacitySteps(t *testing.T) {
	directory, observation, media, _, _ := qualificationTestWireBundle(t, 40)
	observation.ConfiguredCapacitySteps = []int{50, 25}
	observation.CapacityStepObservations[0], observation.CapacityStepObservations[1] =
		observation.CapacityStepObservations[1], observation.CapacityStepObservations[0]
	if err := validateQualificationWireBundle(directory, observation, media, qualificationEvidenceSmoke); err == nil {
		t.Fatal("qualification wire bundle accepted reversed capacity steps")
	}
}

func TestQualificationWireBundleRejectsSwappedTransitionChronology(t *testing.T) {
	directory, observation, media, deliveries, transitions := qualificationTestWireBundle(t, 40)
	rows := readQualificationTestCSV(t, filepath.Join(directory, observation.RawWireSamples))
	for _, row := range rows[1:] {
		if row[0] != "transition" {
			continue
		}
		switch row[1] {
		case "25":
			row[1] = "50"
		case "50":
			row[1] = "25"
		}
	}
	mutatedDirectory := t.TempDir()
	path := filepath.Join(mutatedDirectory, observation.RawWireSamples)
	writeQualificationTestCSV(t, path, rows)
	observation.RawWireSamplesSHA256, observation.RawWireSamplesBytes, _ = qualificationFileSHA256(path)
	transitions[25], transitions[50] = transitions[50], transitions[25]
	observation.CapacityStepObservations = nil
	epoch := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
	for _, reduction := range observation.ConfiguredCapacitySteps {
		step := qualificationCapacityStepObservation(deliveries, transitions[reduction], media, reduction)
		step.TransitionAfter = transitions[reduction].Sub(epoch)
		step.RecomputationSource = observation.RawWireSamples
		observation.CapacityStepObservations = append(observation.CapacityStepObservations, step)
	}
	if err := validateQualificationWireBundle(mutatedDirectory, observation, media, qualificationEvidenceSmoke); err == nil ||
		!strings.Contains(err.Error(), "transition order invalid") {
		t.Fatalf("qualification swapped transition chronology rejection = %v", err)
	}
}

func TestQualificationWireBundleRejectsSelfConsistentNoncanonicalSteps(t *testing.T) {
	directory, observation, media, deliveries, transitions := qualificationTestWireBundle(t, 40)
	rows := readQualificationTestCSV(t, filepath.Join(directory, observation.RawWireSamples))
	for _, row := range rows[1:] {
		if row[0] != "transition" {
			continue
		}
		switch row[1] {
		case "25":
			row[1] = "10"
		case "50":
			row[1] = "20"
		}
	}
	mutatedDirectory := t.TempDir()
	path := filepath.Join(mutatedDirectory, observation.RawWireSamples)
	writeQualificationTestCSV(t, path, rows)
	observation.RawWireSamplesSHA256, observation.RawWireSamplesBytes, _ = qualificationFileSHA256(path)
	observation.ConfiguredCapacitySteps = []int{10, 20}
	observation.CapacityStepObservations = nil
	for index, reduction := range observation.ConfiguredCapacitySteps {
		transition := transitions[[]int{25, 50}[index]]
		step := qualificationCapacityStepObservation(deliveries, transition, media, reduction)
		step.TransitionAfter = transition.Sub(time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC))
		step.RecomputationSource = observation.RawWireSamples
		observation.CapacityStepObservations = append(observation.CapacityStepObservations, step)
	}
	if err := validateQualificationWireBundle(mutatedDirectory, observation, media, qualificationEvidenceSmoke); err == nil {
		t.Fatal("qualification wire bundle accepted self-consistent noncanonical capacity steps")
	}
}

func TestQualificationWireBundleRejectsNegativeDeliveryRowsWithoutPanic(t *testing.T) {
	directory, observation, media, _, _ := qualificationTestWireBundle(t, 40)
	observation.RawWireRows = 1
	observation.RawWireDeliveryRows = -1
	observation.RawWireTransitionRows = 2
	defer func() {
		if recovered := recover(); recovered != nil {
			t.Fatalf("qualification wire bundle panicked for negative delivery rows: %v", recovered)
		}
	}()
	if err := validateQualificationWireBundle(directory, observation, media, qualificationEvidenceSmoke); err == nil {
		t.Fatal("qualification wire bundle accepted negative delivery rows")
	}
}

func TestQualificationNormativeRejectsExactSentConvergenceFailure(t *testing.T) {
	directory, observation, media, _, _ := qualificationTestWireBundle(t, qualificationImpairmentPacketCount)
	err := validateQualificationWireBundle(directory, observation, media, qualificationEvidenceNormative)
	if err == nil || !strings.Contains(err.Error(), "normative capacity gate failed") || strings.Contains(err.Error(), "sent=") {
		t.Fatalf("normative exact-sent convergence rejection = %v", err)
	}
}

func TestQualificationEvidenceReadersRejectOversizedFieldsBeforeParsing(t *testing.T) {
	t.Run("wire", func(t *testing.T) {
		directory, observation, media, _, _ := qualificationTestWireBundle(t, 40)
		rows := readQualificationTestCSV(t, filepath.Join(directory, observation.RawWireSamples))
		for index := 1; index < len(rows); index++ {
			if rows[index][0] == "delivery" {
				rows[index][3] = strings.Repeat("9", 4096)
				break
			}
		}
		mutatedDirectory := t.TempDir()
		path := filepath.Join(mutatedDirectory, observation.RawWireSamples)
		writeQualificationTestCSV(t, path, rows)
		observation.RawWireSamplesSHA256, observation.RawWireSamplesBytes, _ = qualificationFileSHA256(path)
		if err := validateQualificationWireBundle(mutatedDirectory, observation, media, qualificationEvidenceSmoke); err == nil || !strings.Contains(err.Error(), "field bound") {
			t.Fatalf("wire oversized field rejection = %v", err)
		}
	})

	t.Run("fairness", func(t *testing.T) {
		epoch := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
		deliveries := make([]qualificationFlowDelivery, 0, 8)
		for _, flow := range []string{"one", "two", "three", "four", "five", "six", "seven", "eight"} {
			deliveries = append(deliveries, qualificationFlowDelivery{at: epoch, flow: flow, bytes: 1023})
		}
		directory := t.TempDir()
		path := filepath.Join(directory, "fairness.csv.gz")
		if err := writeQualificationPacerSamples(path, deliveries, epoch); err != nil {
			t.Fatal(err)
		}
		rows := readQualificationTestCSV(t, path)
		rows[1][1] = strings.Repeat("f", 4096)
		mutatedDirectory := t.TempDir()
		mutatedPath := filepath.Join(mutatedDirectory, "fairness.csv.gz")
		writeQualificationTestCSV(t, mutatedPath, rows)
		evidence := qualificationFairnessEvidence{RawSamples: "fairness.csv.gz", CapacitySteps: []qualificationCapacityStep{{}, {}}}
		evidence.RawSamplesSHA256, evidence.RawSamplesBytes, _ = qualificationFileSHA256(mutatedPath)
		if err := validateQualificationFairnessBundle(mutatedDirectory, evidence); err == nil || !strings.Contains(err.Error(), "field bound") {
			t.Fatalf("fairness oversized field rejection = %v", err)
		}
	})
}

func TestQualificationBoundedCSVReaderRejectsCompressedAndDecompressedOverflow(t *testing.T) {
	t.Run("compressed", func(t *testing.T) {
		path := filepath.Join(t.TempDir(), "oversized.csv.gz")
		file, err := os.Create(path)
		if err != nil {
			t.Fatal(err)
		}
		if err := file.Truncate(33); err != nil {
			t.Fatal(err)
		}
		if err := file.Close(); err != nil {
			t.Fatal(err)
		}
		_, err = openQualificationBoundedCSV(path, qualificationCSVLimits{compressedBytes: 32, decompressedBytes: 32, headerBytes: []int{5}, fieldBytes: []int{64}}, 1)
		if err == nil || !strings.Contains(err.Error(), "compressed byte bound") {
			t.Fatalf("compressed overflow rejection = %v", err)
		}
	})

	t.Run("decompressed", func(t *testing.T) {
		path := filepath.Join(t.TempDir(), "oversized.csv.gz")
		writeQualificationTestCSV(t, path, [][]string{{"value"}, {strings.Repeat("x", 64)}})
		reader, err := openQualificationBoundedCSV(path, qualificationCSVLimits{
			compressedBytes: 1024, decompressedBytes: 32, headerBytes: []int{5}, fieldBytes: []int{64},
		}, 1)
		if err != nil {
			t.Fatal(err)
		}
		defer reader.Close()
		if _, err := reader.Read(); err == nil || !strings.Contains(err.Error(), "decompressed byte bound") {
			t.Fatalf("decompressed overflow rejection = %v", err)
		}
	})
}

func TestQualificationWireEvidenceRejectsIncompleteOrMalformedBundles(t *testing.T) {
	profile := qualificationImpairmentProfiles()[5]
	media := qualificationMediaProfiles()[0]
	sourceDirectory := t.TempDir()
	observation, err := runQualificationImpairment(t, profile, media, 40,
		filepath.Join(sourceDirectory, "impairment-constrained-1080p60-h264.csv.gz"))
	if err != nil {
		t.Fatal(err)
	}
	if err := validateQualificationWireBundle(sourceDirectory, observation, media, qualificationEvidenceSmoke); err != nil {
		t.Fatal(err)
	}
	wireRows := readQualificationTestCSV(t, filepath.Join(sourceDirectory, observation.RawWireSamples))

	metadataCases := []struct {
		name   string
		mutate func(*qualificationImpairmentObservation)
	}{
		{"missing-artifact", func(value *qualificationImpairmentObservation) { value.RawWireSamples = "" }},
		{"wrong-hash", func(value *qualificationImpairmentObservation) { value.RawWireSamplesSHA256 = strings.Repeat("0", 64) }},
		{"wrong-size", func(value *qualificationImpairmentObservation) { value.RawWireSamplesBytes++ }},
		{"wrong-count", func(value *qualificationImpairmentObservation) { value.RawWireRows++ }},
		{"manifest-mismatch", func(value *qualificationImpairmentObservation) { value.CapacityStepObservations[0].Convergence++ }},
	}
	for _, testCase := range metadataCases {
		t.Run(testCase.name, func(t *testing.T) {
			mutated := observation
			mutated.CapacityStepObservations = append([]qualificationCapacityStep(nil), observation.CapacityStepObservations...)
			testCase.mutate(&mutated)
			if err := validateQualificationWireBundle(sourceDirectory, mutated, media, qualificationEvidenceSmoke); err == nil {
				t.Fatal("malformed qualification wire bundle accepted")
			}
		})
	}

	type rowMutation struct {
		name   string
		mutate func([][]string, *qualificationImpairmentObservation) [][]string
	}
	rowCases := []rowMutation{
		{"duplicate-transition", func(rows [][]string, value *qualificationImpairmentObservation) [][]string {
			for index := 1; index < len(rows); index++ {
				if rows[index][0] == "transition" {
					duplicate := append([]string(nil), rows[index]...)
					rows = append(rows, nil)
					copy(rows[index+1:], rows[index:])
					rows[index] = duplicate
					value.RawWireRows++
					value.RawWireTransitionRows++
					break
				}
			}
			return rows
		}},
		{"missing-transition", func(rows [][]string, value *qualificationImpairmentObservation) [][]string {
			for index := 1; index < len(rows); index++ {
				if rows[index][0] == "transition" {
					copy(rows[index:], rows[index+1:])
					rows[len(rows)-1] = nil
					value.RawWireRows--
					value.RawWireTransitionRows--
					break
				}
			}
			return rows
		}},
		{"delivery-before-epoch", func(rows [][]string, _ *qualificationImpairmentObservation) [][]string {
			for index := 1; index < len(rows); index++ {
				if rows[index][0] == "delivery" {
					rows[index][3] = "-1"
					break
				}
			}
			return rows
		}},
		{"invalid-encoded-length", func(rows [][]string, _ *qualificationImpairmentObservation) [][]string {
			for index := 1; index < len(rows); index++ {
				if rows[index][0] == "delivery" {
					rows[index][4] = "0"
					break
				}
			}
			return rows
		}},
		{"logical-frame-substitution", func(rows [][]string, _ *qualificationImpairmentObservation) [][]string {
			for index := 1; index < len(rows); index++ {
				if rows[index][0] == "delivery" {
					rows[index][4] = "1179"
					break
				}
			}
			return rows
		}},
		{"invalid-empty-fields", func(rows [][]string, _ *qualificationImpairmentObservation) [][]string {
			for index := 1; index < len(rows); index++ {
				if rows[index][0] == "delivery" {
					rows[index][1] = "25"
					break
				}
			}
			return rows
		}},
		{"nonmonotonic", func(rows [][]string, _ *qualificationImpairmentObservation) [][]string {
			first := int64(-1)
			for index := 1; index < len(rows); index++ {
				if rows[index][0] != "delivery" {
					continue
				}
				if first < 0 {
					first, _ = strconv.ParseInt(rows[index][3], 10, 64)
					continue
				}
				rows[index][3] = strconv.FormatInt(max(first-1, 0), 10)
				break
			}
			return rows
		}},
	}
	for _, testCase := range rowCases {
		t.Run(testCase.name, func(t *testing.T) {
			rows := make([][]string, len(wireRows))
			for index := range wireRows {
				rows[index] = append([]string(nil), wireRows[index]...)
			}
			mutated := observation
			mutated.CapacityStepObservations = append([]qualificationCapacityStep(nil), observation.CapacityStepObservations...)
			rows = testCase.mutate(rows, &mutated)
			trimmed := rows[:0]
			for _, row := range rows {
				if row != nil {
					trimmed = append(trimmed, row)
				}
			}
			directory := t.TempDir()
			path := filepath.Join(directory, mutated.RawWireSamples)
			writeQualificationTestCSV(t, path, trimmed)
			mutated.RawWireSamplesSHA256, mutated.RawWireSamplesBytes, _ = qualificationFileSHA256(path)
			if err := validateQualificationWireBundle(directory, mutated, media, qualificationEvidenceSmoke); err == nil {
				t.Fatal("malformed qualification wire bundle accepted")
			}
		})
	}
}

func TestQualificationCapacityStepsMeasurePublicWireDatagrams(t *testing.T) {
	const packetCount = qualificationImpairmentPacketCount
	profile := qualificationMediaProfiles()[0]
	started := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
	spacing := time.Duration(int64(time.Second) * int64(profile.PacketBytes) * 8 / (profile.BitrateKbps * 1000))
	stepAt := map[int]time.Time{
		25: started.Add(time.Duration(packetCount/3) * spacing),
		50: started.Add(time.Duration(packetCount*2/3) * spacing),
	}
	if phase := stepAt[50].Sub(stepAt[25]); phase != 1_571_842_800*time.Nanosecond {
		t.Fatalf("25%% phase = %s, want exact source-shaped transition", phase)
	}

	pacer := newFairPacer(qualificationMediaPacerKbps(profile, 0))
	const flow = "qualification-wire-flow"
	now := started
	stalled := false
	quarterApplied := false
	halfApplied := false
	debtBeforeQuarter := time.Duration(0)
	logicalDeliveries := make([]qualificationDeliverySample, 0, packetCount)
	type wireSample struct {
		reserved time.Time
		delivery qualificationDeliverySample
	}
	wireSamples := make([]wireSample, 0, packetCount*2)
	for index := 0; index < packetCount; index++ {
		sourceAt := started.Add(time.Duration(index) * spacing)
		if now.Before(sourceAt) {
			now = sourceAt
		}
		if !stalled && !sourceAt.Before(stepAt[25].Add(-150*time.Millisecond)) {
			now = now.Add(42 * time.Millisecond)
			stalled = true
		}
		for _, encodedBytes := range []int{1200, 25} {
			if !quarterApplied && !now.Before(stepAt[25]) {
				pacer.mu.Lock()
				debtBeforeQuarter = pacer.flows[flow].debt
				pacer.mu.Unlock()
				pacer.setKbps(qualificationMediaPacerKbps(profile, 25))
				quarterApplied = true
			}
			if !halfApplied && !now.Before(stepAt[50]) {
				pacer.setKbps(qualificationMediaPacerKbps(profile, 50))
				halfApplied = true
			}
			reserved := pacer.reserveAt(now, flow, encodedBytes)
			if reserved.After(now) {
				now = reserved
			}
			wireSamples = append(wireSamples, wireSample{
				reserved: reserved,
				delivery: qualificationDeliverySample{At: now, Bytes: int64(encodedBytes)},
			})
		}
		logicalDeliveries = append(logicalDeliveries, qualificationDeliverySample{At: now, Bytes: 1225})
	}
	if !quarterApplied || !halfApplied || debtBeforeQuarter <= 0 || debtBeforeQuarter > nativeApolloVideoQueueLatency-fairPacerMaximumCatchup {
		t.Fatalf("capacity transition state quarter=%t half=%t debt=%s", quarterApplied, halfApplied, debtBeforeQuarter)
	}
	pacer.mu.Lock()
	remainingDebt := pacer.flows[flow].debt
	pacer.mu.Unlock()
	if remainingDebt != 0 {
		t.Fatalf("remaining debt = %s, want zero", remainingDebt)
	}
	t.Logf("wire model: datagrams=%d debt_before_25=%s remaining_debt=%s", len(wireSamples), debtBeforeQuarter, remainingDebt)
	wireDeliveries := make([]qualificationDeliverySample, len(wireSamples))
	for index, sample := range wireSamples {
		if sample.delivery.At.Before(sample.reserved) {
			t.Fatalf("wire datagram %d delivered at %s before reservation %s", index, sample.delivery.At, sample.reserved)
		}
		wireDeliveries[index] = sample.delivery
	}
	directory := t.TempDir()
	wirePath := filepath.Join(directory, "impairment-constrained-1080p60-h264-wire.csv.gz")
	bundle := qualificationImpairmentObservation{
		Sent:                    packetCount,
		ConfiguredCapacitySteps: []int{25, 50},
		QUICDatagramsSent:       len(wireDeliveries),
		RawWireTimebase:         qualificationWireTimebase,
	}

	for _, reduction := range []int{25, 50} {
		wireBytesPerSecond := qualificationMediaPacerKbps(profile, reduction) * 1000 / 8
		wireAfterStep := qualificationDeliveriesAfter(wireDeliveries, stepAt[reduction])
		maximum := qualificationMaximumDeliveryBytes(wireAfterStep, 5*time.Second)
		if maximum > wireBytesPerSecond*5*105/100 {
			t.Fatalf("%d%% wire five-second maximum = %d, cap = %d", reduction, maximum, wireBytesPerSecond*5*105/100)
		}
		payloadBytesPerSecond := profile.BitrateKbps * int64(100-reduction) * 1000 / 100 / 8
		legacy := qualificationMeasuredConvergence(
			qualificationDeliveriesAfter(logicalDeliveries, stepAt[reduction]), stepAt[reduction], payloadBytesPerSecond,
		)
		if reduction == 25 && legacy != 11*time.Second {
			t.Fatalf("25%% complete-frame classifier convergence = %s, want 11s sentinel reproduction", legacy)
		}
		stepObservation := qualificationCapacityStepObservation(wireDeliveries, stepAt[reduction], profile, reduction)
		stepObservation.TransitionAfter = stepAt[reduction].Sub(started)
		stepObservation.RecomputationSource = filepath.Base(wirePath)
		if stepObservation.Convergence > 10*time.Second {
			for offset := time.Duration(0); offset < 2*time.Second; offset += 250 * time.Millisecond {
				var bucket int64
				for _, delivery := range wireAfterStep {
					if !delivery.At.Before(stepAt[reduction].Add(offset)) && delivery.At.Before(stepAt[reduction].Add(offset+250*time.Millisecond)) {
						bucket += delivery.Bytes
					}
				}
				t.Logf("%d%% wire bucket %s = %d bytes (%d B/s)", reduction, offset, bucket, bucket*4)
			}
			t.Fatalf("%d%% wire convergence = %s sentinel with wire maximum %d under cap %d", reduction, stepObservation.Convergence, maximum, wireBytesPerSecond*5*105/100)
		}
		t.Logf("%d%%: payload_target=%d wire_target=%d legacy=%s wire_convergence=%s wire_max_5s=%d wire_cap_105=%d",
			reduction, payloadBytesPerSecond, wireBytesPerSecond, legacy, stepObservation.Convergence, maximum, wireBytesPerSecond*5*105/100)
		bundle.CapacityStepObservations = append(bundle.CapacityStepObservations, stepObservation)
	}
	wireEvidence, err := writeQualificationWireSamples(wirePath, started, []int{25, 50}, stepAt, wireDeliveries, len(wireDeliveries)+2)
	if err != nil {
		t.Fatal(err)
	}
	bundle.RawWireSamples = wireEvidence.Name
	bundle.RawWireSamplesSHA256 = wireEvidence.SHA256
	bundle.RawWireSamplesBytes = wireEvidence.Bytes
	bundle.RawWireRows = wireEvidence.Rows
	bundle.RawWireDeliveryRows = wireEvidence.DeliveryRows
	bundle.RawWireTransitionRows = wireEvidence.TransitionRows
	if err := validateQualificationWireBundle(directory, bundle, profile, qualificationEvidenceNormative); err != nil {
		t.Fatal(err)
	}
}

func TestQualificationUsesPublicQUICAndProductionPacer(t *testing.T) {
	qualificationTraverseProfiles(t, qualificationMediaProfiles())
	directory := t.TempDir()
	evidence, err := qualificationPacerEvidence(t, filepath.Join(directory, "fairness.csv.gz"), 2*time.Second, 4*time.Second)
	if err != nil {
		t.Fatal(err)
	}
	if len(evidence.PerFlowBytes) != 8 || len(evidence.CapacitySteps) != 2 ||
		len(evidence.Series) != 10 || evidence.RawSamplesSHA256 == "" || evidence.JainIndex < 0.99 {
		t.Fatalf("pacer evidence = %#v", evidence)
	}
	for _, step := range evidence.CapacitySteps {
		if step.Convergence > 10*time.Second || step.MaximumFiveSecond > step.FiveSecondCap*105/100 {
			t.Fatalf("capacity step = %#v", step)
		}
	}
	if err := validateQualificationFairnessBundle(directory, evidence); err != nil {
		t.Fatal(err)
	}
}

func TestQualificationSmokeTraversesNativeApolloRecoveryQueuePacerAndQUIC(t *testing.T) {
	trace := qualificationProductionPathSmoke(t, qualificationMediaProfiles()[0])
	if !trace.ApolloRecovered || !trace.ProductionQueue || !trace.ProductionPacer ||
		!trace.VerseQUIC || !trace.PayloadPreserved {
		t.Fatalf("qualification production-path trace = %#v", trace)
	}
}

func TestQualificationCarriesCompleteLargeFramesThroughPublicPath(t *testing.T) {
	profile := qualificationMediaProfiles()[2]
	path := newQualificationPath(t, profile, 200000)
	defer path.Close()

	for _, size := range []int{24 * 1024, 96 * 1024, 384 * 1024} {
		payload := make([]byte, size)
		for index := range payload {
			payload[index] = byte(index*31 + size)
		}
		trace, _, err := path.traverse(t, payload)
		if err != nil {
			t.Fatalf("frame bytes=%d: %v", size, err)
		}
		if !trace.NativeUDPIngress || !trace.ApolloRecovered || !trace.ProductionQueue ||
			!trace.ProductionMediaLoop || !trace.ProductionPacer || !trace.VerseQUIC ||
			!trace.PublicClientDecode || !trace.PayloadPreserved {
			t.Fatalf("frame bytes=%d skipped path: %#v", size, trace)
		}
	}
}

func TestQualificationPacerRepaysThreeMediaLoopStallsThroughPublicPath(t *testing.T) {
	profile := qualificationMediaProfiles()[0]
	const totalFrames = 480
	barriers := []int64{30, 180, 330}
	type stall struct {
		public  chan struct{}
		blocked chan uint64
		release chan struct{}
	}
	stalls := make([]stall, len(barriers))
	for index := range stalls {
		stalls[index] = stall{public: make(chan struct{}), blocked: make(chan uint64, 1), release: make(chan struct{})}
	}

	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	var path *qualificationPath
	var completed int64
	var observationMu sync.Mutex
	var maximumQueueDelay time.Duration
	observer := func(observation mediaTimingObservation) {
		observationMu.Lock()
		maximumQueueDelay = max(maximumQueueDelay, observation.QueueDelay)
		completed++
		current := completed
		observationMu.Unlock()
		for index, barrier := range barriers {
			if current != barrier {
				continue
			}
			select {
			case <-stalls[index].public:
			case <-ctx.Done():
				return
			}
			baseline := path.session.mediaRecovered.Load()
			select {
			case stalls[index].blocked <- baseline:
			case <-ctx.Done():
				return
			}
			select {
			case <-stalls[index].release:
			case <-ctx.Done():
			}
			return
		}
	}
	path = newQualificationPathWithNativeBackendAndObserver(
		t, profile, qualificationFramePacerKbps(profile), nil, NewNativeApolloBackend(), observer,
	)
	defer path.Close()

	spacing := time.Second / time.Duration(profile.FPS)
	started := time.Now().Add(10 * time.Millisecond)
	producerDone := make(chan error, 1)
	go func() {
		var nextRelease time.Time
		for index := int64(0); index < totalFrames; index++ {
			release := qualificationBoundedRelease(
				started.Add(time.Duration(index)*spacing), nextRelease, time.Now(), spacing,
			)
			if err := qualificationWaitContext(ctx, release); err != nil {
				producerDone <- err
				return
			}
			nextRelease = release.Add(spacing)
			if _, err := path.emit(t, qualificationFramePayload(profile, index)); err != nil {
				producerDone <- err
				return
			}
		}
		producerDone <- nil
	}()

	var expectedVersePackets uint64
	receiverDone := make(chan error, 1)
	go func() {
		for index := int64(0); index < totalFrames; index++ {
			payload, err := path.receivePayload(ctx)
			if err != nil {
				receiverDone <- err
				return
			}
			expected := qualificationFramePayload(profile, index)
			if !bytes.Equal(payload, expected) {
				receiverDone <- fmt.Errorf("public payload changed at frame %d", index)
				return
			}
			fragments := (len(payload) + frameV2PayloadSize - 1) / frameV2PayloadSize
			expectedVersePackets += uint64(fragments)
			for barrierIndex, barrier := range barriers {
				if index+1 == barrier {
					close(stalls[barrierIndex].public)
				}
			}
		}
		receiverDone <- nil
	}()

	for index := range stalls {
		var baseline uint64
		select {
		case baseline = <-stalls[index].blocked:
		case <-ctx.Done():
			t.Fatalf("media loop did not block at public frame %d: %v", barriers[index], ctx.Err())
		}
		target := baseline + 6
		for path.session.mediaRecovered.Load() < target {
			select {
			case <-ctx.Done():
				t.Fatalf("stall %d recovered %d, want at least %d: %v", index, path.session.mediaRecovered.Load(), target, ctx.Err())
			default:
				runtime.Gosched()
			}
		}
		close(stalls[index].release)
	}
	if err := <-producerDone; err != nil {
		t.Fatalf("source fixture: %v", err)
	}
	if err := <-receiverDone; err != nil {
		t.Fatalf("independent client: %v; expected=%d sent=%d source=%d ingress=%d recovered=%d enqueued=%d provider_drops=%d gateway_drops=%d queue_count=%d queue_bytes=%d",
			err, path.expectedSourceUDP.Load(), path.fixture.sentPackets.Load(), path.sourceUDP.Load(),
			path.session.mediaIngress.Load(), path.session.mediaRecovered.Load(), path.session.mediaEnqueued.Load(),
			path.session.Telemetry().MediaDrops, path.server.Metrics().MediaDrops,
			path.session.mediaQueueMaximum.Load(), path.session.mediaQueueMaximumBytes.Load())
	}
	for {
		observationMu.Lock()
		observed := completed
		observationMu.Unlock()
		if observed == totalFrames {
			break
		}
		select {
		case <-ctx.Done():
			t.Fatalf("media observer completed %d frames, want %d: %v", observed, totalFrames, ctx.Err())
		default:
			runtime.Gosched()
		}
	}

	if source := path.expectedSourceUDP.Load(); source != path.fixture.sentPackets.Load() || source != path.sourceUDP.Load() || source != path.session.mediaIngress.Load() {
		t.Fatalf("source accounting expected=%d sent=%d source=%d ingress=%d", source, path.fixture.sentPackets.Load(), path.sourceUDP.Load(), path.session.mediaIngress.Load())
	}
	metrics := path.server.Metrics()
	observationMu.Lock()
	completedFrames := completed
	queueDelay := maximumQueueDelay
	observationMu.Unlock()
	if path.backend.setups.Load() != 1 || path.backend.opens.Load() != 1 || completedFrames != totalFrames ||
		path.session.mediaRecovered.Load() != totalFrames || path.session.mediaEnqueued.Load() != totalFrames ||
		metrics.ProcessingSamples != totalFrames || metrics.MediaPackets != expectedVersePackets ||
		path.server.pacer.reservations.Load() != expectedVersePackets ||
		path.session.Telemetry().MediaDrops != 0 || path.server.Metrics().MediaDrops != 0 {
		t.Fatalf("media accounting setup=%d open=%d completed=%d recovered=%d enqueued=%d processing=%d media=%d pacer=%d provider_drops=%d gateway_drops=%d",
			path.backend.setups.Load(), path.backend.opens.Load(), completedFrames,
			path.session.mediaRecovered.Load(), path.session.mediaEnqueued.Load(), metrics.ProcessingSamples,
			metrics.MediaPackets, path.server.pacer.reservations.Load(), path.session.Telemetry().MediaDrops, metrics.MediaDrops)
	}
	if path.session.mediaQueueMaximum.Load() > nativeApolloVideoQueuePackets ||
		path.session.mediaQueueMaximumBytes.Load() > nativeApolloVideoQueueBytes || queueDelay > nativeApolloVideoQueueLatency {
		t.Fatalf("queue bounds count=%d bytes=%d residence=%s", path.session.mediaQueueMaximum.Load(), path.session.mediaQueueMaximumBytes.Load(), queueDelay)
	}
	t.Logf("three-stall public path: frames=%d source=%d ingress=%d recovered=%d enqueued=%d drops=%d/%d queue=%d/%d residence=%s verse_packets=%d",
		totalFrames, path.sourceUDP.Load(), path.session.mediaIngress.Load(), path.session.mediaRecovered.Load(),
		path.session.mediaEnqueued.Load(), path.session.Telemetry().MediaDrops, metrics.MediaDrops,
		path.session.mediaQueueMaximum.Load(), path.session.mediaQueueMaximumBytes.Load(), queueDelay, metrics.MediaPackets)

	path.Close()
	select {
	case <-path.session.readDone:
	case <-time.After(time.Second):
		t.Fatal("native media workers did not stop")
	}
}

func TestQualificationLateSourceBatchDoesNotCollapseThroughPublicPath(t *testing.T) {
	profile := qualificationMediaProfiles()[2]
	path := newQualificationPath(t, profile, profile.BitrateKbps)
	defer path.Close()

	const shardCount = 662
	payload := qualificationFramePayload(profile, 0)
	if len(payload) != 666_664 {
		t.Fatalf("4K60 keyframe bytes = %d, want 666664", len(payload))
	}
	packets := qualificationSourceVideoPackets(t, path.key, 1, payload)
	if len(packets) != shardCount {
		t.Fatalf("source packet count = %d, want %d", len(packets), shardCount)
	}

	var batchIndices []int
	var batchStarts []time.Time
	var firstBatch time.Time
	path.fixture.observeVideoBatch = func(index int, at time.Time) {
		if firstBatch.IsZero() {
			firstBatch = at
		}
		batchIndices = append(batchIndices, index)
		batchStarts = append(batchStarts, at)
	}
	stalled := false
	path.fixture.beforeVideoBatch = func(ctx context.Context, index int) error {
		if index != 63 || stalled {
			return nil
		}
		stalled = true
		return qualificationWaitContext(ctx, firstBatch.Add(5*time.Millisecond))
	}

	trace, _, err := path.traverse(t, payload)
	if err != nil {
		t.Fatal(err)
	}
	if !trace.NativeSetup || !trace.NativeOpen || !trace.NativeUDPIngress || !trace.ApolloRecovered ||
		!trace.ProductionQueue || !trace.ProductionMediaLoop || !trace.ProductionPacer ||
		!trace.VerseQUIC || !trace.PublicClientDecode || !trace.PayloadPreserved {
		t.Fatalf("late-batch path skipped production stages: %#v", trace)
	}
	if got := path.sourceUDP.Load(); got != shardCount {
		t.Fatalf("source UDP count = %d, want %d", got, shardCount)
	}
	if len(batchStarts) != (shardCount+62)/63 {
		t.Fatalf("batch count = %d, want %d", len(batchStarts), (shardCount+62)/63)
	}
	packetsPerMillisecond, _ := qualificationApolloVideoPacing(apolloVideoRawPacketSize)
	for index := 1; index < len(batchStarts); index++ {
		previousPackets := min(63, shardCount-batchIndices[index-1])
		minimum := qualificationApolloVideoOffset(previousPackets, packetsPerMillisecond)
		actual := batchStarts[index].Sub(batchStarts[index-1])
		if actual < minimum {
			t.Fatalf("batch %d at packet %d started %s after packet %d, before raw serialization interval %s; starts=%v",
				index, batchIndices[index], actual, batchIndices[index-1], minimum, batchStarts)
		}
	}
	t.Logf("late-batch public path: packets=%d starts=%v ingress=%d recovered=%d enqueued=%d",
		path.sourceUDP.Load(), batchStarts, path.session.mediaIngress.Load(),
		path.session.mediaRecovered.Load(), path.session.mediaEnqueued.Load())
}

func TestQualificationBlockedVideoAEADDoesNotBlockProviderIngress(t *testing.T) {
	profile := qualificationMediaProfiles()[2]
	blockCtx, cancelBlock := context.WithCancel(context.Background())
	defer cancelBlock()
	blocked := make(chan struct{})
	release := make(chan struct{})
	var releaseOnce sync.Once
	releaseAEAD := func() { releaseOnce.Do(func() { close(release) }) }
	t.Cleanup(releaseAEAD)
	native := NewNativeApolloBackend()
	native.configureMedia = func(media *apolloMediaCodec) {
		media.aead = &qualificationBlockingAEAD{
			AEAD: media.aead, ctx: blockCtx, blocked: blocked, release: release,
		}
	}
	path := newQualificationPathWithNativeBackend(t, profile, profile.BitrateKbps, nil, native)
	t.Cleanup(func() {
		path.Close()
		select {
		case <-path.session.readDone:
		case <-time.After(time.Second):
			t.Error("native media workers did not stop during qualification cleanup")
		}
	})

	payload := qualificationFramePayload(profile, 0)
	packets := qualificationSourceVideoPackets(t, path.key, 1, payload)
	if len(payload) != 666_664 || len(packets) != 662 {
		t.Fatalf("4K60 keyframe = %d bytes/%d shards, want 666664/662", len(payload), len(packets))
	}

	beforeMetrics := path.server.Metrics()
	beforeIngress := path.session.mediaIngress.Load()
	beforeRecovered := path.session.mediaRecovered.Load()
	beforeEnqueued := path.session.mediaEnqueued.Load()
	beforePacer := path.server.pacer.reservations.Load()
	if _, err := path.emit(t, payload); err != nil {
		t.Fatal(err)
	}
	select {
	case <-blocked:
	case <-time.After(time.Second):
		t.Fatal("video AEAD did not block")
	}
	if sent := path.fixture.sentPackets.Load(); sent != 662 {
		t.Fatalf("fixture sent packets = %d, want 662", sent)
	}
	deadline := time.NewTimer(250 * time.Millisecond)
	defer deadline.Stop()
	for path.session.mediaIngress.Load()-beforeIngress != 662 {
		select {
		case <-deadline.C:
			t.Fatalf("blocked-AEAD ingress = %d, want 662 after 662 successful fixture writes",
				path.session.mediaIngress.Load()-beforeIngress)
		default:
			runtime.Gosched()
		}
	}

	releaseAEAD()
	recovered, err := path.receivePayload(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	afterMetrics := path.server.Metrics()
	stageDeadline := time.Now().Add(2 * time.Second)
	for (path.session.mediaRecovered.Load() <= beforeRecovered ||
		path.session.mediaEnqueued.Load() <= beforeEnqueued ||
		afterMetrics.ProcessingSamples <= beforeMetrics.ProcessingSamples ||
		afterMetrics.MediaPackets <= beforeMetrics.MediaPackets) && time.Now().Before(stageDeadline) {
		runtime.Gosched()
		afterMetrics = path.server.Metrics()
	}
	if path.backend.setups.Load() != 1 || path.backend.opens.Load() != 1 ||
		path.session.mediaRecovered.Load()-beforeRecovered != 1 ||
		path.session.mediaEnqueued.Load()-beforeEnqueued != 1 ||
		afterMetrics.ProcessingSamples <= beforeMetrics.ProcessingSamples ||
		path.server.pacer.reservations.Load() <= beforePacer ||
		afterMetrics.MediaPackets <= beforeMetrics.MediaPackets || !bytes.Equal(recovered, payload) {
		t.Fatalf("blocked-AEAD public path: setup=%d open=%d ingress=%d recovered=%d enqueued=%d processing=%d pacer=%d media=%d payload=%t",
			path.backend.setups.Load(), path.backend.opens.Load(), path.session.mediaIngress.Load()-beforeIngress,
			path.session.mediaRecovered.Load()-beforeRecovered, path.session.mediaEnqueued.Load()-beforeEnqueued,
			afterMetrics.ProcessingSamples-beforeMetrics.ProcessingSamples,
			path.server.pacer.reservations.Load()-beforePacer, afterMetrics.MediaPackets-beforeMetrics.MediaPackets,
			bytes.Equal(recovered, payload))
	}
}

func TestQualificationProcessingRetainsProviderIngressDiagnostics(t *testing.T) {
	profile := qualificationMediaProfiles()[2]
	path := newQualificationProcessingPath(t, profile, profile.BitrateKbps)
	defer path.Close()

	payload := qualificationFramePayload(profile, 0)
	packets := qualificationSourceVideoPackets(t, path.key, 1, payload)
	if len(payload) != 666_664 || len(packets) != 662 {
		t.Fatalf("4K60 keyframe = %d bytes/%d shards, want 666664/662", len(payload), len(packets))
	}
	before, err := path.process.snapshot()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := path.emit(t, payload); err != nil {
		t.Fatal(err)
	}
	recovered, err := path.receivePayload(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	diagnostics := path.processingDiagnostics(1)
	if diagnostics.SnapshotError != "" {
		t.Fatal(diagnostics.SnapshotError)
	}
	after := diagnostics.Gateway
	batches := diagnostics.BatchHistory
	if diagnostics.ProcessedFrames != 1 || diagnostics.SourceFrames != 1 ||
		diagnostics.ExpectedWritesThroughLastFrame != 662 ||
		diagnostics.FixtureSentPackets != 662 || diagnostics.SourceUDP != 662 ||
		after.MediaIngress-before.MediaIngress != 662 ||
		after.MediaRecovered-before.MediaRecovered != 1 ||
		after.MediaEnqueued-before.MediaEnqueued != 1 || !bytes.Equal(recovered, payload) {
		t.Fatalf("provider ingress diagnostics: sent=%d source=%d ingress=%d recovered=%d enqueued=%d payload=%t",
			diagnostics.FixtureSentPackets, diagnostics.SourceUDP, after.MediaIngress-before.MediaIngress,
			after.MediaRecovered-before.MediaRecovered, after.MediaEnqueued-before.MediaEnqueued,
			bytes.Equal(recovered, payload))
	}
	if len(batches) != 11 {
		t.Fatalf("provider batch history = %d entries, want 11", len(batches))
	}
	for index := range batches {
		wantPacket := index * 63
		if batches[index].ProviderFrame != 1 || batches[index].PacketWithinFrame != wantPacket ||
			batches[index].SourcePacket != uint64(wantPacket) {
			t.Fatalf("provider batch %d = %#v, want frame 1 source/within %d", index, batches[index], wantPacket)
		}
		if index > 0 {
			previousPackets := min(63, 662-batches[index-1].PacketWithinFrame)
			minimum := qualificationApolloVideoOffset(previousPackets, 96)
			if batches[index].StartedAfter-batches[index-1].StartedAfter < minimum {
				t.Fatalf("provider batch %d collapsed: %#v", index, batches)
			}
		}
	}
	if (runtime.GOOS == "darwin" || runtime.GOOS == "linux") && !after.VideoReceiveBufferAvailable {
		t.Fatal("provider video SO_RCVBUF unavailable on a supported diagnostic platform")
	}
	if after.VideoReceiveBufferAvailable && after.VideoReceiveBuffer <= 0 {
		t.Fatalf("provider video SO_RCVBUF = %d available=%t, want measured >0",
			after.VideoReceiveBuffer, after.VideoReceiveBufferAvailable)
	}
	if after.KernelDropsAvailable && after.KernelDrops != 0 {
		t.Fatalf("provider UDP kernel drops = %d, want 0", after.KernelDrops)
	}
	t.Logf("provider ingress diagnostics: %#v", diagnostics)
}

func TestQualificationActualEmissionDoesNotCatchUpAfterPreWritePause(t *testing.T) {
	key := bytes.Repeat([]byte{0x4d}, 16)
	packets := qualificationSourceVideoPackets(t, key, 1, make([]byte, 189*apolloVideoShardPayloadSize-8))
	receiver, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
	if err != nil {
		t.Fatal(err)
	}
	defer receiver.Close()
	sender, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
	if err != nil {
		t.Fatal(err)
	}
	defer sender.Close()
	fixture := &qualificationApolloFixture{video: sender, failures: make(chan error, 1)}
	remote := *receiver.LocalAddr().(*net.UDPAddr)
	fixture.videoRemote.Store(&remote)
	stalled := false
	fixture.beforeVideoFirstWrite = func(ctx context.Context, packetWithinFrame int) error {
		if packetWithinFrame != 63 || stalled {
			return nil
		}
		stalled = true
		return qualificationWaitContext(ctx, time.Now().Add(5*time.Millisecond))
	}
	var emitted []time.Time
	fixture.observeVideoBatch = func(_ int, at time.Time) { emitted = append(emitted, at) }
	drained := make(chan struct{})
	go func() {
		defer close(drained)
		buffer := make([]byte, 2048)
		for range len(packets) {
			if _, _, readErr := receiver.ReadFromUDP(buffer); readErr != nil {
				return
			}
		}
	}()
	if err := fixture.sendVideo(context.Background(), 1, packets); err != nil {
		t.Fatal(err)
	}
	select {
	case <-drained:
	case <-time.After(time.Second):
		t.Fatal("actual-emission receiver did not drain")
	}
	if len(emitted) != 3 {
		t.Fatalf("actual emitted batch starts = %d, want 3", len(emitted))
	}
	minimum := qualificationApolloVideoOffset(63, 96)
	if actual := emitted[2].Sub(emitted[1]); actual < minimum {
		t.Fatalf("post-write batch 126 started %s after batch 63, before %s; emitted=%v", actual, minimum, emitted)
	}
	t.Logf("post-write emitted batch starts: %v", emitted)
}

func TestQualificationProcessingDiagnosticsFormatUsesMonotonicOffsets(t *testing.T) {
	diagnostics := qualificationProcessingDiagnostics{
		BatchHistory: []qualificationVideoBatchObservation{{StartedAfter: 1234567 * time.Nanosecond}},
	}
	failureText := fmt.Errorf("qualification failed; diagnostics=%#v", diagnostics).Error()
	if !strings.Contains(failureText, "StartedAfter") || !strings.Contains(failureText, "1234567") ||
		strings.Contains(failureText, "time.Date(") {
		t.Fatalf("processing failure text does not retain explicit monotonic offsets: %s", failureText)
	}
}