test(gateway): observe qualification batch pacing

gateway/qualification_contract_test.go

@@ -75,9 +75,15 @@ func TestQualificationApolloFixturePacesSourceShapedVideo(t *testing.T) {
 	if len(packets) != 1000 || len(packets[0]) != 1072 {
 		t.Fatalf("source vector = %d packets of %d bytes, want 1000 packets of 1072 bytes", len(packets), len(packets[0]))
 	}
-	packetsPerMillisecond, batchSize := qualificationApolloVideoPacing(len(packets[0]))
-	if packetsPerMillisecond != 93 || batchSize != 61 {
-		t.Fatalf("Apollo pacing vector = %d packets/ms, batch %d; want 93 and 61", packetsPerMillisecond, batchSize)
+	packetsPerMillisecond, batchSize := qualificationApolloVideoPacing(apolloVideoRawPacketSize)
+	if apolloVideoRawPacketSize != 1040 || packetsPerMillisecond != 96 || batchSize != 63 {
+		t.Fatalf("Apollo raw pacing vector = %d bytes, %d packets/ms, batch %d; want 1040, 96, and 63", apolloVideoRawPacketSize, packetsPerMillisecond, batchSize)
+	}
+	wantOffsets := []time.Duration{0, 656250 * time.Nanosecond, 1312500 * time.Nanosecond, 10416666 * time.Nanosecond}
+	for index, sent := range []int{0, 63, 126, 1000} {
+		if got := qualificationApolloVideoOffset(sent, packetsPerMillisecond); got != wantOffsets[index] {
+			t.Fatalf("Apollo pacing offset after %d packets = %s, want %s", sent, got, wantOffsets[index])
+		}
 	}
 
 	receiver, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
@@ -94,6 +100,15 @@ func TestQualificationApolloFixturePacesSourceShapedVideo(t *testing.T) {
 	fixture := &qualificationApolloFixture{video: sender, failures: make(chan error, 1)}
 	remote := *receiver.LocalAddr().(*net.UDPAddr)
 	fixture.videoRemote.Store(&remote)
+	var batchIndices []int
+	var batchStarts []time.Duration
+	var started time.Time
+	fixture.observeVideoBatch = func(index int, at time.Time) {
+		if len(batchIndices) < 3 {
+			batchIndices = append(batchIndices, index)
+			batchStarts = append(batchStarts, at.Sub(started))
+		}
+	}
 	drained := make(chan struct{})
 	go func() {
 		defer close(drained)
@@ -105,7 +120,7 @@ func TestQualificationApolloFixturePacesSourceShapedVideo(t *testing.T) {
 		}
 	}()
 
-	started := time.Now()
+	started = time.Now()
 	if err := fixture.sendVideo(context.Background(), packets); err != nil {
 		t.Fatal(err)
 	}
@@ -113,10 +128,23 @@ func TestQualificationApolloFixturePacesSourceShapedVideo(t *testing.T) {
 		t.Fatal(err)
 	}
 	elapsed := time.Since(started)
-	wantCarry := 10 * time.Millisecond // floor(1000 / 93) ms at Apollo's pinned 80%-of-1-Gbps rate.
+	wantCarry := qualificationApolloVideoOffset(len(packets), packetsPerMillisecond)
+	if wantCarry != 10416666*time.Nanosecond {
+		t.Fatalf("1000-packet carry = %s, want 10.416666 ms", wantCarry)
+	}
 	if elapsed < wantCarry {
 		t.Fatalf("source fixture sent the next frame after %s, before Apollo pacing carry %s", elapsed, wantCarry)
 	}
+	wantIndices := []int{0, 63, 126}
+	wantStarts := []time.Duration{0, 656250 * time.Nanosecond, 1312500 * time.Nanosecond}
+	if len(batchIndices) != len(wantIndices) {
+		t.Fatalf("observed %d batch starts, want %d", len(batchIndices), len(wantIndices))
+	}
+	for index := range wantIndices {
+		if batchIndices[index] != wantIndices[index] || batchStarts[index] < wantStarts[index] {
+			t.Fatalf("batch starts = indices %v at %v; want indices %v no earlier than %v", batchIndices, batchStarts, wantIndices, wantStarts)
+		}
+	}
 	select {
 	case <-drained:
 	case <-time.After(time.Second):

gateway/qualification_harness_test.go

@@ -414,22 +414,23 @@ 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
-	videoPaceMu sync.Mutex
-	videoNext   time.Time
+	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
+	observeVideoBatch func(int, time.Time)
 
 	controlImpairmentMu sync.Mutex
 	controlRTT          time.Duration
@@ -762,7 +763,7 @@ func (f *qualificationApolloFixture) sendVideo(ctx context.Context, packets [][]
 	if len(packets) == 0 {
 		return nil
 	}
-	packetsPerMillisecond, batchSize := qualificationApolloVideoPacing(len(packets[0]))
+	packetsPerMillisecond, batchSize := qualificationApolloVideoPacing(apolloVideoRawPacketSize)
 	if packetsPerMillisecond == 0 || batchSize == 0 {
 		return ErrProviderMalformed
 	}
@@ -773,16 +774,16 @@ func (f *qualificationApolloFixture) sendVideo(ctx context.Context, packets [][]
 	if f.videoNext.After(frameStart) {
 		frameStart = f.videoNext
 	}
-	framePackets, groupPackets := 0, 0
+	framePackets := 0
 	for batchStart := 0; batchStart < len(packets); batchStart += batchSize {
-		if framePackets == 0 || groupPackets >= packetsPerMillisecond {
-			due := frameStart.Add(time.Millisecond * time.Duration(framePackets) / time.Duration(packetsPerMillisecond))
-			if err := qualificationWaitContext(ctx, due); err != nil {
-				return err
-			}
-			groupPackets = 0
-		}
 		batchEnd := min(batchStart+batchSize, len(packets))
+		due := frameStart.Add(qualificationApolloVideoOffset(framePackets, packetsPerMillisecond))
+		if err := qualificationWaitContext(ctx, due); err != nil {
+			return err
+		}
+		if f.observeVideoBatch != nil {
+			f.observeVideoBatch(framePackets, time.Now())
+		}
 		for _, packet := range packets[batchStart:batchEnd] {
 			if len(packet) != len(packets[0]) {
 				return ErrProviderMalformed
@@ -794,9 +795,8 @@ func (f *qualificationApolloFixture) sendVideo(ctx context.Context, packets [][]
 		}
 		currentBatch := batchEnd - batchStart
 		framePackets += currentBatch
-		groupPackets += currentBatch
 	}
-	f.videoNext = frameStart.Add(time.Millisecond * time.Duration(framePackets) / time.Duration(packetsPerMillisecond))
+	f.videoNext = frameStart.Add(qualificationApolloVideoOffset(framePackets, packetsPerMillisecond))
 	return nil
 }
 
@@ -809,6 +809,10 @@ func qualificationApolloVideoPacing(packetBytes int) (packetsPerMillisecond, bat
 	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 {

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

@@ -4,6 +4,8 @@ The current harness sends one fixed 1,179-byte payload per logical sample. It re
 
 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, each batch begins at its cumulative wire-rate offset. This preserves Apollo's raw-block rate and batch ceilings without collapsing multiple batches into an instantaneous loopback burst.
+
 ## Goals / Non-Goals
 
 **Goals:**

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

@@ -14,5 +14,5 @@ Within each complete frame the source fixture SHALL reproduce pinned Apollo's so
 - **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 for consecutive complete frames
-- **THEN** it uses 93 packets per millisecond, batches at most 61 shards, carries the integer next-send offset into the following frame, and emits no shard after a cancelled pacing wait
+- **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 at offsets derived from cumulative packet count, carries the next-send offset into the following frame, and emits no shard after a cancelled pacing wait