VerseVDI-Data-Plane/gateway/qualification_contract_test.go

package gateway

import (
	"bytes"
	"compress/gzip"
	"context"
	"crypto/cipher"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"net"
	"os"
	"path/filepath"
	"reflect"
	"runtime"
	"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:] {
		observation, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0],
			qualificationImpairmentPacketCount, filepath.Join(t.TempDir(), 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)
		}
	}
}

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
	}

	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)
		}
		observation := qualificationCapacityStepObservation(wireDeliveries, stepAt[reduction], profile, reduction)
		if observation.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, observation.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, observation.Convergence, maximum, wireBytesPerSecond*5*105/100)
	}
}

func TestQualificationUsesPublicQUICAndProductionPacer(t *testing.T) {
	qualificationTraverseProfiles(t, qualificationMediaProfiles())
	evidence, err := qualificationPacerEvidence(t, filepath.Join(t.TempDir(), "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)
		}
	}
}

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)
	}
}