test(gateway): retain raw wire qualification evidence

.gitea/workflows/verify.yml

@@ -41,6 +41,7 @@ jobs:
           name: verse-gateway-linux-${{ gitea.sha }}
           path: dist/*
           if-no-files-found: error
-          retention-days: 30
+          # Gitea 1.27 floors the positive upload delay; request 31 days to retain at least 30 elapsed days.
+          retention-days: 31
       - name: Verify clean checkout
         run: git diff --exit-code

gateway/apollo_native.go

@@ -33,6 +33,9 @@ const (
 	nativeApolloVideoQueueLatency = 250 * time.Millisecond
 	nativeApolloAudioQueuePackets = 16
 	nativeApolloEventQueuePackets = 16
+	nativeApolloVideoIngressSlots = 2048
+	nativeApolloVideoPacketBytes  = apolloVideoHeaderSize + apolloVideoRawPacketSize
+	nativeApolloVideoReadBuffer   = nativeApolloVideoIngressSlots * nativeApolloVideoPacketBytes
 )
 
 // NativeApolloBackend keeps provider sockets inside the gateway process. The
@@ -41,8 +44,9 @@ const (
 type NativeApolloBackend struct {
 	Dialer *net.Dialer
 
-	mu      sync.Mutex
-	pending map[string]*apolloRTSPSetup
+	mu             sync.Mutex
+	pending        map[string]*apolloRTSPSetup
+	configureMedia func(*apolloMediaCodec)
 }
 
 func NewNativeApolloBackend() *NativeApolloBackend {
@@ -268,7 +272,7 @@ func (b *NativeApolloBackend) Open(ctx context.Context, request LaunchRequest, _
 	if !ok || setup == nil {
 		return nil, ErrProviderDisconnected
 	}
-	return newNativeApolloProviderSession(ctx, setup)
+	return newNativeApolloProviderSession(ctx, setup, b.configureMedia)
 }
 
 func readBounded(reader io.Reader, max int) ([]byte, error) {
@@ -336,7 +340,7 @@ func newNativeApolloSession(sessionID string) *nativeApolloSession {
 	}
 }
 
-func newNativeApolloProviderSession(ctx context.Context, setup *apolloRTSPSetup) (*nativeApolloSession, error) {
+func newNativeApolloProviderSession(ctx context.Context, setup *apolloRTSPSetup, configureMedia func(*apolloMediaCodec)) (*nativeApolloSession, error) {
 	if setup == nil || len(setup.streamKey) != 16 || setup.controlPort == 0 || setup.audioPort == 0 || setup.videoPort == 0 {
 		return nil, ErrProviderMalformed
 	}
@@ -360,6 +364,9 @@ func newNativeApolloProviderSession(ctx context.Context, setup *apolloRTSPSetup)
 		peer.close(err)
 		return nil, err
 	}
+	if configureMedia != nil {
+		configureMedia(media)
+	}
 	session := newNativeApolloSession(setup.sessionID)
 	managementClient, err := newPinnedApolloHTTPClient(setup.providerWork)
 	if err != nil {
@@ -391,6 +398,12 @@ func newNativeApolloProviderSession(ctx context.Context, setup *apolloRTSPSetup)
 		peer.close(err)
 		return nil, err
 	}
+	if err := videoConn.SetReadBuffer(nativeApolloVideoReadBuffer); err != nil {
+		_ = audioConn.Close()
+		_ = videoConn.Close()
+		peer.close(err)
+		return nil, fmt.Errorf("set Apollo video receive buffer: %w", err)
+	}
 	if _, err := audioConn.Write(apolloMediaPing(setup.audioPing, 1)); err != nil {
 		_ = audioConn.Close()
 		_ = videoConn.Close()
@@ -896,19 +909,20 @@ func (s *nativeApolloSession) readUDPMedia() {
 		s.closeMediaChannels()
 		return
 	}
-	var readers sync.WaitGroup
-	readers.Add(2)
-	read := func(conn *net.UDPConn, output chan ProviderMedia, video bool) {
-		defer readers.Done()
+	videoIngress := newNativeApolloVideoIngress()
+	var workers sync.WaitGroup
+	workers.Add(3)
+	go func() {
+		defer workers.Done()
 		buffer := make([]byte, apolloMediaMaximumPacket+1)
 		for {
 			if s.mediaQuiesced.Load() {
 				return
 			}
-			if err := conn.SetReadDeadline(time.Now().Add(250 * time.Millisecond)); err != nil {
+			if err := s.audioConn.SetReadDeadline(time.Now().Add(250 * time.Millisecond)); err != nil {
 				return
 			}
-			count, err := conn.Read(buffer)
+			count, err := s.audioConn.Read(buffer)
 			receivedAt := time.Now()
 			if err != nil {
 				if networkErr, ok := err.(net.Error); ok && networkErr.Timeout() {
@@ -928,45 +942,121 @@ func (s *nativeApolloSession) readUDPMedia() {
 				return
 			}
 			s.mediaIngress.Add(1)
-			var payloads [][]byte
-			if video {
-				shard, openErr := s.media.OpenVideo(buffer[:count])
-				if openErr != nil {
-					continue
-				}
-				payload, err := s.videoFEC.Add(shard)
-				if err != nil || len(payload) == 0 {
-					continue
-				}
-				payloads = [][]byte{payload}
-			} else {
-				shard, openErr := s.media.OpenAudio(buffer[:count])
-				if openErr != nil {
-					continue
-				}
-				var evicted bool
-				payloads, evicted, err = s.audioFEC.Add(s.media, shard)
-				if evicted {
-					s.mediaDrops.Add(1)
-				}
+			shard, openErr := s.media.OpenAudio(buffer[:count])
+			if openErr != nil {
+				continue
+			}
+			payloads, evicted, err := s.audioFEC.Add(s.media, shard)
+			if evicted {
+				s.mediaDrops.Add(1)
 			}
 			if err != nil {
 				continue
 			}
 			for _, payload := range payloads {
-				s.enqueueMedia(output, payload, receivedAt)
+				s.enqueueMedia(s.audio, payload, receivedAt)
 			}
 		}
-	}
-	go read(s.audioConn, s.audio, false)
-	go read(s.videoConn, s.video, true)
+	}()
 	go func() {
-		readers.Wait()
+		defer workers.Done()
+		s.drainApolloVideo(videoIngress)
+	}()
+	go func() {
+		defer workers.Done()
+		s.processApolloVideo(videoIngress)
+	}()
+	go func() {
+		workers.Wait()
 		close(s.readDone)
 		s.closeMediaChannels()
 	}()
 }
 
+type nativeApolloVideoIngressSlot struct {
+	packet     [apolloMediaMaximumPacket + 1]byte
+	count      int
+	receivedAt time.Time
+}
+
+type nativeApolloVideoIngress struct {
+	slots   [nativeApolloVideoIngressSlots]nativeApolloVideoIngressSlot
+	free    chan uint16
+	ready   chan uint16
+	scratch [apolloMediaMaximumPacket + 1]byte
+}
+
+func newNativeApolloVideoIngress() *nativeApolloVideoIngress {
+	ingress := &nativeApolloVideoIngress{
+		free:  make(chan uint16, nativeApolloVideoIngressSlots),
+		ready: make(chan uint16, nativeApolloVideoIngressSlots),
+	}
+	for index := range nativeApolloVideoIngressSlots {
+		ingress.free <- uint16(index)
+	}
+	return ingress
+}
+
+func (s *nativeApolloSession) drainApolloVideo(ingress *nativeApolloVideoIngress) {
+	defer close(ingress.ready)
+	for {
+		if s.mediaQuiesced.Load() {
+			return
+		}
+		select {
+		case index := <-ingress.free:
+			slot := &ingress.slots[index]
+			count, err := s.videoConn.Read(slot.packet[:])
+			receivedAt := time.Now()
+			if err != nil {
+				ingress.free <- index
+				return
+			}
+			if count > apolloMediaMaximumPacket {
+				ingress.free <- index
+				continue
+			}
+			if s.mediaQuiesced.Load() {
+				ingress.free <- index
+				return
+			}
+			s.mediaIngress.Add(1)
+			slot.count, slot.receivedAt = count, receivedAt
+			ingress.ready <- index
+		default:
+			count, err := s.videoConn.Read(ingress.scratch[:])
+			if err != nil {
+				return
+			}
+			if count > apolloMediaMaximumPacket {
+				continue
+			}
+			if s.mediaQuiesced.Load() {
+				return
+			}
+			s.mediaIngress.Add(1)
+			s.mediaDrops.Add(1)
+		}
+	}
+}
+
+func (s *nativeApolloSession) processApolloVideo(ingress *nativeApolloVideoIngress) {
+	for index := range ingress.ready {
+		slot := &ingress.slots[index]
+		if !s.mediaQuiesced.Load() {
+			shard, err := s.media.OpenVideo(slot.packet[:slot.count])
+			if err == nil {
+				payload, fecErr := s.videoFEC.Add(shard)
+				if fecErr == nil && len(payload) > 0 {
+					s.enqueueMedia(s.video, payload, slot.receivedAt)
+				}
+			}
+		}
+		slot.count, slot.receivedAt = 0, time.Time{}
+		ingress.free <- index
+	}
+}
+
 func pushLatest[T any](channel chan T, payload T) bool {
 	select {
 	case channel <- payload:

gateway/apollo_native_test.go

@@ -17,8 +17,10 @@ import (
 	"net"
 	"net/http"
 	"net/http/httptest"
+	"runtime"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -760,6 +762,34 @@ func TestNativeApolloSessionRelaysOnlyAuthenticatedEncodedUDPMedia(t *testing.T)
 	if drops := session.Telemetry().MediaDrops; drops < 2 {
 		t.Fatalf("stale FEC eviction drops = %d, want at least 2", drops)
 	}
+	beforeIngress := session.mediaIngress.Load()
+	beforeDrops := session.mediaDrops.Load()
+	if _, err := videoServer.WriteToUDP(make([]byte, apolloMediaMaximumPacket+2), videoClient.LocalAddr().(*net.UDPAddr)); err != nil {
+		t.Fatal(err)
+	}
+	marker := []byte{0xde, 0xad, 0xbe, 0xef}
+	markerPayload := make([]byte, apolloVideoShardPayloadSize)
+	markerPayload[0], markerPayload[3] = 0x01, 0x01
+	binary.LittleEndian.PutUint16(markerPayload[4:6], uint16(8+len(marker)))
+	copy(markerPayload[8:], marker)
+	markerPacket := sourceEncryptVideoRaw(t, key, sourceShapedVideoRaw(44, 103, 103, 0x07, 1, 0, 0, markerPayload), "0123456789dV")
+	if _, err := videoServer.WriteToUDP(markerPacket, videoClient.LocalAddr().(*net.UDPAddr)); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case media := <-session.Video():
+		if !bytes.Equal(media.Payload, marker) {
+			t.Fatalf("post-oversize marker relay = %x, want %x", media.Payload, marker)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("post-oversize marker was not relayed")
+	}
+	if got := session.mediaIngress.Load() - beforeIngress; got != 1 {
+		t.Fatalf("post-oversize video ingress = %d, want 1 valid marker", got)
+	}
+	if got := session.mediaDrops.Load(); got != beforeDrops {
+		t.Fatalf("post-oversize video drops = %d, want unchanged %d", got, beforeDrops)
+	}
 	terminateCtx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
 	if err := session.Terminate(terminateCtx); err != nil {
@@ -767,6 +797,166 @@ func TestNativeApolloSessionRelaysOnlyAuthenticatedEncodedUDPMedia(t *testing.T)
 	}
 }
 
+func TestNativeApolloVideoIngressSaturationIsBoundedAndAccounted(t *testing.T) {
+	if nativeApolloVideoIngressSlots != 2048 || nativeApolloVideoReadBuffer != 2_195_456 {
+		t.Fatalf("video ingress bounds = %d slots/%d bytes, want 2048/2195456", nativeApolloVideoIngressSlots, nativeApolloVideoReadBuffer)
+	}
+	ingress := newNativeApolloVideoIngress()
+	if len(ingress.slots) != 2048 || len(ingress.free) != 2048 || cap(ingress.ready) != 2048 || len(ingress.slots[0].packet) != apolloMediaMaximumPacket+1 {
+		t.Fatalf("video ingress pool = slots:%d free:%d ready-cap:%d packet:%d",
+			len(ingress.slots), len(ingress.free), cap(ingress.ready), len(ingress.slots[0].packet))
+	}
+
+	key := []byte("0123456789abcdef")
+	session, audioServer, videoServer := newNativeApolloMediaTestSession(t, key)
+	defer audioServer.Close()
+	defer videoServer.Close()
+	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) }) }
+	defer releaseAEAD()
+	session.media.aead = &qualificationBlockingAEAD{
+		AEAD: session.media.aead, ctx: blockCtx, blocked: blocked, release: release,
+	}
+	go session.readUDPMedia()
+
+	packet := sourceShapedEncryptedVideoPacket(t, key, []byte{1})
+	if _, err := videoServer.WriteToUDP(packet, session.videoConn.LocalAddr().(*net.UDPAddr)); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-blocked:
+	case <-time.After(time.Second):
+		t.Fatal("video processor did not block")
+	}
+	waitIngress := func(want uint64) {
+		t.Helper()
+		deadline := time.NewTimer(time.Second)
+		defer deadline.Stop()
+		for session.mediaIngress.Load() < want {
+			select {
+			case <-deadline.C:
+				t.Fatalf("video ingress = %d, want at least %d", session.mediaIngress.Load(), want)
+			default:
+				runtime.Gosched()
+			}
+		}
+	}
+	waitIngress(1)
+	const overflow = 33
+	sent := uint64(1)
+	for remaining := nativeApolloVideoIngressSlots - 1 + overflow; remaining > 0; {
+		batch := min(32, remaining)
+		for range batch {
+			if _, err := videoServer.WriteToUDP([]byte{1}, session.videoConn.LocalAddr().(*net.UDPAddr)); err != nil {
+				t.Fatal(err)
+			}
+		}
+		sent += uint64(batch)
+		waitIngress(sent)
+		remaining -= batch
+	}
+	if got := session.mediaDrops.Load(); got != overflow {
+		t.Fatalf("saturated video ingress drops = %d, want %d", got, overflow)
+	}
+
+	session.quiesceMedia()
+	cancelBlock()
+	select {
+	case <-session.readDone:
+	case <-time.After(time.Second):
+		t.Fatal("video ingress workers did not stop")
+	}
+}
+
+func TestNativeApolloTerminateStopsActiveVideoDrainAndProcessor(t *testing.T) {
+	key := []byte("0123456789abcdef")
+	session, audioServer, videoServer := newNativeApolloMediaTestSession(t, key)
+	defer audioServer.Close()
+	defer videoServer.Close()
+	blockCtx, cancelBlock := context.WithCancel(context.Background())
+	t.Cleanup(cancelBlock)
+	blocked := make(chan struct{})
+	session.media.aead = &qualificationBlockingAEAD{
+		AEAD: session.media.aead, ctx: blockCtx, blocked: blocked, release: make(chan struct{}),
+	}
+	go session.readUDPMedia()
+	if _, err := videoServer.WriteToUDP(sourceShapedEncryptedVideoPacket(t, key, []byte{1}), session.videoConn.LocalAddr().(*net.UDPAddr)); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-blocked:
+	case <-time.After(time.Second):
+		t.Fatal("video processor did not block")
+	}
+
+	terminateCtx, cancelTerminate := context.WithTimeout(context.Background(), time.Second)
+	defer cancelTerminate()
+	terminated := make(chan error, 1)
+	go func() { terminated <- session.Terminate(terminateCtx) }()
+	cancelBlock()
+	if err := <-terminated; err != nil {
+		t.Fatalf("Terminate() error = %v", err)
+	}
+	select {
+	case <-session.readDone:
+	default:
+		t.Fatal("Terminate returned before video drain and processor stopped")
+	}
+	select {
+	case _, ok := <-session.Video():
+		if ok {
+			t.Fatal("video channel remained open after worker shutdown")
+		}
+	default:
+		t.Fatal("video channel was not closed after worker shutdown")
+	}
+}
+
+func newNativeApolloMediaTestSession(t *testing.T, key []byte) (*nativeApolloSession, *net.UDPConn, *net.UDPConn) {
+	t.Helper()
+	audioServer, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
+	if err != nil {
+		t.Fatal(err)
+	}
+	videoServer, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
+	if err != nil {
+		_ = audioServer.Close()
+		t.Fatal(err)
+	}
+	audioClient, err := net.DialUDP("udp", nil, audioServer.LocalAddr().(*net.UDPAddr))
+	if err != nil {
+		_ = audioServer.Close()
+		_ = videoServer.Close()
+		t.Fatal(err)
+	}
+	videoClient, err := net.DialUDP("udp", nil, videoServer.LocalAddr().(*net.UDPAddr))
+	if err != nil {
+		_ = audioClient.Close()
+		_ = audioServer.Close()
+		_ = videoServer.Close()
+		t.Fatal(err)
+	}
+	session := newNativeApolloSession("session-video-ingress")
+	session.media, err = newApolloMediaCodec(key, 1)
+	if err != nil {
+		_ = audioClient.Close()
+		_ = videoClient.Close()
+		_ = audioServer.Close()
+		_ = videoServer.Close()
+		t.Fatal(err)
+	}
+	session.audioConn, session.videoConn = audioClient, videoClient
+	t.Cleanup(func() {
+		_ = audioClient.Close()
+		_ = videoClient.Close()
+	})
+	return session, audioServer, videoServer
+}
+
 func TestNativeApolloTerminateReleasesPressedProviderInput(t *testing.T) {
 	server, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
 	if err != nil {

gateway/fair_pacer_test.go

@@ -41,6 +41,83 @@ func TestFairPacerBoundsCatchupAfterHostStall(t *testing.T) {
 	if next.Before(resumed.Add(-fairPacerMaximumCatchup)) || next.After(resumed.Add(10*time.Millisecond)) {
 		t.Fatalf("post-stall reservation = %s, want bounded catchup near %s", next, resumed)
 	}
+	pacer.mu.Lock()
+	debt := pacer.flows["one"].debt
+	pacer.mu.Unlock()
+	if debt <= 0 || debt > nativeApolloVideoQueueLatency-fairPacerMaximumCatchup {
+		t.Fatalf("post-stall debt = %s, want bounded valid schedule debt", debt)
+	}
+}
+
+func TestFairPacerRepaysBoundedDebtAfterHostStall(t *testing.T) {
+	start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
+	pacer := newFairPacer(8000)
+	next := make(map[string]time.Time)
+	deliveries := runSyntheticPacerWithStall(
+		pacer, start, start.Add(6*time.Second), []string{"one"}, next,
+		start.Add(time.Second), 100*time.Millisecond,
+	)
+	if total := syntheticDeliveryBytes(deliveries); total < 5_990_000 || total > 6_010_000 {
+		t.Fatalf("post-stall delivery bytes = %d, want nominal throughput after bounded debt repayment", total)
+	}
+	pacer.mu.Lock()
+	remaining := pacer.flows["one"].debt
+	pacer.mu.Unlock()
+	if remaining != 0 {
+		t.Fatalf("post-stall debt = %s after repayment, want zero", remaining)
+	}
+	assertSyntheticCap(t, deliveries, 1_000_000)
+	t.Logf("single-flow debt repaid: bytes=%d remaining=%s", syntheticDeliveryBytes(deliveries), remaining)
+}
+
+func TestFairPacerRepaysSimultaneousEightFlowDebtAcrossCapacitySteps(t *testing.T) {
+	start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
+	flows := []string{"one", "two", "three", "four", "five", "six", "seven", "eight"}
+
+	tests := []struct {
+		name           string
+		kbps           int64
+		bytesPerSecond int64
+		minimumBytes   int64
+	}{
+		{name: "baseline", kbps: 8000, bytesPerSecond: 1_000_000, minimumBytes: 9_980_000},
+		{name: "quarter", kbps: 6000, bytesPerSecond: 750_000, minimumBytes: 7_480_000},
+		{name: "half", kbps: 4000, bytesPerSecond: 500_000, minimumBytes: 4_980_000},
+	}
+	for _, test := range tests {
+		pacer := newFairPacer(8000)
+		next := make(map[string]time.Time, len(flows))
+		_ = runSyntheticPacer(pacer, start, start.Add(time.Second), flows, next)
+		resumed := start.Add(1100 * time.Millisecond)
+		for _, flow := range flows {
+			next[flow] = pacer.reserveAt(resumed, flow, 1000)
+		}
+		assertSyntheticDebt(t, pacer, flows, true)
+		pacer.setKbps(test.kbps)
+		deliveries := runSyntheticPacer(pacer, resumed, resumed.Add(10*time.Second), flows, next)
+		if total := syntheticDeliveryBytes(deliveries); total < test.minimumBytes {
+			t.Fatalf("%s post-stall delivery bytes = %d, want at least %d", test.name, total, test.minimumBytes)
+		}
+		assertSyntheticFairness(t, deliveries, flows)
+		assertSyntheticCap(t, deliveries, test.bytesPerSecond)
+		assertSyntheticDebt(t, pacer, flows, false)
+		t.Logf("%s eight-flow debt repaid: bytes=%d cap=%d", test.name, syntheticDeliveryBytes(deliveries), test.bytesPerSecond*5*105/100)
+	}
+}
+
+func assertSyntheticDebt(t *testing.T, pacer *fairPacer, flows []string, wantDebt bool) {
+	t.Helper()
+	pacer.mu.Lock()
+	defer pacer.mu.Unlock()
+	for _, flow := range flows {
+		debt := pacer.flows[flow].debt
+		if wantDebt && (debt <= 0 || debt > nativeApolloVideoQueueLatency-fairPacerMaximumCatchup) {
+			t.Fatalf("flow %s active debt = %s, want bounded nonzero debt", flow, debt)
+		}
+		if !wantDebt && debt != 0 {
+			t.Fatalf("flow %s debt = %s after repayment, want zero", flow, debt)
+		}
+	}
 }
 
 func runSyntheticPacer(pacer *fairPacer, start, end time.Time, flows []string, next map[string]time.Time) []syntheticPacerDelivery {
@@ -67,6 +144,50 @@ func runSyntheticPacer(pacer *fairPacer, start, end time.Time, flows []string, n
 	}
 }
 
+func runSyntheticPacerWithStall(pacer *fairPacer, start, end time.Time, flows []string, next map[string]time.Time, stallAt time.Time, stall time.Duration) []syntheticPacerDelivery {
+	const packetBytes = 1000
+	for _, flow := range flows {
+		if next[flow].IsZero() {
+			next[flow] = pacer.reserveAt(start, flow, packetBytes)
+		}
+	}
+	now := start
+	stalled := false
+	var deliveries []syntheticPacerDelivery
+	for {
+		flow := ""
+		target := end.Add(time.Nanosecond)
+		for _, candidate := range flows {
+			if next[candidate].Before(target) {
+				flow, target = candidate, next[candidate]
+			}
+		}
+		if target.After(end) {
+			return deliveries
+		}
+		if !stalled && !target.Before(stallAt) {
+			now = stallAt.Add(stall)
+			stalled = true
+		}
+		if now.Before(target) {
+			now = target
+		}
+		if now.After(end) {
+			return deliveries
+		}
+		deliveries = append(deliveries, syntheticPacerDelivery{at: now, flow: flow, bytes: packetBytes})
+		next[flow] = pacer.reserveAt(now, flow, packetBytes)
+	}
+}
+
+func syntheticDeliveryBytes(deliveries []syntheticPacerDelivery) int64 {
+	var total int64
+	for _, delivery := range deliveries {
+		total += delivery.bytes
+	}
+	return total
+}
+
 func assertSyntheticFairness(t *testing.T, deliveries []syntheticPacerDelivery, flows []string) {
 	t.Helper()
 	counts := make(map[string]int64, len(flows))

gateway/gateway_test.go

@@ -940,12 +940,20 @@ func (c *independentGatewayClient) ReceiveFrame(ctx context.Context) (Frame, err
 }
 
 func (c *independentGatewayClient) ReceiveMedia(ctx context.Context) ([]byte, error) {
+	return c.receiveMedia(ctx, nil)
+}
+
+func (c *independentGatewayClient) receiveMedia(ctx context.Context, observe func(time.Time, int)) ([]byte, error) {
 	for {
 		data, err := c.connection.ReceiveDatagram(ctx)
 		if err != nil {
 			return nil, err
 		}
-		payload, complete, err := c.media.Add(data, time.Now())
+		receivedAt := time.Now()
+		if observe != nil {
+			observe(receivedAt, len(data))
+		}
+		payload, complete, err := c.media.Add(data, receivedAt)
 		if err != nil {
 			return nil, err
 		}

gateway/qualification_harness_test.go

@@ -10,6 +10,7 @@ import (
 	"crypto/sha256"
 	"crypto/tls"
 	"encoding/binary"
+	"encoding/csv"
 	"encoding/hex"
 	"encoding/json"
 	"errors"
@@ -31,7 +32,6 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
-	"syscall"
 	"testing"
 	"time"
 
@@ -39,15 +39,19 @@ import (
 )
 
 const (
-	qualificationToolVersion                   = "versevdi-gateway-qualification/v6"
-	qualificationImpairmentQueuePackets        = nativeApolloVideoQueuePackets
-	qualificationImpairmentMaxPackets          = 100_000
-	qualificationImpairmentPacketCount         = 10_000
-	qualificationProcessingLimit               = 5 * time.Millisecond
-	qualificationImpairmentSeed         uint64 = 0x3c6a11ce
-	qualificationClockOverheadMethod           = "median of 1000 batches of 100 monotonic time reads"
-	qualificationGatewayCPUScope               = "isolated gateway subprocess; bounded recorder/control included, fixture and client driver excluded"
-	qualificationResourceMethod                = "RUSAGE_SELF user+system CPU; runtime/metrics heap objects, allocated objects/bytes, and live goroutines sampled once per second"
+	qualificationToolVersion                         = "versevdi-gateway-qualification/v10"
+	qualificationImpairmentQueuePackets              = nativeApolloVideoQueuePackets
+	qualificationImpairmentMaxPackets                = 100_000
+	qualificationImpairmentPacketCount               = 10_000
+	qualificationProcessingLimit                     = 5 * time.Millisecond
+	qualificationImpairmentSeed               uint64 = 0x3c6a11ce
+	qualificationClockOverheadMethod                 = "median of 1000 batches of 100 monotonic time reads"
+	qualificationGatewayCPUScope                     = "isolated gateway subprocess; bounded recorder/control included, fixture and client driver excluded"
+	qualificationResourceMethod                      = "RUSAGE_SELF user+system CPU; runtime/metrics heap objects, allocated objects/bytes, and live goroutines sampled once per second"
+	qualificationApolloVideoRateBitsPerSecond        = 1_000_000_000 * 80 / 100
+	qualificationApolloVideoBatchBytes               = 64 * 1024
+	qualificationApolloVideoBatchPackets             = 64
+	qualificationWireTimebase                        = "monotonic offsets from constrained run start"
 )
 
 type qualificationMediaProfile struct {
@@ -73,20 +77,39 @@ type qualificationPathTrace struct {
 	PayloadPreserved    bool
 }
 
+type qualificationVideoBatchObservation struct {
+	ProviderFrame     uint32
+	SourcePacket      uint64
+	PacketWithinFrame int
+	StartedAfter      time.Duration
+}
+
 type qualificationPath struct {
-	client    *independentGatewayClient
-	server    *Server
-	session   *nativeApolloSession
-	fixture   *qualificationApolloFixture
-	backend   *qualificationTracingBackend
-	process   *qualificationGatewayProcess
-	key       []byte
-	flow      string
-	frame     uint32
-	bootTrace atomic.Bool
-	sourceUDP atomic.Uint64
-	closeOnce sync.Once
-	shutdown  func()
+	client            *independentGatewayClient
+	server            *Server
+	session           *nativeApolloSession
+	fixture           *qualificationApolloFixture
+	backend           *qualificationTracingBackend
+	process           *qualificationGatewayProcess
+	key               []byte
+	flow              string
+	frame             uint32
+	bootTrace         atomic.Bool
+	sourceUDP         atomic.Uint64
+	expectedSourceUDP atomic.Uint64
+	closeOnce         sync.Once
+	shutdown          func()
+}
+
+type qualificationProcessingDiagnostics struct {
+	ProcessedFrames                int64
+	SourceFrames                   uint32
+	ExpectedWritesThroughLastFrame uint64
+	FixtureSentPackets             uint64
+	SourceUDP                      uint64
+	BatchHistory                   []qualificationVideoBatchObservation
+	Gateway                        qualificationGatewayProcessSnapshot
+	SnapshotError                  string
 }
 
 type qualificationImpairmentProfile struct {
@@ -183,13 +206,22 @@ type qualificationImpairmentObservation struct {
 	RawSamples               string                      `json:"raw_samples"`
 	RawSamplesSHA256         string                      `json:"raw_samples_sha256"`
 	RawSamplesBytes          int64                       `json:"raw_samples_bytes"`
+	RawWireSamples           string                      `json:"raw_wire_samples,omitempty"`
+	RawWireSamplesSHA256     string                      `json:"raw_wire_samples_sha256,omitempty"`
+	RawWireSamplesBytes      int64                       `json:"raw_wire_samples_bytes,omitempty"`
+	RawWireRows              int                         `json:"raw_wire_rows,omitempty"`
+	RawWireDeliveryRows      int                         `json:"raw_wire_delivery_rows,omitempty"`
+	RawWireTransitionRows    int                         `json:"raw_wire_transition_rows,omitempty"`
+	RawWireTimebase          string                      `json:"raw_wire_timebase,omitempty"`
 }
 
 type qualificationCapacityStep struct {
-	ReductionPercent  int           `json:"reduction_percent"`
-	Convergence       time.Duration `json:"convergence_ns"`
-	MaximumFiveSecond int64         `json:"maximum_five_second_bytes"`
-	FiveSecondCap     int64         `json:"five_second_cap_bytes"`
+	ReductionPercent    int           `json:"reduction_percent"`
+	TransitionAfter     time.Duration `json:"transition_after_ns"`
+	Convergence         time.Duration `json:"convergence_ns"`
+	MaximumFiveSecond   int64         `json:"maximum_five_second_bytes"`
+	FiveSecondCap       int64         `json:"five_second_cap_bytes"`
+	RecomputationSource string        `json:"recomputation_source"`
 }
 
 type qualificationResourceSample struct {
@@ -206,6 +238,15 @@ type qualificationDeliverySample struct {
 	Bytes int64
 }
 
+type qualificationWireFileEvidence struct {
+	Name           string
+	SHA256         string
+	Bytes          int64
+	Rows           int
+	DeliveryRows   int
+	TransitionRows int
+}
+
 type qualificationFairnessEvidence struct {
 	Evaluation       time.Duration                 `json:"evaluation_ns"`
 	PerFlowBytes     map[string]int64              `json:"per_flow_bytes"`
@@ -411,20 +452,29 @@ func (b *qualificationTracingBackend) session(sessionID string) *nativeApolloSes
 }
 
 type qualificationApolloFixture struct {
-	sessionID   string
-	management  *httptest.Server
-	stream      net.Listener
-	control     *net.UDPConn
-	audio       *net.UDPConn
-	video       *net.UDPConn
-	videoRemote atomic.Pointer[net.UDPAddr]
-	key         atomic.Pointer[[]byte]
-	keyReady    chan []byte
-	failures    chan error
-	closed      atomic.Bool
-	closeOnce   sync.Once
-	sentPackets atomic.Uint64
-	work        protocol.ProviderSessionWork
+	sessionID             string
+	management            *httptest.Server
+	stream                net.Listener
+	control               *net.UDPConn
+	audio                 *net.UDPConn
+	video                 *net.UDPConn
+	videoRemote           atomic.Pointer[net.UDPAddr]
+	key                   atomic.Pointer[[]byte]
+	keyReady              chan []byte
+	failures              chan error
+	closed                atomic.Bool
+	closeOnce             sync.Once
+	sentPackets           atomic.Uint64
+	work                  protocol.ProviderSessionWork
+	videoPaceMu           sync.Mutex
+	videoNext             time.Time
+	beforeVideoBatch      func(context.Context, int) error
+	beforeVideoFirstWrite func(context.Context, int) error
+	observeVideoBatch     func(int, time.Time)
+	videoBatchMu          sync.Mutex
+	videoBatchEpoch       time.Time
+	videoBatchCount       uint64
+	videoBatches          [128]qualificationVideoBatchObservation
 
 	controlImpairmentMu sync.Mutex
 	controlRTT          time.Duration
@@ -432,6 +482,30 @@ type qualificationApolloFixture struct {
 	controlRandom       uint64
 }
 
+func (f *qualificationApolloFixture) videoBatchHistory() []qualificationVideoBatchObservation {
+	f.videoBatchMu.Lock()
+	defer f.videoBatchMu.Unlock()
+	count := min(f.videoBatchCount, uint64(len(f.videoBatches)))
+	result := make([]qualificationVideoBatchObservation, 0, count)
+	for index := f.videoBatchCount - count; index < f.videoBatchCount; index++ {
+		result = append(result, f.videoBatches[index%uint64(len(f.videoBatches))])
+	}
+	return result
+}
+
+func (f *qualificationApolloFixture) recordVideoBatch(providerFrame uint32, sourcePacket uint64, packetWithinFrame int, started time.Time) {
+	f.videoBatchMu.Lock()
+	if f.videoBatchEpoch.IsZero() {
+		f.videoBatchEpoch = started
+	}
+	f.videoBatches[f.videoBatchCount%uint64(len(f.videoBatches))] = qualificationVideoBatchObservation{
+		ProviderFrame: providerFrame, SourcePacket: sourcePacket,
+		PacketWithinFrame: packetWithinFrame, StartedAfter: started.Sub(f.videoBatchEpoch),
+	}
+	f.videoBatchCount++
+	f.videoBatchMu.Unlock()
+}
+
 func newQualificationApolloFixture(t *testing.T, serverTLS, clientTLS *tls.Config, sessionID string, profile qualificationMediaProfile) *qualificationApolloFixture {
 	t.Helper()
 	fixture := &qualificationApolloFixture{sessionID: sessionID, keyReady: make(chan []byte, 1), failures: make(chan error, 8)}
@@ -743,7 +817,7 @@ func (f *qualificationApolloFixture) serveMedia(socket *net.UDPConn, video bool)
 	}
 }
 
-func (f *qualificationApolloFixture) sendVideo(ctx context.Context, packets [][]byte) error {
+func (f *qualificationApolloFixture) sendVideo(ctx context.Context, providerFrame uint32, packets [][]byte) error {
 	for f.videoRemote.Load() == nil {
 		select {
 		case err := <-f.failures:
@@ -754,15 +828,95 @@ func (f *qualificationApolloFixture) sendVideo(ctx context.Context, packets [][]
 		}
 	}
 	remote := f.videoRemote.Load()
-	for _, packet := range packets {
-		if _, err := f.video.WriteToUDP(packet, remote); err != nil {
+	if len(packets) == 0 {
+		return nil
+	}
+	packetsPerMillisecond, batchSize := qualificationApolloVideoPacing(apolloVideoRawPacketSize)
+	if packetsPerMillisecond == 0 || batchSize == 0 {
+		return ErrProviderMalformed
+	}
+
+	f.videoPaceMu.Lock()
+	defer f.videoPaceMu.Unlock()
+	framePackets := 0
+	for batchStart := 0; batchStart < len(packets); batchStart += batchSize {
+		batchEnd := min(batchStart+batchSize, len(packets))
+		sourceStart := f.sentPackets.Load()
+		if err := qualificationWaitContext(ctx, f.videoNext); err != nil {
 			return err
 		}
+		if f.beforeVideoBatch != nil {
+			if err := f.beforeVideoBatch(ctx, framePackets); err != nil {
+				return err
+			}
+		}
+		if f.beforeVideoFirstWrite != nil {
+			if err := f.beforeVideoFirstWrite(ctx, framePackets); err != nil {
+				return err
+			}
+		}
+		firstPacket := packets[batchStart]
+		if len(firstPacket) != len(packets[0]) {
+			return ErrProviderMalformed
+		}
+		if _, err := f.video.WriteToUDP(firstPacket, remote); err != nil {
+			return err
+		}
+		batchStarted := time.Now()
 		f.sentPackets.Add(1)
+		f.recordVideoBatch(providerFrame, sourceStart, framePackets, batchStarted)
+		if f.observeVideoBatch != nil {
+			f.observeVideoBatch(framePackets, batchStarted)
+		}
+		for _, packet := range packets[batchStart+1 : batchEnd] {
+			if len(packet) != len(packets[0]) {
+				return ErrProviderMalformed
+			}
+			if _, err := f.video.WriteToUDP(packet, remote); err != nil {
+				return err
+			}
+			f.sentPackets.Add(1)
+		}
+		currentBatch := batchEnd - batchStart
+		framePackets += currentBatch
+		f.videoNext = batchStarted.Add(qualificationApolloVideoOffset(currentBatch, packetsPerMillisecond))
 	}
 	return nil
 }
 
+func qualificationApolloVideoPacing(packetBytes int) (packetsPerMillisecond, batchSize int) {
+	if packetBytes <= 0 {
+		return 0, 0
+	}
+	packetsPerMillisecond = qualificationApolloVideoRateBitsPerSecond / 1000 / packetBytes / 8
+	batchSize = min(qualificationApolloVideoBatchBytes/packetBytes, qualificationApolloVideoBatchPackets)
+	return packetsPerMillisecond, batchSize
+}
+
+func qualificationApolloVideoOffset(packets, packetsPerMillisecond int) time.Duration {
+	return time.Millisecond * time.Duration(packets) / time.Duration(packetsPerMillisecond)
+}
+
+func qualificationWaitContext(ctx context.Context, due time.Time) error {
+	delay := time.Until(due)
+	if delay <= 0 {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+			return nil
+		}
+	}
+	timer := time.NewTimer(delay)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}
+
 func (f *qualificationApolloFixture) streamKey() ([]byte, error) {
 	key := f.key.Load()
 	if key == nil || len(*key) != 16 {
@@ -921,13 +1075,21 @@ func newQualificationImpairedPath(t *testing.T, profile qualificationMediaProfil
 }
 
 func newQualificationPathWithImpairment(t *testing.T, profile qualificationMediaProfile, pacerKbps int64, impairment *qualificationImpairmentProfile) *qualificationPath {
+	return newQualificationPathWithNativeBackend(t, profile, pacerKbps, impairment, NewNativeApolloBackend())
+}
+
+func newQualificationPathWithNativeBackend(t *testing.T, profile qualificationMediaProfile, pacerKbps int64, impairment *qualificationImpairmentProfile, native *NativeApolloBackend) *qualificationPath {
+	return newQualificationPathWithNativeBackendAndObserver(t, profile, pacerKbps, impairment, native, nil)
+}
+
+func newQualificationPathWithNativeBackendAndObserver(t *testing.T, profile qualificationMediaProfile, pacerKbps int64, impairment *qualificationImpairmentProfile, native *NativeApolloBackend, observer func(mediaTimingObservation)) *qualificationPath {
 	t.Helper()
 	serverTLS, clientTLS := testTLS(t)
 	fixture := newQualificationApolloFixture(t, serverTLS, clientTLS, "qualification-session", profile)
 	if impairment != nil {
 		fixture.setControlImpairment(*impairment)
 	}
-	backend := &qualificationTracingBackend{native: NewNativeApolloBackend()}
+	backend := &qualificationTracingBackend{native: native}
 	provider := NewApolloAdapter(backend, ProviderIdentity{})
 	authority := protocol.SessionAuthority{
 		Version: "1", SessionID: "qualification-session", GatewayID: "gateway-1",
@@ -943,7 +1105,7 @@ func newQualificationPathWithImpairment(t *testing.T, profile qualificationMedia
 		ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: authority.GatewayID,
 		Capabilities: DefaultCapabilities(), ProviderCapabilities: DefaultCapabilities(),
 		Admission: admission, ProviderStateReporter: &recordingProviderStateReporter{},
-		Provider: provider, PacerKbps: pacerKbps,
+		Provider: provider, PacerKbps: pacerKbps, mediaObserver: observer,
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -1075,19 +1237,39 @@ func (p *qualificationPath) emit(t *testing.T, payload []byte) (qualificationPat
 	}
 	p.frame++
 	packets := qualificationSourceVideoPackets(t, p.key, p.frame, payload)
+	p.expectedSourceUDP.Add(uint64(len(packets)))
 	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
 	defer cancel()
-	if err := p.fixture.sendVideo(ctx, packets); err != nil {
+	if err := p.fixture.sendVideo(ctx, p.frame, packets); err != nil {
 		return qualificationPathTrace{}, err
 	}
 	p.sourceUDP.Add(uint64(len(packets)))
 	return qualificationPathTrace{}, nil
 }
 
+func (p *qualificationPath) processingDiagnostics(processed int64) qualificationProcessingDiagnostics {
+	diagnostics := qualificationProcessingDiagnostics{
+		ProcessedFrames: processed, SourceFrames: p.frame,
+		ExpectedWritesThroughLastFrame: p.expectedSourceUDP.Load(),
+		FixtureSentPackets:             p.fixture.sentPackets.Load(), SourceUDP: p.sourceUDP.Load(),
+		BatchHistory: p.fixture.videoBatchHistory(),
+	}
+	var err error
+	diagnostics.Gateway, err = p.process.snapshot()
+	if err != nil {
+		diagnostics.SnapshotError = err.Error()
+	}
+	return diagnostics
+}
+
 func (p *qualificationPath) receivePayload(parent context.Context) ([]byte, error) {
+	return p.receivePayloadObserved(parent, nil)
+}
+
+func (p *qualificationPath) receivePayloadObserved(parent context.Context, observe func(time.Time, int)) ([]byte, error) {
 	ctx, cancel := context.WithTimeout(parent, 2*time.Second)
 	defer cancel()
-	return p.client.ReceiveMedia(ctx)
+	return p.client.receiveMedia(ctx, observe)
 }
 
 func qualificationSourceVideoPackets(t *testing.T, key []byte, frame uint32, encoded []byte) [][]byte {
@@ -1345,6 +1527,10 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro
 	beforeProviderDrops := path.session.mediaDrops.Load()
 	beforeSourceUDP := path.sourceUDP.Load()
 	beforePacer := path.server.pacer.reservations.Load()
+	wireFragmentsPerUnit := (media.PacketBytes + frameV2PayloadSize - 1) / frameV2PayloadSize
+	wireDeliveryLimit := len(jobs) * wireFragmentsPerUnit
+	wireDeliveries := make([]qualificationDeliverySample, 0, wireDeliveryLimit)
+	wireOverflow := false
 	grace := max(2*profile.RTT+2*profile.Jitter, 2*time.Second)
 	lastTarget := time.Duration(packetCount) * spacing
 	if len(jobs) > 0 {
@@ -1364,7 +1550,13 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro
 		metrics := beforeMetrics
 		seen := make([]bool, packetCount)
 		for len(result.packets) < len(jobs) {
-			recovered, receiveErr := path.receivePayload(receiveCtx)
+			recovered, receiveErr := path.receivePayloadObserved(receiveCtx, func(receivedAt time.Time, bytes int) {
+				if len(wireDeliveries) >= wireDeliveryLimit {
+					wireOverflow = true
+					return
+				}
+				wireDeliveries = append(wireDeliveries, qualificationDeliverySample{At: receivedAt, Bytes: int64(bytes)})
+			})
 			if receiveErr != nil {
 				if errors.Is(receiveErr, context.DeadlineExceeded) || errors.Is(receiveErr, context.Canceled) {
 					break
@@ -1436,8 +1628,10 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro
 	if received.err != nil {
 		return qualificationImpairmentObservation{}, received.err
 	}
+	if wireOverflow {
+		return qualificationImpairmentObservation{}, errors.New("qualification public-wire observation bound exceeded")
+	}
 
-	var deliveries []qualificationDeliverySample
 	var totalLatency, totalJitter, previousLatency time.Duration
 	previousDelivered := -1
 	for _, packet := range received.packets {
@@ -1462,10 +1656,6 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro
 			observation.ObservedOutOfOrder++
 		}
 		previousDelivered = packet.index
-		datagrams := (media.PacketBytes + frameV2PayloadSize - 1) / frameV2PayloadSize
-		deliveries = append(deliveries, qualificationDeliverySample{
-			At: packet.deliveredAt, Bytes: int64(media.PacketBytes + datagrams*frameV2HeaderSize),
-		})
 		observation.MaxQueuePackets = max(observation.MaxQueuePackets, packet.queuePackets)
 	}
 	observation.Delivered = len(received.packets)
@@ -1553,18 +1743,33 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro
 	if err != nil {
 		return qualificationImpairmentObservation{}, err
 	}
+	wirePath := strings.TrimSuffix(rawPath, ".csv.gz") + "-wire.csv.gz"
 	for _, reduction := range profile.CapacitySteps {
-		bytesPerSecond := media.BitrateKbps * int64(100-reduction) * 1000 / 100 / 8
-		stepDeliveries := qualificationDeliveriesAfter(deliveries, stepAt[reduction])
-		convergence := qualificationMeasuredConvergence(stepDeliveries, stepAt[reduction], bytesPerSecond)
-		maximum := qualificationMaximumDeliveryBytes(stepDeliveries, 5*time.Second)
-		observation.CapacityStepObservations = append(observation.CapacityStepObservations, qualificationCapacityStep{
-			ReductionPercent: reduction, Convergence: convergence,
-			MaximumFiveSecond: maximum, FiveSecondCap: bytesPerSecond * 5,
-		})
+		step := qualificationCapacityStepObservation(wireDeliveries, stepAt[reduction], media, reduction)
+		step.TransitionAfter = stepAt[reduction].Sub(started)
+		step.RecomputationSource = filepath.Base(wirePath)
+		observation.CapacityStepObservations = append(observation.CapacityStepObservations, step)
+	}
+	if len(profile.CapacitySteps) > 0 {
+		wireEvidence, writeErr := writeQualificationWireSamples(
+			wirePath, started, profile.CapacitySteps, stepAt, wireDeliveries,
+			wireDeliveryLimit+len(profile.CapacitySteps),
+		)
+		if writeErr != nil {
+			return qualificationImpairmentObservation{}, writeErr
+		}
+		observation.RawWireSamples = wireEvidence.Name
+		observation.RawWireSamplesSHA256 = wireEvidence.SHA256
+		observation.RawWireSamplesBytes = wireEvidence.Bytes
+		observation.RawWireRows = wireEvidence.Rows
+		observation.RawWireDeliveryRows = wireEvidence.DeliveryRows
+		observation.RawWireTransitionRows = wireEvidence.TransitionRows
+		observation.RawWireTimebase = qualificationWireTimebase
+	}
+	for _, step := range observation.CapacityStepObservations {
 		if packetCount >= qualificationImpairmentPacketCount &&
-			(convergence > 10*time.Second || maximum > bytesPerSecond*5*105/100) {
-			return qualificationImpairmentObservation{}, fmt.Errorf("capacity step %d failed measured convergence=%s five-second=%d", reduction, convergence, maximum)
+			(step.Convergence > 10*time.Second || step.MaximumFiveSecond > step.FiveSecondCap*105/100) {
+			return qualificationImpairmentObservation{}, fmt.Errorf("capacity step %d failed measured convergence=%s five-second=%d", step.ReductionPercent, step.Convergence, step.MaximumFiveSecond)
 		}
 	}
 	if observation.Delivered+observation.Dropped != observation.Sent ||
@@ -1575,6 +1780,17 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro
 	return observation, nil
 }
 
+func qualificationCapacityStepObservation(wireDeliveries []qualificationDeliverySample, start time.Time, media qualificationMediaProfile, reduction int) qualificationCapacityStep {
+	bytesPerSecond := qualificationMediaPacerKbps(media, reduction) * 1000 / 8
+	deliveries := qualificationDeliveriesAfter(wireDeliveries, start)
+	return qualificationCapacityStep{
+		ReductionPercent:  reduction,
+		Convergence:       qualificationMeasuredConvergence(deliveries, start, bytesPerSecond),
+		MaximumFiveSecond: qualificationMaximumDeliveryBytes(deliveries, 5*time.Second),
+		FiveSecondCap:     bytesPerSecond * 5,
+	}
+}
+
 func qualificationKnownImpairment(profile qualificationImpairmentProfile) bool {
 	for _, known := range qualificationImpairmentProfiles() {
 		if profile.Name != known.Name || profile.RTT != known.RTT || profile.Jitter != known.Jitter ||
@@ -1601,9 +1817,16 @@ func qualificationDeliveriesAfter(deliveries []qualificationDeliverySample, star
 func qualificationMeasuredConvergence(deliveries []qualificationDeliverySample, start time.Time, targetBytesPerSecond int64) time.Duration {
 	const window = 250 * time.Millisecond
 	const requiredWindows = 4
+	if len(deliveries) == 0 {
+		return 11 * time.Second
+	}
+	windowOrigin := start
+	if deliveries[0].At.After(windowOrigin) {
+		windowOrigin = deliveries[0].At
+	}
 	consecutive := 0
 	for offset := time.Duration(0); offset <= 10*time.Second; offset += window {
-		windowStart := start.Add(offset)
+		windowStart := windowOrigin.Add(offset)
 		var total int64
 		for _, delivery := range deliveries {
 			if !delivery.At.Before(windowStart) && delivery.At.Before(windowStart.Add(window)) {
@@ -1614,7 +1837,7 @@ func qualificationMeasuredConvergence(deliveries []qualificationDeliverySample,
 		if rate >= targetBytesPerSecond*90/100 && rate <= targetBytesPerSecond*105/100 {
 			consecutive++
 			if consecutive == requiredWindows {
-				return offset + window
+				return windowStart.Add(window).Sub(start)
 			}
 		} else {
 			consecutive = 0
@@ -1641,14 +1864,21 @@ func qualificationMaximumDeliveryBytes(deliveries []qualificationDeliverySample,
 	return maximum
 }
 
-func runQualificationProcessing(t *testing.T, profile qualificationMediaProfile, rawPath string) (qualificationProcessingSummary, error) {
+func runQualificationProcessing(t *testing.T, profile qualificationMediaProfile, rawPath string) (summary qualificationProcessingSummary, err error) {
 	t.Helper()
 	payload := qualificationFramePayload(profile, 0)
 	if len(payload) < 4 {
 		return qualificationProcessingSummary{}, errors.New("qualification payload too small")
 	}
 	path := newQualificationProcessingPath(t, profile, qualificationFramePacerKbps(profile))
-	defer path.Close()
+	var processed int64
+	defer func() {
+		if err != nil {
+			diagnostics := path.processingDiagnostics(processed)
+			err = fmt.Errorf("%w; diagnostics=%#v", err, diagnostics)
+		}
+		path.Close()
+	}()
 	if err := runQualificationProcessWarmup(t, path, profile, payload); err != nil {
 		return qualificationProcessingSummary{}, err
 	}
@@ -1703,15 +1933,13 @@ func runQualificationProcessing(t *testing.T, profile qualificationMediaProfile,
 		}
 		receivedDone <- nil
 	}()
-	var processed int64
 	var nextRelease time.Time
 	payloadDigest := sha256.New()
 	for processed < targetFrames {
 		select {
 		case receiveErr := <-receivedDone:
 			if receiveErr != nil {
-				snapshot, _ := path.process.snapshot()
-				return qualificationProcessingSummary{}, fmt.Errorf("%w; gateway snapshot=%#v", receiveErr, snapshot)
+				return qualificationProcessingSummary{}, receiveErr
 			}
 			return qualificationProcessingSummary{}, errors.New("qualification processing receiver ended early")
 		default:
@@ -1729,8 +1957,7 @@ func runQualificationProcessing(t *testing.T, profile qualificationMediaProfile,
 		processed++
 	}
 	if err := <-receivedDone; err != nil {
-		snapshot, _ := path.process.snapshot()
-		return qualificationProcessingSummary{}, fmt.Errorf("%w; gateway snapshot=%#v", err, snapshot)
+		return qualificationProcessingSummary{}, err
 	}
 	qualificationWaitUntil(started.Add(profile.Duration))
 	actualDuration := time.Since(started)
@@ -1757,7 +1984,7 @@ func runQualificationProcessing(t *testing.T, profile qualificationMediaProfile,
 	if err != nil {
 		return qualificationProcessingSummary{}, err
 	}
-	summary, err := summarizeQualificationSamples(samples)
+	summary, err = summarizeQualificationSamples(samples)
 	if err != nil {
 		return qualificationProcessingSummary{}, err
 	}
@@ -1880,12 +2107,7 @@ func runQualificationWarmup(t *testing.T, path *qualificationPath, profile quali
 }
 
 func qualificationRuntimeSample(started time.Time) qualificationResourceSample {
-	var usage syscall.Rusage
-	cpuSeconds := -1.0
-	if syscall.Getrusage(syscall.RUSAGE_SELF, &usage) == nil {
-		cpuSeconds = float64(usage.Utime.Sec+usage.Stime.Sec) +
-			float64(usage.Utime.Usec+usage.Stime.Usec)/1_000_000
-	}
+	cpuSeconds := qualificationProcessCPUSeconds()
 	samples := []runtimemetrics.Sample{
 		{Name: "/memory/classes/heap/objects:bytes"},
 		{Name: "/sched/goroutines:goroutines"},
@@ -1957,6 +2179,100 @@ func qualificationFileSHA256(path string) (string, int64, error) {
 	return hex.EncodeToString(hash.Sum(nil)), size, nil
 }
 
+func writeQualificationWireSamples(
+	path string,
+	epoch time.Time,
+	reductions []int,
+	transitions map[int]time.Time,
+	deliveries []qualificationDeliverySample,
+	maximumRows int,
+) (qualificationWireFileEvidence, error) {
+	type wireRecord struct {
+		kind      string
+		reduction int
+		after     time.Duration
+		bytes     int64
+	}
+	if maximumRows < len(reductions)+len(deliveries) {
+		return qualificationWireFileEvidence{}, errors.New("qualification public-wire row bound exceeded")
+	}
+	records := make([]wireRecord, 0, len(reductions)+len(deliveries))
+	for _, reduction := range reductions {
+		at := transitions[reduction]
+		if at.IsZero() || at.Before(epoch) {
+			return qualificationWireFileEvidence{}, fmt.Errorf("qualification capacity transition %d missing or before epoch", reduction)
+		}
+		records = append(records, wireRecord{kind: "transition", reduction: reduction, after: at.Sub(epoch)})
+	}
+	for _, delivery := range deliveries {
+		if delivery.At.Before(epoch) || delivery.Bytes <= 0 {
+			return qualificationWireFileEvidence{}, errors.New("qualification public-wire delivery invalid")
+		}
+		records = append(records, wireRecord{kind: "delivery", after: delivery.At.Sub(epoch), bytes: delivery.Bytes})
+	}
+	sort.SliceStable(records, func(first, second int) bool {
+		if records[first].after != records[second].after {
+			return records[first].after < records[second].after
+		}
+		return records[first].kind == "transition" && records[second].kind != "transition"
+	})
+
+	file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640)
+	if err != nil {
+		return qualificationWireFileEvidence{}, err
+	}
+	compressed := gzip.NewWriter(file)
+	buffered := bufio.NewWriter(compressed)
+	writer := csv.NewWriter(buffered)
+	closeAll := func() error {
+		writer.Flush()
+		if err := writer.Error(); err != nil {
+			_ = compressed.Close()
+			_ = file.Close()
+			return err
+		}
+		if err := buffered.Flush(); err != nil {
+			_ = compressed.Close()
+			_ = file.Close()
+			return err
+		}
+		if err := compressed.Close(); err != nil {
+			_ = file.Close()
+			return err
+		}
+		return file.Close()
+	}
+	if err := writer.Write([]string{"record_type", "reduction_percent", "transition_after_ns", "received_after_ns", "encoded_bytes"}); err != nil {
+		_ = closeAll()
+		return qualificationWireFileEvidence{}, err
+	}
+	evidence := qualificationWireFileEvidence{Name: filepath.Base(path)}
+	for _, record := range records {
+		var row []string
+		switch record.kind {
+		case "transition":
+			row = []string{"transition", strconv.Itoa(record.reduction), strconv.FormatInt(record.after.Nanoseconds(), 10), "", ""}
+			evidence.TransitionRows++
+		case "delivery":
+			row = []string{"delivery", "", "", strconv.FormatInt(record.after.Nanoseconds(), 10), strconv.FormatInt(record.bytes, 10)}
+			evidence.DeliveryRows++
+		default:
+			_ = closeAll()
+			return qualificationWireFileEvidence{}, errors.New("qualification public-wire record type invalid")
+		}
+		if err := writer.Write(row); err != nil {
+			_ = closeAll()
+			return qualificationWireFileEvidence{}, err
+		}
+	}
+	if err := closeAll(); err != nil {
+		return qualificationWireFileEvidence{}, err
+	}
+	evidence.Rows = evidence.TransitionRows + evidence.DeliveryRows
+	evidence.SHA256, evidence.Bytes, err = qualificationFileSHA256(path)
+	return evidence, err
+}
+
 type qualificationFlowDelivery struct {
 	at    time.Time
 	flow  string
@@ -2018,7 +2334,7 @@ func runQualificationFleetStage(t *testing.T, fleet *qualificationFleet, profile
 
 func qualificationPacerEvidence(t *testing.T, rawPath string, baselineDuration, stepDuration time.Duration) (qualificationFairnessEvidence, error) {
 	t.Helper()
-	flows := []string{"one", "two", "three", "four", "five", "six", "seven", "eight"}
+	flows := qualificationFairnessFlows()
 	profile := qualificationMediaProfile{Name: "fairness-h264", Codec: "h264", BitrateKbps: 8000, PacketBytes: 1000}
 	fleet := newQualificationFleet(t, len(flows), profile, 8000)
 	defer fleet.Close()
@@ -2075,8 +2391,8 @@ func qualificationPacerEvidence(t *testing.T, rawPath string, baselineDuration,
 			return qualificationFairnessEvidence{}, fmt.Errorf("capacity step %d failed convergence=%s five-second=%d", step.reduction, convergence, maximum)
 		}
 		evidence.CapacitySteps = append(evidence.CapacitySteps, qualificationCapacityStep{
-			ReductionPercent: step.reduction, Convergence: convergence,
-			MaximumFiveSecond: maximum, FiveSecondCap: step.cap * 5,
+			ReductionPercent: step.reduction, TransitionAfter: stepStart.Sub(start), Convergence: convergence,
+			MaximumFiveSecond: maximum, FiveSecondCap: step.cap * 5, RecomputationSource: filepath.Base(rawPath),
 		})
 	}
 	end := start.Add(baselineDuration + 2*stepDuration)
@@ -2092,6 +2408,10 @@ func qualificationPacerEvidence(t *testing.T, rawPath string, baselineDuration,
 	return evidence, nil
 }
 
+func qualificationFairnessFlows() []string {
+	return []string{"one", "two", "three", "four", "five", "six", "seven", "eight"}
+}
+
 func qualificationPacerConvergence(deliveries []qualificationFlowDelivery, start time.Time, flows []string, targetBytesPerSecond int64) time.Duration {
 	consecutive := 0
 	for second := time.Duration(0); second < 10*time.Second; second += time.Second {

gateway/qualification_process_test.go

@@ -47,17 +47,21 @@ type qualificationGatewayProcessReady struct {
 }
 
 type qualificationGatewayProcessSnapshot struct {
-	Metrics                MetricsSnapshot
-	NativeSetups           uint64
-	NativeOpens            uint64
-	MediaIngress           uint64
-	MediaRecovered         uint64
-	MediaEnqueued          uint64
-	MediaDrops             uint64
-	MediaQueueMaximum      uint64
-	MediaQueueMaximumBytes uint64
-	PacerReservations      uint64
-	ProviderTelemetry      ProviderTelemetry
+	Metrics                     MetricsSnapshot
+	NativeSetups                uint64
+	NativeOpens                 uint64
+	MediaIngress                uint64
+	MediaRecovered              uint64
+	MediaEnqueued               uint64
+	MediaDrops                  uint64
+	MediaQueueMaximum           uint64
+	MediaQueueMaximumBytes      uint64
+	PacerReservations           uint64
+	ProviderTelemetry           ProviderTelemetry
+	VideoReceiveBuffer          int
+	VideoReceiveBufferAvailable bool
+	KernelDrops                 uint64
+	KernelDropsAvailable        bool
 }
 
 type qualificationProcessRecordRequest struct {
@@ -242,7 +246,7 @@ func (r *qualificationProcessRecorder) stop() (qualificationProcessRecordResult,
 		return qualificationProcessRecordResult{}, err
 	}
 	first, last := resources[0], resources[len(resources)-1]
-	result.CPUSeconds = max(last.CPUSeconds-first.CPUSeconds, 0)
+	result.CPUSeconds = qualificationCPUSecondsDelta(first.CPUSeconds, last.CPUSeconds)
 	result.AllocatedObjects = last.AllocatedObjects - first.AllocatedObjects
 	result.AllocatedBytes = last.AllocatedBytes - first.AllocatedBytes
 	for _, sample := range resources {
@@ -252,6 +256,35 @@ func (r *qualificationProcessRecorder) stop() (qualificationProcessRecordResult,
 	return result, nil
 }
 
+func qualificationCPUSecondsDelta(first, last float64) float64 {
+	cpuSeconds := -1.0
+	if first >= 0 && last >= first {
+		cpuSeconds = last - first
+	}
+	return cpuSeconds
+}
+
+func TestQualificationCPUSecondsDeltaRejectsUnavailableOrDecreasingSamples(t *testing.T) {
+	tests := []struct {
+		name        string
+		first, last float64
+		want        float64
+	}{
+		{name: "positive", first: 1.25, last: 1.75, want: 0.5},
+		{name: "zero", first: 1.25, last: 1.25, want: 0},
+		{name: "unavailable first", first: -1, last: 1.25, want: -1},
+		{name: "unavailable last", first: 1.25, last: -1, want: -1},
+		{name: "decreasing", first: 1.75, last: 1.25, want: -1},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if got := qualificationCPUSecondsDelta(test.first, test.last); got != test.want {
+				t.Fatalf("CPU delta for first=%f last=%f = %f, want %f", test.first, test.last, got, test.want)
+			}
+		})
+	}
+}
+
 type qualificationGatewayProcess struct {
 	command *exec.Cmd
 	cancel  context.CancelFunc
@@ -461,6 +494,9 @@ func TestQualificationGatewayProcessChild(t *testing.T) {
 			snapshot.MediaQueueMaximum = session.mediaQueueMaximum.Load()
 			snapshot.MediaQueueMaximumBytes = session.mediaQueueMaximumBytes.Load()
 			snapshot.ProviderTelemetry = session.Telemetry()
+			snapshot.VideoReceiveBuffer, snapshot.VideoReceiveBufferAvailable,
+				snapshot.KernelDrops, snapshot.KernelDropsAvailable =
+				qualificationProviderVideoSocketDiagnostics(session.videoConn)
 		}
 		_ = json.NewEncoder(response).Encode(snapshot)
 	})

gateway/qualification_socket_other_test.go

@@ -0,0 +1,11 @@
+//go:build !darwin && !linux
+
+package gateway
+
+import "net"
+
+func qualificationProcessCPUSeconds() float64 { return -1 }
+
+func qualificationProviderVideoSocketDiagnostics(*net.UDPConn) (receiveBuffer int, receiveBufferAvailable bool, kernelDrops uint64, kernelDropsAvailable bool) {
+	return 0, false, 0, false
+}

gateway/qualification_socket_unix_test.go

@@ -0,0 +1,80 @@
+//go:build darwin || linux
+
+package gateway
+
+import (
+	"io"
+	"net"
+	"os"
+	"runtime"
+	"strconv"
+	"strings"
+	"syscall"
+)
+
+func qualificationProcessCPUSeconds() float64 {
+	var usage syscall.Rusage
+	if syscall.Getrusage(syscall.RUSAGE_SELF, &usage) != nil {
+		return -1
+	}
+	return float64(usage.Utime.Sec+usage.Stime.Sec) +
+		float64(usage.Utime.Usec+usage.Stime.Usec)/1_000_000
+}
+
+func qualificationProviderVideoSocketDiagnostics(connection *net.UDPConn) (receiveBuffer int, receiveBufferAvailable bool, kernelDrops uint64, kernelDropsAvailable bool) {
+	if connection == nil {
+		return 0, false, 0, false
+	}
+	raw, err := connection.SyscallConn()
+	if err != nil {
+		return 0, false, 0, false
+	}
+	var inode uint64
+	var socketErr error
+	if err := raw.Control(func(descriptor uintptr) {
+		receiveBuffer, socketErr = syscall.GetsockoptInt(int(descriptor), syscall.SOL_SOCKET, syscall.SO_RCVBUF)
+		if runtime.GOOS == "linux" {
+			var stat syscall.Stat_t
+			if statErr := syscall.Fstat(int(descriptor), &stat); statErr == nil {
+				inode = stat.Ino
+			}
+		}
+	}); err != nil || socketErr != nil {
+		return 0, false, 0, false
+	}
+	receiveBufferAvailable = true
+	if runtime.GOOS == "linux" {
+		kernelDrops, kernelDropsAvailable = qualificationLinuxUDPDrops(inode)
+	}
+	return receiveBuffer, receiveBufferAvailable, kernelDrops, kernelDropsAvailable
+}
+
+func qualificationLinuxUDPDrops(inode uint64) (uint64, bool) {
+	if inode == 0 {
+		return 0, false
+	}
+	inodeText := strconv.FormatUint(inode, 10)
+	for _, path := range []string{"/proc/net/udp", "/proc/net/udp6"} {
+		file, err := os.Open(path)
+		if err != nil {
+			continue
+		}
+		raw, readErr := io.ReadAll(io.LimitReader(file, 1<<20))
+		closeErr := file.Close()
+		if readErr != nil || closeErr != nil {
+			continue
+		}
+		for _, line := range strings.Split(string(raw), "\n") {
+			fields := strings.Fields(line)
+			if len(fields) < 11 || fields[9] != inodeText {
+				continue
+			}
+			drops, err := strconv.ParseUint(fields[len(fields)-1], 10, 64)
+			if err != nil {
+				return 0, false
+			}
+			return drops, true
+		}
+	}
+	return 0, false
+}

gateway/telemetry.go

@@ -126,6 +126,7 @@ type fairPacer struct {
 type fairPacerFlow struct {
 	next     time.Time
 	lastSeen time.Time
+	debt     time.Duration
 }
 
 const fairPacerMaximumCatchup = 5 * time.Millisecond
@@ -180,9 +181,14 @@ func (p *fairPacer) reserveAt(now time.Time, flow string, bytes int) time.Time {
 		base = now
 	} else if lag := now.Sub(base); lag > fairPacerMaximumCatchup {
 		base = now.Add(-fairPacerMaximumCatchup)
+		state.debt = min(state.debt+lag-fairPacerMaximumCatchup, nativeApolloVideoQueueLatency-fairPacerMaximumCatchup)
 	}
 	numerator := int64(bytes) * int64(len(p.flows)) * int64(time.Second)
 	delay := time.Duration((numerator + p.bytesPerSecond - 1) / p.bytesPerSecond)
+	if repayment := min(delay/21, state.debt); repayment > 0 {
+		delay -= repayment
+		state.debt -= repayment
+	}
 	state.next = base.Add(delay)
 	p.flows[flow] = state
 	return state.next

openspec/changes/truthful-fixed-frame-qualification/design.md

@@ -2,6 +2,39 @@
 
 The current harness sends one fixed 1,179-byte payload per logical sample. It reaches the production path but does not represent encoded frames at 60/120 FPS or exercise realistic fragmentation, reassembly, queue bytes, and keyframe pressure.
 
+The complete-frame fixture also must preserve the pinned Apollo source schedule. For each frame it derives packets per millisecond from the raw UDP block size at 80% of 1 Gbps, limits source batches to both 64 KiB and 64 packets, and carries the next-send time into the following frame. Waiting is context-cancellable. This is qualification-fixture behavior only; production transport and queue behavior remain unchanged.
+
+Because the bounded fixture uses loopback rather than a physical 1 Gbps link, v8 writes the first shard of a batch successfully, captures that actual monotonic emission start, and schedules the next batch no earlier than that start plus the current batch's raw-block serialization interval. The persistent schedule carries across frames. A delayed batch therefore remains late instead of collapsing overdue batches into a catch-up burst.
+
+The retained v6 qualification run passed its then-current checks but is superseded because its tight-loop sender contradicted the pinned Apollo schedule. Private Linux runs 123 and 124 remain failed evidence. One local v8 sustained run passed on Darwin, but it is neither Linux proof nor normative Section 7 evidence.
+
+The later Darwin non-sustained pre-CI invocation was not green and was not retried. Its 1440p120 profile delivered the exact 6,250,000 bytes in 120 frames plus all 6,483 source and warm-up shards with zero drops, but measured 46,973.13 kbps over an implied approximately 1.0644383 seconds and failed the 5% throughput gate. Private Linux full verification/artifact retention and the replacement v10 normative run remain open.
+
+Private Linux run 125 at the frozen v8 harness head is retained as failed evidence. Its exact 33-datagram gap between successful fixture writes and production `MediaIngress` equaled the Linux socket's 33 measured kernel UDP drops. The complete-frame queue, fair pacer, QUIC fragmentation, and public decoder were downstream and did not account for the loss.
+
+The one authorized v8 Section 7 invocation at production candidate `55afea72a1487fa071501615d806e68efc0a436b` was consumed and failed. Its directory `gateway-rc10-55afea7` is retained byte-for-byte with two partial processing files and no manifest. The failure occurred at payload sequence 16801 after 16,834 provider frames had been recovered and enqueued; the provider queue reached 15 entries and dropped one valid frame while source-write and ingress accounting remained balanced at the diagnostic boundary. This attempt is failed evidence and is not eligible for retry or relabeling.
+
+The production fair pacer previously limited instantaneous recovery to 5 ms by moving an overdue flow's schedule to `now-5ms`, but silently discarded every additional valid scheduling interval. Repeated host stalls therefore accumulated complete frames in the existing provider queue until its 250 ms residence horizon correctly expired one. The repair keeps the 5 ms instantaneous ceiling, carries only the remaining debt up to that existing horizon, and shortens later nominal intervals by at most one twenty-first. That 20/21 interval is exactly 5% above nominal rate; once the debt reaches zero, the flow returns to its unchanged nominal interval. Per-flow debt and the shared nominal fair-share calculation preserve the existing eight-flow fairness and rolling aggregate cap through the existing 25% and 50% capacity changes.
+
+Private Linux run 127/job 481 at exact source `122080ab342d20585d9a45db0017337b9ece570a` is retained as failed evidence. `TestQualificationLossAndSteppedThroughputBounds` reported the 25% step's 11-second convergence sentinel and a 5,207,475-byte five-second maximum. No artifact was uploaded and the run was not retried. The retained log `/private/tmp/versevdi-gitea-run-127-job-481.log` has SHA-256 `2b368413d4b0e954c64ba6f6dcefb1e165166d773b6d8f84ee372bc2c92ff5f1`. The failure exposed a measurement-unit defect: capacity samples were emitted only after complete logical-payload reassembly and were compared with a payload-derived target even though the pacer reserves encoded public datagram bytes. It did not establish a production pacer defect, and `122080ab` is superseded as a final source candidate.
+
+Qualification v9 observes every raw public QUIC datagram immediately after the independent client's `ReceiveDatagram` returns and before the existing decoder/reassembler. Capacity convergence and rolling five-second maxima use those monotonic receive times and encoded lengths. Their target bytes per second and five-second cap derive from `qualificationMediaPacerKbps(profile, reduction) * 1000 / 8`. Complete logical-payload observations remain separate and continue to own payload integrity, loss, reorder, latency, throughput, and queue assertions. The first public delivery at or after a step anchors the four consecutive 250 ms windows so an arbitrary control-plane timestamp cannot split the first observed datagram pair. The delivery-after-step boundary, 90%-105% window bounds, ten-second convergence ceiling, and rolling-five-second 105% gate are unchanged.
+
+Private Linux run 128/job 482 was the single push-triggered attempt at exact source `22433e5c45c179e9d487b59e5d80f1dcf3b285ce`. Linux `make verify`, the sustained gate, strict OpenSpec, deterministic artifact generation, upload, and the clean-checkout step passed. Artifact 28's binaries and SPDX matched the frozen hashes and source metadata. The retained log `/private/tmp/versevdi-gitea-run-128-job-482.log` has SHA-256 `6f5852dce815ba87a55d61aae34cb2fce17a1a62c5ee50e18c0034259ad0a029`. The workflow requested 30 retention days, but Gitea 1.27 floored the positive request delay and the API recorded `2026-08-09T23:18:18+07:00` through `2026-09-07T23:18:18+07:00`, exactly 2,505,600 seconds or 29 elapsed days. The run was not retried. It is passing Linux execution and artifact-byte evidence but retention-nonconforming, so it neither satisfies the private Linux artifact gate nor authorizes normative Section 7. The local 31-day request compensates for verified platform rounding without changing the acceptance threshold; a separately authorized future run must prove an API interval of at least 2,592,000 seconds.
+
+At exact source `c0e362c0285d267f8af4087d07943822311f60e1`, the first Section 7 process created mode-0750 `gateway-rc10-c0e362c` and stopped before fixture startup because the sandbox denied its required loopback bind. The directory remains empty as environment-boundary evidence. The separately authorized escalated attempt retained `gateway-rc10-c0e362c-a2`, passed the v9 runtime checks, and wrote 16 files with manifest SHA-256 `61ea55140dfe6b37332de03879ac33206dd5d47763d09fb0fc2f65bb1f8e02b4`. Independent review denied normative acceptance: the constrained logical CSV and manifest aggregates did not persist every raw public QUIC datagram receive offset/encoded length, and `fairness.csv.gz` lacked manifest transition offsets. A2 therefore remains runtime-passing but normative-raw-evidence-incomplete, without mutation or relabeling.
+
+Qualification v10 keeps the logical impairment CSV unchanged and adds only `impairment-constrained-1080p60-h264-wire.csv.gz`. One monotonic epoch is captured before the constrained delivery and transition sequence. The bounded CSV interleaves two explicit transition records with every public datagram observation in monotonic order using `record_type,reduction_percent,transition_after_ns,received_after_ns,encoded_bytes`; transition-only and delivery-only fields remain empty and are validated as such. Its maximum row count is derived from the existing constrained job bound, the frame-fragment count, and the two configured transitions rather than a captured-run row constant. The manifest binds the file name, SHA-256, compressed bytes, total/delivery/transition row counts, timebase, exact transition offsets, and each capacity summary's recomputation source. Each impairment step is recomputed from every delivery at or after its transition through completion, preserving the original v9 classifier semantics even after the next transition. Fairness remains stage-bounded because `runQualificationFleetStage` records each capacity stage as a separate slice; its summaries retain their 25% and 50% offsets relative to the existing fairness CSV epoch.
+
+The independent parser's caller explicitly selects smoke or normative authority. Normative validation requires exactly 10,000 sent logical units and always enforces the ten-second convergence and 105% rolling-cap gates; retained `sent` data cannot weaken them. Both wire and fairness readers reject compressed input before hashing when it exceeds a writer-derived bound, feed gzip output through a bounded standard-library reader before CSV parsing, and bound fields, offsets, flows, and encoded lengths from the canonical row count, schema, time horizon, and writer values. Aggregate-only v9 data and malformed schema/hash/count/order/transition/length or oversized inputs remain rejected.
+
+The ingress repair follows reviewed behavior rather than copying implementation source:
+
+- Apollo `adc5c5a0bd80831ce495434bb16aee2cd4175fb8`, GPL-3.0, `src/stream.cpp:1463-1474,1573-1627`, supplies the 80%-of-1-Gbps raw-block pacing, 64-KiB/64-packet batch cap, and cross-frame send schedule used by the fixture.
+- Moonlight common-C pin `2ea47752c3051d72a64bcca190024e8b354fa1ef`, GPL-3.0, `src/VideoStream.c:28-35,331-333` and `src/PlatformSockets.c:364-405`, supplies the reviewed 2,048-video-packet receive-buffer request and dedicated receive-thread behavior. The cited `VideoStream.c` blob is byte-identical at the local standalone `703a06946861ff82cd33e5e13c59c1b017f7ded9` checkout.
+
+The native provider therefore requests `2,048 * 1,072 = 2,195,456` bytes with `SetReadBuffer()` on the connected video socket immediately after dialing it. A setter error aborts setup; an OS-imposed cap is accepted without privilege or getter dependence. A dedicated drain owns a fixed 2,048-slot FIFO pool. Every slot is 1,433 bytes (`apolloMediaMaximumPacket + 1`), so oversized datagrams remain observably invalid rather than being truncated into the accepted range; packet storage is 2,934,784 bytes (about 2.80 MiB) plus fixed index and timestamp metadata. The existing single decrypt/FEC processor consumes those slots. When every slot is occupied, the drain keeps reading into one fixed 1,433-byte scratch buffer and counts each accepted-size discard in both ingress and drop telemetry; oversized datagrams retain the existing rejection semantics. Socket close cancels the blocking read, and media channels close only after the unchanged audio reader, video drain, and video processor exit. Audio and control behavior are unchanged.
+
 ## Goals / Non-Goals
 
 **Goals:**
@@ -19,8 +52,15 @@ The current harness sends one fixed 1,179-byte payload per logical sample. It re
 - Derive bytes per fixed interval from bitrate and FPS, distribute integer remainder deterministically, and shift bounded bytes into periodic keyframes while keeping the interval total exact.
 - Carry a deterministic frame index/pattern only in the generated payload bytes; no codec semantics are claimed.
 - Keep the existing path/impairment/resource driver and change its unit from datagram payload to complete frame.
+- Keep video decrypt/FEC single-threaded; only the bounded connected-socket drain is separated so crypto stalls cannot become unexplained kernel loss.
+- Preserve valid scheduling debt after bounded host stalls instead of converting it into provider-queue residence; repay it within the existing fair pacer without a new queue, interface, or configured headroom.
+- Classify constrained-capacity evidence from actual public datagram observations and configured wire capacity; do not infer transport timing from completed logical frames.
+- Persist constrained public-wire observations and transition events in one bounded monotonic-offset CSV, recompute impairment summaries from each transition through completion, and bind both impairment and stage-bounded fairness capacity summaries to their retained raw sources.
+- Select smoke versus normative evidence validation through trusted caller input and bound compressed bytes, decompressed bytes, fields, offsets, flows, and encoded lengths before independent CSV parsing.
 
 ## Risks / Trade-offs
 
 - [Keyframes can exceed queue budget] → use the reviewed 1 MiB frame ceiling and production byte-bound queue.
 - [Short smoke windows have rounding effects] → assert exact generated totals and report measured duration separately from normative ten-minute gates.
+- [A stalled video processor exhausts the user-space pool] → keep draining into one fixed scratch buffer and attribute accepted-size overflow to existing ingress/drop counters rather than kernel loss or unbounded allocation.
+- [Debt repayment creates a burst or aggregate oversubscription] → retain the 5 ms instantaneous ceiling and limit repayment to a 20/21 nominal interval per flow, with every rolling five-second aggregate window bounded to 105%.

openspec/changes/truthful-fixed-frame-qualification/proposal.md

@@ -2,11 +2,27 @@
 
 The existing fixed-profile harness treats each 1,179-byte datagram as an encoded frame, so its reported frame rate, frame boundaries, bitrate, queue pressure, and processing evidence do not model the named 60/120 FPS profiles.
 
+The v6 complete-frame fixture subsequently exposed a source-fidelity defect on ordinary Linux runners: it emitted every UDP shard in one tight loop, unlike pinned Apollo's bounded intra-frame rate and batch schedule. The affected v6 qualification evidence remains retained but is superseded for candidate-readiness purposes.
+
+Private Linux run 125 then demonstrated a separate production-ingress defect after source pacing was corrected: 33 successful fixture writes missing from `MediaIngress` matched 33 measured kernel UDP drops while decrypt/FEC, queue, pacer, QUIC, and client counters remained downstream of the shortfall.
+
+The consumed v8 Section 7 attempt at `55afea72` subsequently failed after accumulated host scheduling delays exposed the production pacer's discarded schedule debt beyond its 5 ms instantaneous catch-up allowance. The retained partial evidence remains failed and supersedes `55afea72` as a final executable candidate.
+
+Private Linux run 127/job 481 at exact source `122080ab342d20585d9a45db0017337b9ece570a` then failed `TestQualificationLossAndSteppedThroughputBounds`: the 25% capacity step reported the 11-second convergence sentinel and 5,207,475 bytes in the measured five-second window. The run produced no artifact and was not retried. Its retained log is `/private/tmp/versevdi-gitea-run-127-job-481.log`, SHA-256 `2b368413d4b0e954c64ba6f6dcefb1e165166d773b6d8f84ee372bc2c92ff5f1`; `122080ab` is superseded as a final source candidate.
+
+Private Linux run 128/job 482 was the single push-triggered attempt at exact source `22433e5c45c179e9d487b59e5d80f1dcf3b285ce`. Linux `make verify`, the sustained gate, strict OpenSpec, deterministic artifact generation, upload, and the clean-checkout step passed. Artifact 28's binaries and SPDX matched the frozen hashes and source metadata. The retained log `/private/tmp/versevdi-gitea-run-128-job-482.log` has SHA-256 `6f5852dce815ba87a55d61aae34cb2fce17a1a62c5ee50e18c0034259ad0a029`. The workflow requested 30 retention days, but Gitea 1.27 floored the positive request delay and its API scheduled exactly 29 elapsed days. The run was not retried: it is passing Linux execution and artifact-byte evidence but retention-nonconforming, so it does not satisfy the private Linux artifact gate or authorize normative Section 7. The local 31-day request preserves the at-least-30-elapsed-day requirement; a separately authorized future run must prove `expires_at - created_at >= 2,592,000` seconds.
+
+At exact source `c0e362c0285d267f8af4087d07943822311f60e1`, the first Section 7 attempt created `gateway-rc10-c0e362c` and stopped at the sandbox loopback-bind boundary, leaving that mode-0750 directory empty. The separately authorized escalated attempt `gateway-rc10-c0e362c-a2` then passed the v9 runtime gates and retained 16 files; its manifest has SHA-256 `61ea55140dfe6b37332de03879ac33206dd5d47763d09fb0fc2f65bb1f8e02b4`. Independent review denied normative acceptance because v9 retained only logical impairment rows and aggregate capacity summaries: it did not retain the raw public datagram timestamps/lengths or fairness transition offsets needed to recompute those summaries. Both attempts remain preserved without relabeling; a2 is runtime-passing but normative-raw-evidence-incomplete.
+
 ## What Changes
 
 - Generate deterministic variable-size encoded frame units at the named frame rates and target bitrates, including bounded keyframes.
 - Traverse native Apollo recovery, production queues, the production pacer, QUIC framing, and independent reassembly.
 - Assert frame count/rate, bitrate, exact bytes and boundaries, clean loss attribution, latency, and resource bounds.
+- Decouple native video socket draining from the single decrypt/FEC processor with a fixed provider-scoped receive pool and request the source-backed video receive-buffer size.
+- Retain bounded valid per-flow pacing debt after a host stall and repay it at no more than 5% above nominal fair share.
+- Measure capacity convergence and rolling caps from each raw public QUIC datagram's observed length and receive time while retaining completed logical-payload observations for integrity and traversal results.
+- Persist the constrained public datagram observations and capacity transitions under one monotonic epoch, bind their hash/counts into the v10 manifest, and retain fairness transition offsets for independent recomputation. Impairment summaries use the full delivery tail after each transition; trusted caller input selects normative validation, and bounded readers reject oversized compressed, decompressed, and field data.
 - Keep short smoke tests separate and leave all prior normative artifacts unchanged.
 
 ## Capabilities
@@ -21,4 +37,4 @@ None.
 
 ## Impact
 
-The existing qualification harness and canonical qualification specification only. No codec operation, production dependency, or normative run before immutable consumer publication. Requirements: P3C-002, P3C-008, P3C-029, P3C-030, P3C-033, VER-009, VER-010, OPS-015.
+The qualification harness and its canonical specification, plus the already-reviewed native Apollo video ingress in `gateway/apollo_native.go` and production fair pacer in `gateway/telemetry.go`. This v10 correction changes test-only retained evidence, not production behavior. Downstream complete-frame queues, audio/control ingress, codec/FEC formats, QUIC, dependencies, and public interfaces remain unchanged. No codec operation or normative rerun is included. Requirements: P3C-002, P3C-008, P3C-029, P3C-030, P3C-033, VER-009, VER-010, OPS-015.

openspec/changes/truthful-fixed-frame-qualification/specs/gateway-qualification/spec.md

@@ -3,6 +3,18 @@
 ### Requirement: Fixed media processing qualification
 The qualification harness SHALL drive pinned-mTLS Apollo management, encrypted RTSP, ENet, and provider UDP through native source validation, `readUDPMedia`, recovery/FEC, byte/count/latency-bounded production queues, the production fair pacer, Protocol complete-frame fragmentation, Verse framing/QUIC, and an independent bounded client reassembler for 1080p60 H.264 at 20 Mbps, 1440p120 HEVC at 50 Mbps, and 4K60 HEVC at 80 Mbps. The source fixture SHALL emit deterministic variable-size complete encoded frame units at the named 60/120 FPS rate, preserve exact target bytes over each fixed interval, and include bounded larger keyframes without codec operation. After a recorded warm-up, the frozen candidate SHALL run each profile for ten wall-clock minutes, preserve every frame's bytes and boundary, retain every monotonic processing sample plus bounded provider-queue observations, and report frame count, frame rate, bitrate, count, min, median, p90, p95, p99, max, mean, standard deviation, measured batched monotonic-clock overhead and method, and observed bitrate. Processing begins at complete provider-frame receipt and ends at QUIC handoff, excluding client transit and pacing. Queue delay SHALL measure provider-queue residence, processing SHALL measure gateway work before pacing, and pacing delay SHALL measure scheduler waiting. CPU, heap, allocations, and goroutines SHALL be measured from the isolated gateway process only; CPU SHALL be actual OS user plus system consumption and MUST NOT include idle wall capacity or unrelated parent fixture/client work. Successive profiles SHALL use independent resource-counter baselines. Any bypass, payload or boundary mutation, frame-rate/count mismatch, wall-duration violation, bitrate outside both lower and upper bounds, unexplained clean-path loss, zero or unbounded clock overhead, or p95 above 5 ms SHALL fail.
 
+Within each complete frame the source fixture SHALL reproduce pinned Apollo's source schedule by deriving packets per millisecond from the raw UDP block size at 80% of 1 Gbps, bounding each source batch to the smaller of 64 KiB or 64 packets, capturing the monotonic batch start immediately after the first successful shard write, scheduling the next batch no earlier than that start plus the current batch's raw-block serialization interval, carrying that schedule across frames, and making pacing waits context-cancellable. A delayed batch SHALL remain late rather than trigger an overdue catch-up burst.
+
+Native Apollo video ingress SHALL request a 2,195,456-byte socket receive buffer before media ping or worker startup and SHALL drain the connected video socket into a fixed FIFO pool of exactly 2,048 slots before the existing single decrypt/FEC processor. Each slot and the saturation scratch buffer SHALL be `apolloMediaMaximumPacket + 1` bytes so oversized datagrams remain rejected. A full pool SHALL NOT stop socket draining: each successfully read accepted-size discard SHALL increment both media-ingress and media-drop telemetry without allocation, while oversized reads SHALL retain the existing rejection accounting. Socket closure SHALL cancel the video read, and video/audio channels SHALL close only after the audio reader, video drain, and video processor exit. Audio and control ingress SHALL remain unchanged.
+
+The production fair pacer SHALL retain its 5 ms instantaneous catch-up ceiling. When a flow resumes later than that ceiling, it SHALL carry the remaining valid schedule debt only within the existing 250 ms provider-queue horizon and SHALL repay that debt using an interval no shorter than 20/21 of its nominal equal-tier fair-share interval. It SHALL return to the nominal interval when the debt is repaid. Simultaneous debt across eight equal-tier flows and the existing 25% and 50% capacity changes SHALL preserve the existing share-error contract and SHALL NOT exceed 105% of configured aggregate capacity in any rolling five-second window.
+
+Capacity-step convergence and rolling-cap evidence SHALL use the monotonic receive time and encoded length of every raw public QUIC datagram observed immediately after the independent client's `ReceiveDatagram` returns and before decode or reassembly. For each reduction, target bytes per second and the five-second cap SHALL derive from the configured public-wire rate, `qualificationMediaPacerKbps(profile, reduction) * 1000 / 8`. Completed logical-payload observations SHALL remain separate and SHALL continue to measure payload integrity, loss, reorder, latency, throughput, and queue behavior. An impairment capacity step SHALL include every delivery observation at or after its recorded transition through constrained-run completion; a later transition SHALL NOT truncate the earlier step's retained tail. It SHALL anchor measurement windows at the first such public delivery, require four consecutive 250 ms windows between 90% and 105% of its target, converge within ten seconds, and remain at or below 105% in every rolling five-second window.
+
+The constrained profile SHALL retain a bounded gzip CSV containing exactly two capacity-transition records and every observed raw public datagram delivery under one monotonic epoch captured before the constrained sequence. The schema SHALL distinguish transition and delivery records and SHALL contain `record_type`, `reduction_percent`, `transition_after_ns`, `received_after_ns`, and `encoded_bytes`; fields not applicable to a record type SHALL remain empty and SHALL be rejected when populated. The manifest SHALL bind the file name, SHA-256, compressed byte count, total row count, delivery row count, transition row count, monotonic timebase, exact 25% and 50% transition offsets, and each capacity summary's raw recomputation source. The row bound SHALL derive from the configured constrained-job and fragment bounds rather than a prior run's observed row count.
+
+The retained fairness manifest SHALL bind its 25% and 50% transition offsets to the monotonic epoch of `fairness.csv.gz`. Fairness recomputation SHALL remain bounded to each separately collected `runQualificationFleetStage` capacity slice. An independent parser SHALL be able to reconstruct each stage and reproduce the rolling-five-second maximum, configured cap, and exact two-second fairness convergence from the retained raw files and manifest alone. The parser SHALL select normative versus smoke validation only from trusted caller input; normative validation SHALL require exactly 10,000 sent logical units and SHALL enforce the ten-second convergence and 105% rolling-cap gates unconditionally. Compressed, decompressed, row, field, offset, flow, and encoded-length limits SHALL derive from canonical writer schemas, configured row limits, and canonical run horizons and SHALL be enforced before CSV parsing can allocate an unbounded record. Missing raw-wire evidence; a wrong file hash, size, or count; duplicate or missing transitions; negative or nonmonotonic offsets; invalid encoded lengths; completed-logical-frame substitution; populated not-applicable fields; oversized input; or a summary mismatch SHALL fail qualification evidence acceptance.
+
 #### Scenario: Healthy fixed profile
 - **WHEN** a frozen candidate runs one fixed profile for the normative duration in the isolated qualification command
 - **THEN** the harness emits compressed raw frame/path and gateway-process resource samples plus a summary tied to the exact command, CPU scope, timing-overhead method, topology, source commit, immutable Protocol version, environment, and payload hash
@@ -10,3 +22,46 @@ The qualification harness SHALL drive pinned-mTLS Apollo management, encrypted R
 #### Scenario: Processing gate failure
 - **WHEN** any production path stage lacks a per-frame observation, stage accounting does not balance, payload or frame boundaries change, duration, frame-rate, frame-count, or bitrate bounds fail, measured p95 exceeds 5 ms, parent work changes gateway CPU, idle capacity is reported as consumed CPU, or timing overhead is absent
 - **THEN** the qualification command exits unsuccessfully without recording a passing candidate
+
+#### Scenario: Source-shaped Apollo pacing is preserved
+- **WHEN** the fixture emits 1,072-byte encrypted video shards with 1,040-byte raw blocks for consecutive complete frames
+- **THEN** it uses 96 packets per millisecond, batches at most 63 shards, records each batch after its first successful shard write, starts each later batch no earlier than the prior batch's raw serialization interval, carries the schedule into the following frame, and emits no shard after a cancelled pacing wait
+
+#### Scenario: Video crypto processing stalls
+- **WHEN** the first video AEAD operation is blocked while a 662-shard keyframe arrives
+- **THEN** all 662 successful connected-socket reads reach media-ingress accounting before processing resumes, and after release the exact complete frame traverses recovery, the bounded production queue, pacer, QUIC, and independent reassembly
+
+#### Scenario: Video ingress pool saturates
+- **WHEN** all 2,048 fixed video slots are occupied
+- **THEN** accepted-size datagrams are deliberately discarded through the fixed scratch buffer and counted as ingress plus drops, oversized datagrams remain rejected, and cancellation closes every media worker without a race or leak
+
+#### Scenario: Repeated media-loop host stalls
+- **WHEN** three approximately 95 ms scheduling debts are introduced at separated completed-public-frame barriers while source recovery continues
+- **THEN** the pacer limits instantaneous catch-up to 5 ms, repays each remaining debt at no more than 5% above nominal fair share, preserves every frame in exact order and bytes without provider or gateway drops, stays within the existing queue bounds, and closes cleanly on cancellation
+
+#### Scenario: Capacity measurement crosses a short transition phase
+- **WHEN** a constrained 1080p flow carries nonzero bounded debt through the approximately 1.572-second 25% phase before the 50% transition
+- **THEN** convergence and rolling-cap checks use the observed 1,200-byte and 25-byte public datagrams against the configured wire targets, while the separately retained completed-payload observations cannot substitute for transport delivery timing
+
+#### Scenario: Aggregate-only capacity evidence is retained
+- **WHEN** a qualification bundle contains logical-frame impairment rows and aggregate capacity summaries but omits raw public-wire rows or fairness transition offsets
+- **THEN** independent evidence validation rejects the bundle as incomplete even if its in-process runtime assertions passed
+
+#### Scenario: Raw capacity evidence is independently recomputed
+- **WHEN** v10 validation reads the retained constrained wire CSV, fairness CSV, and manifest transitions
+- **THEN** it validates bounded schema, hashes, sizes, counts, monotonic offsets, encoded datagram lengths, and transition uniqueness, then exactly reproduces the full-after-transition impairment targets, four consecutive 250 ms convergence windows, every rolling-five-second maximum, and stage-bounded fairness two-second alignment
+
+#### Scenario: Retained counts cannot weaken normative gates
+- **WHEN** a purported normative bundle retains a sent count other than 10,000 or retains an 11-second convergence or over-cap summary
+- **THEN** validation rejects it regardless of any retained field value, while explicitly selected smoke validation still requires exact raw-summary recomputation
+
+#### Scenario: Retained CSV exceeds bounded evidence grammar
+- **WHEN** a wire or fairness gzip exceeds its canonical compressed or decompressed limit or contains an overlong field, out-of-horizon offset, unknown flow, or out-of-range encoded length
+- **THEN** validation rejects it through the bounded standard-library reader before an unbounded CSV record can be allocated
+
+### Requirement: Retained private Linux candidate artifact
+A retained private Linux candidate artifact SHALL have API metadata whose `expires_at - created_at` interval is at least 30 elapsed days (2,592,000 seconds). Workflow intent, cleanup lag, and a local copy SHALL NOT substitute for the recorded API interval. A shorter interval SHALL fail the artifact-retention gate even when execution and artifact bytes pass. The workflow request MAY exceed 30 calendar days only to compensate for verified platform rounding; the acceptance threshold remains at least 30 elapsed days.
+
+#### Scenario: Platform rounding shortens retention
+- **WHEN** a private Linux candidate run passes execution and artifact-byte checks but its artifact API metadata records less than 2,592,000 seconds between creation and expiry
+- **THEN** the artifact-retention gate remains failed until a separately authorized candidate run records an interval of at least 2,592,000 seconds

openspec/changes/truthful-fixed-frame-qualification/tasks.md

@@ -16,4 +16,37 @@
 
 ## 4. Frozen qualification
 
-- [ ] 4.1 Run the single normative Section 7 qualification after immutable Protocol consumer resolution
+- [x] 4.1 Run the earlier single normative Section 7 qualification after immutable Protocol consumer resolution; later audit findings superseded that candidate
+
+## 5. Pinned Apollo source-fidelity remediation
+
+- [x] 5.1 Retain the v6 qualification attempt and mark its passing result superseded by the tight-loop source defect
+- [x] 5.2 Implement v8 post-first-write, non-collapsing complete-frame UDP pacing with persistent cross-frame carry and verify the focused, race, short-resource, and cross-platform compile checks that passed
+- [x] 5.3 Preserve private Linux runs 123 and 124 as failed evidence, the passing local v8 Darwin sustained run as non-Linux and non-normative, and the un-retried failed Darwin non-sustained pre-CI invocation with its exact throughput evidence
+- [ ] 5.4 Run private Linux full verification and retain deterministic Linux artifacts for the frozen v10 evidence descendant
+- [ ] 5.5 Run one separately approved replacement v10 normative Section 7 qualification
+
+## 6. Native video ingress remediation
+
+- [x] 6.1 Preserve run 125 and reproduce its pre-decrypt shortfall with a public 662-shard blocked-AEAD regression
+- [x] 6.2 Add the video-only 2,195,456-byte socket-buffer request, fixed 2,048-slot drain, single processor, overflow accounting, and bounded cancellation tests
+- [ ] 6.3 Freeze the reviewed production repair through the still-open private Linux full-verification and artifact gate before any replacement normative run
+
+## 7. Fair-pacer schedule-debt remediation
+
+- [x] 7.1 Preserve the consumed failed `55afea72` v8 attempt and its two partial files without retry, relabeling, or modification
+- [x] 7.2 Reproduce repeated host-stall queue expiry through the public native path and add bounded one-flow/eight-flow debt, fairness, rolling-cap, and capacity-step regressions
+- [x] 7.3 Retain the 5 ms instantaneous ceiling, carry valid debt within the 250 ms queue horizon, and repay it at no more than 5% above nominal fair share
+- [ ] 7.4 Freeze and verify a new executable candidate on private Linux before any separately authorized replacement normative run
+
+## 8. Public-wire capacity measurement correction
+
+- [x] 8.1 Preserve run 127/job 481 at `122080ab` as failed evidence with its exact log hash, no artifact, and no retry
+- [x] 8.2 Observe raw independent-client QUIC datagram lengths/times in-process and derive v9 capacity convergence and caps from configured wire rate while keeping logical payload observations separate; this did not persist sufficient raw evidence for independent proof
+- [x] 8.3 Preserve run 128/job 482 as passing Linux execution and artifact-byte evidence but retention-nonconforming, with no retry
+- [ ] 8.4 Freeze the 31-day-request descendant and prove a future private artifact records at least 30 elapsed days before tasks 5.4, 5.5, 6.3, or 7.4 can close
+
+## 9. Retained public-wire evidence correction
+
+- [x] 9.1 Preserve the first empty `gateway-rc10-c0e362c` environment-boundary attempt and the runtime-passing but normative-raw-evidence-incomplete `gateway-rc10-c0e362c-a2` bundle without mutation or relabeling
+- [x] 9.2 Persist bounded v10 constrained public-wire rows and fairness transition offsets; recompute each impairment step from its transition through completion and fairness from its separate stage; select normative authority explicitly; and bound compressed, decompressed, row, field, time, flow, and encoded-length parsing