From ce3b3079837c5745f95a51c98eb6204363962ae7 Mon Sep 17 00:00:00 2001 From: sechmachine <97589681+sechmachine727@users.noreply.github.com> Date: Mon, 10 Aug 2026 02:42:54 +0700 Subject: [PATCH] test(gateway): retain raw wire qualification evidence --- gateway/qualification_contract_test.go | 970 +++++++++++++++++- gateway/qualification_harness_test.go | 169 ++- .../design.md | 10 +- .../proposal.md | 5 +- .../specs/gateway-qualification/spec.md | 22 +- .../tasks.md | 11 +- 6 files changed, 1165 insertions(+), 22 deletions(-) diff --git a/gateway/qualification_contract_test.go b/gateway/qualification_contract_test.go index 220b734..23ac31f 100644 --- a/gateway/qualification_contract_test.go +++ b/gateway/qualification_contract_test.go @@ -1,11 +1,13 @@ package gateway import ( + "bufio" "bytes" "compress/gzip" "context" "crypto/cipher" "encoding/binary" + "encoding/csv" "errors" "fmt" "io" @@ -14,6 +16,7 @@ import ( "path/filepath" "reflect" "runtime" + "sort" "strconv" "strings" "sync" @@ -660,8 +663,9 @@ func TestQualificationSixImpairmentProfilesTraverseProductionPath(t *testing.T) func TestQualificationLossAndSteppedThroughputBounds(t *testing.T) { profiles := qualificationImpairmentProfiles() for _, profile := range profiles[3:] { + directory := t.TempDir() observation, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], - qualificationImpairmentPacketCount, filepath.Join(t.TempDir(), profile.Name+".csv.gz")) + qualificationImpairmentPacketCount, filepath.Join(directory, profile.Name+".csv.gz")) if err != nil { t.Fatalf("%s: %v", profile.Name, err) } @@ -672,6 +676,932 @@ func TestQualificationLossAndSteppedThroughputBounds(t *testing.T) { for _, step := range observation.CapacityStepObservations { t.Logf("%s %d%%: convergence=%s wire_max_5s=%d wire_cap=%d", profile.Name, step.ReductionPercent, step.Convergence, step.MaximumFiveSecond, step.FiveSecondCap) } + if profile.Name == "constrained" { + if err := validateQualificationWireBundle(directory, observation, qualificationMediaProfiles()[0], qualificationEvidenceNormative); err != nil { + t.Fatal(err) + } + } + } +} + +func TestQualificationBundleRetainsIndependentlyRecomputableWireEvidence(t *testing.T) { + profile := qualificationImpairmentProfiles()[5] + directory := t.TempDir() + rawPath := filepath.Join(directory, "impairment-constrained-1080p60-h264.csv.gz") + observation, err := runQualificationImpairment(t, profile, qualificationMediaProfiles()[0], 40, rawPath) + if err != nil { + t.Fatal(err) + } + if err := validateQualificationWireBundle(directory, observation, qualificationMediaProfiles()[0], qualificationEvidenceSmoke); err != nil { + t.Fatal(err) + } +} + +func TestQualificationV9AggregateOnlyBundleIsRejectedAsIncomplete(t *testing.T) { + observation := qualificationImpairmentObservation{ + ConfiguredCapacitySteps: []int{25, 50}, + CapacityStepObservations: []qualificationCapacityStep{ + {ReductionPercent: 25, Convergence: 2 * time.Second, MaximumFiveSecond: 1, FiveSecondCap: 1}, + {ReductionPercent: 50, Convergence: 2 * time.Second, MaximumFiveSecond: 1, FiveSecondCap: 1}, + }, + RawSamples: "impairment-constrained-1080p60-h264.csv.gz", + } + if err := validateQualificationWireBundle(t.TempDir(), observation, qualificationMediaProfiles()[0], qualificationEvidenceSmoke); err == nil || !strings.Contains(err.Error(), "missing retained public-wire samples") { + t.Fatalf("v9 aggregate-only bundle rejection = %v", err) + } +} + +type qualificationEvidenceMode uint8 + +const ( + qualificationEvidenceSmoke qualificationEvidenceMode = iota + qualificationEvidenceNormative +) + +const qualificationCanonicalCapacityStepCount = 2 + +var qualificationCanonicalCapacitySteps = [qualificationCanonicalCapacityStepCount]int{25, 50} + +type qualificationCSVLimits struct { + compressedBytes int64 + decompressedBytes int64 + headerBytes []int + fieldBytes []int +} + +type qualificationBoundedCSVReader struct { + file *os.File + compressed *gzip.Reader + limited *io.LimitedReader + reader *csv.Reader + headerBytes []int + fieldBytes []int + first bool +} + +func qualificationMaximumWireOffset(media qualificationMediaProfile) time.Duration { + minimumKbps := qualificationMediaPacerKbps(media, 50) + spacing := time.Duration(int64(time.Second) * int64(media.PacketBytes) * 8 / (minimumKbps * 1000)) + maximumNetworkDelay := time.Duration(0) + for _, profile := range qualificationImpairmentProfiles() { + maximumNetworkDelay = max(maximumNetworkDelay, profile.RTT+profile.Jitter) + } + return time.Duration(qualificationImpairmentMaxPackets)*spacing + maximumNetworkDelay + nativeApolloVideoQueueLatency + 10*time.Second +} + +func qualificationMaximumFairnessOffset() time.Duration { + return 60*time.Second + 2*10*time.Second + nativeApolloVideoQueueLatency +} + +func qualificationGzipMaximumBytes(decompressedBytes int64) int64 { + const maximumStoredBlockBytes = 16_383 + return decompressedBytes + ((decompressedBytes+maximumStoredBlockBytes-1)/maximumStoredBlockBytes)*5 + 64 +} + +func qualificationCSVLimitsFor(header []string, maximumRows int, maximumFieldBytes []int) qualificationCSVLimits { + rowBytes := int64(len(maximumFieldBytes)) + fieldBytes := make([]int, len(maximumFieldBytes)) + headerBytes := make([]int, len(header)) + for index, size := range maximumFieldBytes { + headerBytes[index] = len(header[index]) + fieldBytes[index] = size + rowBytes += int64(max(size, headerBytes[index])) + } + headerLineBytes := int64(len(strings.Join(header, ",")) + 1) + decompressedBytes := headerLineBytes + int64(maximumRows)*rowBytes + return qualificationCSVLimits{ + compressedBytes: qualificationGzipMaximumBytes(decompressedBytes), decompressedBytes: decompressedBytes, + headerBytes: headerBytes, fieldBytes: fieldBytes, + } +} + +func qualificationWireCSVLimits(media qualificationMediaProfile, maximumRows int) qualificationCSVLimits { + maxOffsetBytes := len(strconv.FormatInt(qualificationMaximumWireOffset(media).Nanoseconds(), 10)) + return qualificationCSVLimitsFor( + []string{"record_type", "reduction_percent", "transition_after_ns", "received_after_ns", "encoded_bytes"}, + maximumRows, + []int{len("transition"), len("100"), maxOffsetBytes, maxOffsetBytes, len(strconv.Itoa(frameV2HeaderSize + frameV2PayloadSize))}, + ) +} + +func qualificationFairnessCSVLimits() qualificationCSVLimits { + maxFlowBytes := 0 + for _, flow := range qualificationFairnessFlows() { + maxFlowBytes = max(maxFlowBytes, len(flow)) + } + return qualificationCSVLimitsFor( + []string{"elapsed_ns", "flow", "bytes"}, qualificationImpairmentMaxPackets, + []int{ + len(strconv.FormatInt(qualificationMaximumFairnessOffset().Nanoseconds(), 10)), + maxFlowBytes, + len(strconv.Itoa(1000 + frameHeaderSize)), + }, + ) +} + +func openQualificationBoundedCSV(path string, limits qualificationCSVLimits, fieldsPerRecord int) (*qualificationBoundedCSVReader, error) { + info, err := os.Stat(path) + if err != nil { + return nil, err + } + if info.Size() > limits.compressedBytes { + return nil, errors.New("qualification CSV compressed byte bound exceeded") + } + file, err := os.Open(path) + if err != nil { + return nil, err + } + compressed, err := gzip.NewReader(file) + if err != nil { + _ = file.Close() + return nil, err + } + limited := &io.LimitedReader{R: compressed, N: limits.decompressedBytes + 1} + reader := csv.NewReader(limited) + reader.FieldsPerRecord = fieldsPerRecord + return &qualificationBoundedCSVReader{ + file: file, compressed: compressed, limited: limited, reader: reader, + headerBytes: limits.headerBytes, fieldBytes: limits.fieldBytes, first: true, + }, nil +} + +func (reader *qualificationBoundedCSVReader) Read() ([]string, error) { + row, err := reader.reader.Read() + if reader.limited.N == 0 { + return nil, errors.New("qualification CSV decompressed byte bound exceeded") + } + if err != nil { + return nil, err + } + bounds := reader.fieldBytes + if reader.first { + reader.first = false + bounds = reader.headerBytes + } + for index, field := range row { + if index >= len(bounds) || len(field) > bounds[index] { + return nil, errors.New("qualification CSV field bound exceeded") + } + } + return row, nil +} + +func (reader *qualificationBoundedCSVReader) Close() { + _ = reader.compressed.Close() + _ = reader.file.Close() +} + +func validateQualificationWireBundle(directory string, observation qualificationImpairmentObservation, media qualificationMediaProfile, mode qualificationEvidenceMode) error { + if mode != qualificationEvidenceSmoke && mode != qualificationEvidenceNormative { + return errors.New("qualification public-wire validation mode invalid") + } + if mode == qualificationEvidenceNormative && observation.Sent != qualificationImpairmentPacketCount { + return fmt.Errorf("qualification normative sent=%d want=%d", observation.Sent, qualificationImpairmentPacketCount) + } + if len(observation.ConfiguredCapacitySteps) != len(qualificationCanonicalCapacitySteps) { + return errors.New("qualification public-wire capacity steps invalid") + } + for index, reduction := range qualificationCanonicalCapacitySteps { + if observation.ConfiguredCapacitySteps[index] != reduction { + return errors.New("qualification public-wire capacity steps invalid") + } + } + if observation.RawWireSamples == "" || filepath.Base(observation.RawWireSamples) != observation.RawWireSamples { + return errors.New("qualification bundle missing retained public-wire samples") + } + if observation.RawWireTimebase != qualificationWireTimebase { + return errors.New("qualification public-wire timebase invalid") + } + fragmentsPerUnit := (media.PacketBytes + frameV2PayloadSize - 1) / frameV2PayloadSize + maximumRows := qualificationImpairmentMaxPackets*fragmentsPerUnit + qualificationCanonicalCapacityStepCount + limits := qualificationWireCSVLimits(media, maximumRows) + if observation.RawWireDeliveryRows <= 0 || observation.RawWireTransitionRows != qualificationCanonicalCapacityStepCount || + observation.RawWireRows < 1 || observation.RawWireRows > maximumRows || + observation.RawWireDeliveryRows > maximumRows-qualificationCanonicalCapacityStepCount || + observation.RawWireRows != observation.RawWireDeliveryRows+qualificationCanonicalCapacityStepCount { + return errors.New("qualification public-wire row counts invalid") + } + path := filepath.Join(directory, observation.RawWireSamples) + info, err := os.Stat(path) + if err != nil { + return err + } + if info.Size() > limits.compressedBytes { + return errors.New("qualification public-wire compressed byte bound exceeded") + } + digest, size, err := qualificationFileSHA256(path) + if err != nil { + return err + } + if digest != observation.RawWireSamplesSHA256 || size != observation.RawWireSamplesBytes { + return errors.New("qualification public-wire hash or size mismatch") + } + reader, err := openQualificationBoundedCSV(path, limits, 5) + if err != nil { + return err + } + defer reader.Close() + header, err := reader.Read() + if err != nil || !reflect.DeepEqual(header, []string{"record_type", "reduction_percent", "transition_after_ns", "received_after_ns", "encoded_bytes"}) { + return errors.New("qualification public-wire schema invalid") + } + + epoch := time.Unix(0, 0) + transitions := make(map[int]time.Duration, qualificationCanonicalCapacityStepCount) + deliveries := make([]qualificationDeliverySample, 0, observation.RawWireDeliveryRows) + expectedLengths := make([]int64, 0, fragmentsPerUnit) + remaining := media.PacketBytes + for remaining > 0 { + payloadBytes := min(remaining, frameV2PayloadSize) + expectedLengths = append(expectedLengths, int64(payloadBytes+frameV2HeaderSize)) + remaining -= payloadBytes + } + lastAfter := int64(-1) + rows, deliveryRows, transitionRows := 0, 0, 0 + for { + row, readErr := reader.Read() + if errors.Is(readErr, io.EOF) { + break + } + if readErr != nil { + return readErr + } + rows++ + if rows > maximumRows { + return errors.New("qualification public-wire row bound exceeded") + } + switch row[0] { + case "transition": + if row[1] == "" || row[2] == "" || row[3] != "" || row[4] != "" { + return errors.New("qualification public-wire transition fields invalid") + } + reduction, reductionErr := strconv.Atoi(row[1]) + after, afterErr := strconv.ParseInt(row[2], 10, 64) + if reductionErr != nil || afterErr != nil || after < 0 || time.Duration(after) > qualificationMaximumWireOffset(media) { + return errors.New("qualification public-wire transition invalid") + } + if _, duplicate := transitions[reduction]; duplicate { + return errors.New("qualification public-wire transition duplicated") + } + transitions[reduction] = time.Duration(after) + transitionRows++ + if after < lastAfter { + return errors.New("qualification public-wire records not monotonic") + } + lastAfter = after + case "delivery": + if row[1] != "" || row[2] != "" || row[3] == "" || row[4] == "" { + return errors.New("qualification public-wire delivery fields invalid") + } + after, afterErr := strconv.ParseInt(row[3], 10, 64) + encodedBytes, bytesErr := strconv.ParseInt(row[4], 10, 64) + if afterErr != nil || bytesErr != nil || after < 0 || time.Duration(after) > qualificationMaximumWireOffset(media) || encodedBytes <= 0 { + return errors.New("qualification public-wire delivery invalid") + } + if after < lastAfter { + return errors.New("qualification public-wire records not monotonic") + } + if encodedBytes != expectedLengths[deliveryRows%len(expectedLengths)] { + return errors.New("qualification public-wire encoded length or logical-frame substitution invalid") + } + lastAfter = after + deliveries = append(deliveries, qualificationDeliverySample{At: epoch.Add(time.Duration(after)), Bytes: encodedBytes}) + deliveryRows++ + default: + return errors.New("qualification public-wire record type invalid") + } + } + if rows != observation.RawWireRows || deliveryRows != observation.RawWireDeliveryRows || + transitionRows != observation.RawWireTransitionRows || deliveryRows != observation.QUICDatagramsSent || + transitionRows != qualificationCanonicalCapacityStepCount || deliveryRows%len(expectedLengths) != 0 { + return errors.New("qualification public-wire retained counts mismatch") + } + if len(observation.CapacityStepObservations) != qualificationCanonicalCapacityStepCount { + return errors.New("qualification public-wire capacity summary count mismatch") + } + if _, ok := transitions[25]; !ok { + return errors.New("qualification public-wire transition missing") + } + if _, ok := transitions[50]; !ok { + return errors.New("qualification public-wire transition missing") + } + if transitions[25] >= transitions[50] { + return errors.New("qualification public-wire transition order invalid") + } + for index, reduction := range qualificationCanonicalCapacitySteps { + transition, ok := transitions[reduction] + if !ok { + return errors.New("qualification public-wire transition missing") + } + step := observation.CapacityStepObservations[index] + if step.ReductionPercent != reduction || step.TransitionAfter != transition || + step.RecomputationSource != observation.RawWireSamples { + return errors.New("qualification public-wire manifest transition mismatch") + } + start := epoch.Add(transition) + first := sort.Search(len(deliveries), func(index int) bool { return !deliveries[index].At.Before(start) }) + afterStep := deliveries[first:] + target := qualificationMediaPacerKbps(media, reduction) * 1000 / 8 + convergence := qualificationMeasuredConvergence(afterStep, start, target) + maximum := qualificationMaximumDeliveryBytes(afterStep, 5*time.Second) + if step.Convergence != convergence || step.MaximumFiveSecond != maximum || step.FiveSecondCap != target*5 { + return fmt.Errorf( + "qualification public-wire capacity summary mismatch reduction=%d convergence=%s/%s maximum=%d/%d cap=%d/%d", + reduction, step.Convergence, convergence, step.MaximumFiveSecond, maximum, step.FiveSecondCap, target*5, + ) + } + if mode == qualificationEvidenceNormative && + (step.Convergence > 10*time.Second || step.MaximumFiveSecond > step.FiveSecondCap*105/100) { + return fmt.Errorf("qualification normative capacity gate failed reduction=%d convergence=%s maximum=%d cap=%d", + reduction, step.Convergence, step.MaximumFiveSecond, step.FiveSecondCap) + } + } + return nil +} + +func validateQualificationFairnessBundle(directory string, evidence qualificationFairnessEvidence) error { + if evidence.RawSamples == "" || filepath.Base(evidence.RawSamples) != evidence.RawSamples { + return errors.New("qualification fairness raw samples missing") + } + path := filepath.Join(directory, evidence.RawSamples) + limits := qualificationFairnessCSVLimits() + info, err := os.Stat(path) + if err != nil { + return err + } + if info.Size() > limits.compressedBytes { + return errors.New("qualification fairness compressed byte bound exceeded") + } + digest, size, err := qualificationFileSHA256(path) + if err != nil { + return err + } + if digest != evidence.RawSamplesSHA256 || size != evidence.RawSamplesBytes { + return errors.New("qualification fairness hash or size mismatch") + } + reader, err := openQualificationBoundedCSV(path, limits, 3) + if err != nil { + return err + } + defer reader.Close() + header, err := reader.Read() + if err != nil || !reflect.DeepEqual(header, []string{"elapsed_ns", "flow", "bytes"}) { + return errors.New("qualification fairness schema invalid") + } + epoch := time.Unix(0, 0) + deliveries := make([]qualificationFlowDelivery, 0, qualificationImpairmentMaxPackets) + flowSet := make(map[string]struct{}, 8) + allowedFlows := make(map[string]struct{}, 8) + for _, flow := range qualificationFairnessFlows() { + allowedFlows[flow] = struct{}{} + } + lastElapsed := int64(-1) + for { + row, readErr := reader.Read() + if errors.Is(readErr, io.EOF) { + break + } + if readErr != nil { + return readErr + } + if len(deliveries) >= qualificationImpairmentMaxPackets { + return errors.New("qualification fairness row bound exceeded") + } + elapsed, elapsedErr := strconv.ParseInt(row[0], 10, 64) + encodedBytes, bytesErr := strconv.ParseInt(row[2], 10, 64) + _, knownFlow := allowedFlows[row[1]] + if elapsedErr != nil || bytesErr != nil || elapsed < 0 || time.Duration(elapsed) > qualificationMaximumFairnessOffset() || + elapsed < lastElapsed || !knownFlow || encodedBytes != int64(1000+frameHeaderSize) { + return errors.New("qualification fairness row invalid") + } + lastElapsed = elapsed + flowSet[row[1]] = struct{}{} + deliveries = append(deliveries, qualificationFlowDelivery{at: epoch.Add(time.Duration(elapsed)), flow: row[1], bytes: encodedBytes}) + } + if len(deliveries) == 0 || len(flowSet) != 8 || len(evidence.CapacitySteps) != 2 { + return errors.New("qualification fairness retained counts invalid") + } + flows := make([]string, 0, len(flowSet)) + for flow := range flowSet { + flows = append(flows, flow) + } + sort.Strings(flows) + previousTransition := time.Duration(-1) + for index, reduction := range []int{25, 50} { + step := evidence.CapacitySteps[index] + if step.ReductionPercent != reduction || step.TransitionAfter < 0 || step.TransitionAfter <= previousTransition || + step.RecomputationSource != evidence.RawSamples { + return errors.New("qualification fairness transition manifest invalid") + } + previousTransition = step.TransitionAfter + start := epoch.Add(step.TransitionAfter) + first := sort.Search(len(deliveries), func(index int) bool { return !deliveries[index].at.Before(start) }) + last := len(deliveries) + if index+1 < len(evidence.CapacitySteps) { + next := epoch.Add(evidence.CapacitySteps[index+1].TransitionAfter) + last = sort.Search(len(deliveries), func(index int) bool { return !deliveries[index].at.Before(next) }) + } + afterStep := append([]qualificationFlowDelivery(nil), deliveries[first:last]...) + target := int64(8_000*1000/8) * int64(100-reduction) / 100 + convergence := qualificationPacerConvergence(afterStep, start, flows, target) + maximum := qualificationMaximumFiveSecondBytes(append([]qualificationFlowDelivery(nil), afterStep...)) + if convergence != 2*time.Second || step.Convergence != convergence || + step.MaximumFiveSecond != maximum || step.FiveSecondCap != target*5 || + step.MaximumFiveSecond > step.FiveSecondCap*105/100 { + return fmt.Errorf( + "qualification fairness capacity summary mismatch reduction=%d transition=%s convergence=%s/%s maximum=%d/%d cap=%d/%d", + reduction, step.TransitionAfter, step.Convergence, convergence, + step.MaximumFiveSecond, maximum, step.FiveSecondCap, target*5, + ) + } + } + return nil +} + +func TestQualificationFairnessTransitionsAreIndependentlyRecomputable(t *testing.T) { + epoch := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC) + flows := []string{"one", "two", "three", "four", "five", "six", "seven", "eight"} + deliveries := make([]qualificationFlowDelivery, 0, 5_520) + for _, stage := range []struct { + start, end time.Duration + packets int + }{ + {0, 2 * time.Second, 12}, + {2 * time.Second, 5 * time.Second, 9}, + {5 * time.Second, 8 * time.Second, 6}, + } { + for elapsed := stage.start; elapsed < stage.end; elapsed += 100 * time.Millisecond { + for packet := 0; packet < stage.packets; packet++ { + for _, flow := range flows { + deliveries = append(deliveries, qualificationFlowDelivery{ + at: epoch.Add(elapsed + time.Duration(packet)*time.Millisecond), flow: flow, bytes: int64(1000 + frameHeaderSize), + }) + } + } + } + } + directory := t.TempDir() + path := filepath.Join(directory, "fairness.csv.gz") + if err := writeQualificationPacerSamples(path, deliveries, epoch); err != nil { + t.Fatal(err) + } + evidence := qualificationFairnessEvidence{RawSamples: filepath.Base(path)} + steps := []struct { + reduction int + transition time.Duration + }{ + {25, 2 * time.Second}, + {50, 5 * time.Second}, + } + for stepIndex, step := range steps { + start := epoch.Add(step.transition) + first := sort.Search(len(deliveries), func(index int) bool { return !deliveries[index].at.Before(start) }) + last := len(deliveries) + if stepIndex+1 < len(steps) { + next := epoch.Add(steps[stepIndex+1].transition) + last = sort.Search(len(deliveries), func(index int) bool { return !deliveries[index].at.Before(next) }) + } + afterStep := append([]qualificationFlowDelivery(nil), deliveries[first:last]...) + target := int64(8_000*1000/8) * int64(100-step.reduction) / 100 + evidence.CapacitySteps = append(evidence.CapacitySteps, qualificationCapacityStep{ + ReductionPercent: step.reduction, TransitionAfter: step.transition, + Convergence: qualificationPacerConvergence(afterStep, start, flows, target), + MaximumFiveSecond: qualificationMaximumFiveSecondBytes(append([]qualificationFlowDelivery(nil), afterStep...)), + FiveSecondCap: target * 5, RecomputationSource: filepath.Base(path), + }) + } + evidence.RawSamplesSHA256, evidence.RawSamplesBytes, _ = qualificationFileSHA256(path) + if err := validateQualificationFairnessBundle(directory, evidence); err != nil { + t.Fatal(err) + } +} + +func readQualificationTestCSV(t *testing.T, path string) [][]string { + t.Helper() + file, err := os.Open(path) + if err != nil { + t.Fatal(err) + } + defer file.Close() + compressed, err := gzip.NewReader(file) + if err != nil { + t.Fatal(err) + } + defer compressed.Close() + rows, err := csv.NewReader(compressed).ReadAll() + if err != nil { + t.Fatal(err) + } + return rows +} + +func writeQualificationTestCSV(t *testing.T, path string, rows [][]string) { + t.Helper() + file, err := os.Create(path) + if err != nil { + t.Fatal(err) + } + compressed := gzip.NewWriter(file) + buffered := bufio.NewWriter(compressed) + writer := csv.NewWriter(buffered) + writer.WriteAll(rows) + if err := writer.Error(); err != nil { + t.Fatal(err) + } + if err := buffered.Flush(); err != nil { + t.Fatal(err) + } + if err := compressed.Close(); err != nil { + t.Fatal(err) + } + if err := file.Close(); err != nil { + t.Fatal(err) + } +} + +func qualificationTestWireBundle(t *testing.T, sent int) (string, qualificationImpairmentObservation, qualificationMediaProfile, []qualificationDeliverySample, map[int]time.Time) { + t.Helper() + media := qualificationMediaProfiles()[0] + epoch := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC) + transitions := map[int]time.Time{25: epoch.Add(time.Second), 50: epoch.Add(2 * time.Second)} + deliveries := []qualificationDeliverySample{ + {At: transitions[25], Bytes: 1200}, + {At: transitions[25].Add(time.Nanosecond), Bytes: 25}, + {At: transitions[50], Bytes: 1200}, + {At: transitions[50].Add(time.Nanosecond), Bytes: 25}, + } + directory := t.TempDir() + path := filepath.Join(directory, "impairment-constrained-1080p60-h264-wire.csv.gz") + evidence, err := writeQualificationWireSamples(path, epoch, []int{25, 50}, transitions, deliveries, len(deliveries)+2) + if err != nil { + t.Fatal(err) + } + observation := qualificationImpairmentObservation{ + Sent: sent, ConfiguredCapacitySteps: []int{25, 50}, QUICDatagramsSent: len(deliveries), + RawWireSamples: evidence.Name, RawWireSamplesSHA256: evidence.SHA256, RawWireSamplesBytes: evidence.Bytes, + RawWireRows: evidence.Rows, RawWireDeliveryRows: evidence.DeliveryRows, + RawWireTransitionRows: evidence.TransitionRows, RawWireTimebase: qualificationWireTimebase, + } + for _, reduction := range observation.ConfiguredCapacitySteps { + step := qualificationCapacityStepObservation(deliveries, transitions[reduction], media, reduction) + step.TransitionAfter = transitions[reduction].Sub(epoch) + step.RecomputationSource = evidence.Name + observation.CapacityStepObservations = append(observation.CapacityStepObservations, step) + } + return directory, observation, media, deliveries, transitions +} + +func TestQualificationCapacityStepRetainsFullPostTransitionTail(t *testing.T) { + media := qualificationMediaProfiles()[0] + epoch := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC) + quarter := epoch.Add(time.Second) + half := epoch.Add(3 * time.Second) + deliveries := make([]qualificationDeliverySample, 0, 120) + for at := quarter; at.Before(epoch.Add(7 * time.Second)); at = at.Add(100 * time.Millisecond) { + deliveries = append(deliveries, + qualificationDeliverySample{At: at, Bytes: 1200}, + qualificationDeliverySample{At: at.Add(time.Nanosecond), Bytes: 25}, + ) + } + got := qualificationCapacityStepObservation(deliveries, quarter, media, 25) + fullTail := qualificationDeliveriesAfter(deliveries, quarter) + wantMaximum := qualificationMaximumDeliveryBytes(fullTail, 5*time.Second) + wantConvergence := qualificationMeasuredConvergence(fullTail, quarter, qualificationMediaPacerKbps(media, 25)*1000/8) + stageEnd := sort.Search(len(deliveries), func(index int) bool { return !deliveries[index].At.Before(half) }) + stageMaximum := qualificationMaximumDeliveryBytes(deliveries[:stageEnd], 5*time.Second) + if wantMaximum <= stageMaximum { + t.Fatalf("test fixture full-tail maximum=%d stage-only=%d", wantMaximum, stageMaximum) + } + if got.MaximumFiveSecond != wantMaximum || got.Convergence != wantConvergence { + t.Fatalf("25%% summary used shortened stage: got maximum=%d convergence=%s, full-tail maximum=%d convergence=%s, stage-only=%d", + got.MaximumFiveSecond, got.Convergence, wantMaximum, wantConvergence, stageMaximum) + } +} + +func TestQualificationNormativeValidationCannotTrustPersistedSent(t *testing.T) { + directory, observation, media, _, _ := qualificationTestWireBundle(t, qualificationImpairmentPacketCount-1) + if err := validateQualificationWireBundle(directory, observation, media, qualificationEvidenceNormative); err == nil { + t.Fatal("normative validation accepted persisted sent=9999 and 11-second convergence sentinel") + } +} + +func TestQualificationWireBundleRejectsReversedCapacitySteps(t *testing.T) { + directory, observation, media, _, _ := qualificationTestWireBundle(t, 40) + observation.ConfiguredCapacitySteps = []int{50, 25} + observation.CapacityStepObservations[0], observation.CapacityStepObservations[1] = + observation.CapacityStepObservations[1], observation.CapacityStepObservations[0] + if err := validateQualificationWireBundle(directory, observation, media, qualificationEvidenceSmoke); err == nil { + t.Fatal("qualification wire bundle accepted reversed capacity steps") + } +} + +func TestQualificationWireBundleRejectsSwappedTransitionChronology(t *testing.T) { + directory, observation, media, deliveries, transitions := qualificationTestWireBundle(t, 40) + rows := readQualificationTestCSV(t, filepath.Join(directory, observation.RawWireSamples)) + for _, row := range rows[1:] { + if row[0] != "transition" { + continue + } + switch row[1] { + case "25": + row[1] = "50" + case "50": + row[1] = "25" + } + } + mutatedDirectory := t.TempDir() + path := filepath.Join(mutatedDirectory, observation.RawWireSamples) + writeQualificationTestCSV(t, path, rows) + observation.RawWireSamplesSHA256, observation.RawWireSamplesBytes, _ = qualificationFileSHA256(path) + transitions[25], transitions[50] = transitions[50], transitions[25] + observation.CapacityStepObservations = nil + epoch := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC) + for _, reduction := range observation.ConfiguredCapacitySteps { + step := qualificationCapacityStepObservation(deliveries, transitions[reduction], media, reduction) + step.TransitionAfter = transitions[reduction].Sub(epoch) + step.RecomputationSource = observation.RawWireSamples + observation.CapacityStepObservations = append(observation.CapacityStepObservations, step) + } + if err := validateQualificationWireBundle(mutatedDirectory, observation, media, qualificationEvidenceSmoke); err == nil || + !strings.Contains(err.Error(), "transition order invalid") { + t.Fatalf("qualification swapped transition chronology rejection = %v", err) + } +} + +func TestQualificationWireBundleRejectsSelfConsistentNoncanonicalSteps(t *testing.T) { + directory, observation, media, deliveries, transitions := qualificationTestWireBundle(t, 40) + rows := readQualificationTestCSV(t, filepath.Join(directory, observation.RawWireSamples)) + for _, row := range rows[1:] { + if row[0] != "transition" { + continue + } + switch row[1] { + case "25": + row[1] = "10" + case "50": + row[1] = "20" + } + } + mutatedDirectory := t.TempDir() + path := filepath.Join(mutatedDirectory, observation.RawWireSamples) + writeQualificationTestCSV(t, path, rows) + observation.RawWireSamplesSHA256, observation.RawWireSamplesBytes, _ = qualificationFileSHA256(path) + observation.ConfiguredCapacitySteps = []int{10, 20} + observation.CapacityStepObservations = nil + for index, reduction := range observation.ConfiguredCapacitySteps { + transition := transitions[[]int{25, 50}[index]] + step := qualificationCapacityStepObservation(deliveries, transition, media, reduction) + step.TransitionAfter = transition.Sub(time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)) + step.RecomputationSource = observation.RawWireSamples + observation.CapacityStepObservations = append(observation.CapacityStepObservations, step) + } + if err := validateQualificationWireBundle(mutatedDirectory, observation, media, qualificationEvidenceSmoke); err == nil { + t.Fatal("qualification wire bundle accepted self-consistent noncanonical capacity steps") + } +} + +func TestQualificationWireBundleRejectsNegativeDeliveryRowsWithoutPanic(t *testing.T) { + directory, observation, media, _, _ := qualificationTestWireBundle(t, 40) + observation.RawWireRows = 1 + observation.RawWireDeliveryRows = -1 + observation.RawWireTransitionRows = 2 + defer func() { + if recovered := recover(); recovered != nil { + t.Fatalf("qualification wire bundle panicked for negative delivery rows: %v", recovered) + } + }() + if err := validateQualificationWireBundle(directory, observation, media, qualificationEvidenceSmoke); err == nil { + t.Fatal("qualification wire bundle accepted negative delivery rows") + } +} + +func TestQualificationNormativeRejectsExactSentConvergenceFailure(t *testing.T) { + directory, observation, media, _, _ := qualificationTestWireBundle(t, qualificationImpairmentPacketCount) + err := validateQualificationWireBundle(directory, observation, media, qualificationEvidenceNormative) + if err == nil || !strings.Contains(err.Error(), "normative capacity gate failed") || strings.Contains(err.Error(), "sent=") { + t.Fatalf("normative exact-sent convergence rejection = %v", err) + } +} + +func TestQualificationEvidenceReadersRejectOversizedFieldsBeforeParsing(t *testing.T) { + t.Run("wire", func(t *testing.T) { + directory, observation, media, _, _ := qualificationTestWireBundle(t, 40) + rows := readQualificationTestCSV(t, filepath.Join(directory, observation.RawWireSamples)) + for index := 1; index < len(rows); index++ { + if rows[index][0] == "delivery" { + rows[index][3] = strings.Repeat("9", 4096) + break + } + } + mutatedDirectory := t.TempDir() + path := filepath.Join(mutatedDirectory, observation.RawWireSamples) + writeQualificationTestCSV(t, path, rows) + observation.RawWireSamplesSHA256, observation.RawWireSamplesBytes, _ = qualificationFileSHA256(path) + if err := validateQualificationWireBundle(mutatedDirectory, observation, media, qualificationEvidenceSmoke); err == nil || !strings.Contains(err.Error(), "field bound") { + t.Fatalf("wire oversized field rejection = %v", err) + } + }) + + t.Run("fairness", func(t *testing.T) { + epoch := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC) + deliveries := make([]qualificationFlowDelivery, 0, 8) + for _, flow := range []string{"one", "two", "three", "four", "five", "six", "seven", "eight"} { + deliveries = append(deliveries, qualificationFlowDelivery{at: epoch, flow: flow, bytes: 1023}) + } + directory := t.TempDir() + path := filepath.Join(directory, "fairness.csv.gz") + if err := writeQualificationPacerSamples(path, deliveries, epoch); err != nil { + t.Fatal(err) + } + rows := readQualificationTestCSV(t, path) + rows[1][1] = strings.Repeat("f", 4096) + mutatedDirectory := t.TempDir() + mutatedPath := filepath.Join(mutatedDirectory, "fairness.csv.gz") + writeQualificationTestCSV(t, mutatedPath, rows) + evidence := qualificationFairnessEvidence{RawSamples: "fairness.csv.gz", CapacitySteps: []qualificationCapacityStep{{}, {}}} + evidence.RawSamplesSHA256, evidence.RawSamplesBytes, _ = qualificationFileSHA256(mutatedPath) + if err := validateQualificationFairnessBundle(mutatedDirectory, evidence); err == nil || !strings.Contains(err.Error(), "field bound") { + t.Fatalf("fairness oversized field rejection = %v", err) + } + }) +} + +func TestQualificationBoundedCSVReaderRejectsCompressedAndDecompressedOverflow(t *testing.T) { + t.Run("compressed", func(t *testing.T) { + path := filepath.Join(t.TempDir(), "oversized.csv.gz") + file, err := os.Create(path) + if err != nil { + t.Fatal(err) + } + if err := file.Truncate(33); err != nil { + t.Fatal(err) + } + if err := file.Close(); err != nil { + t.Fatal(err) + } + _, err = openQualificationBoundedCSV(path, qualificationCSVLimits{compressedBytes: 32, decompressedBytes: 32, headerBytes: []int{5}, fieldBytes: []int{64}}, 1) + if err == nil || !strings.Contains(err.Error(), "compressed byte bound") { + t.Fatalf("compressed overflow rejection = %v", err) + } + }) + + t.Run("decompressed", func(t *testing.T) { + path := filepath.Join(t.TempDir(), "oversized.csv.gz") + writeQualificationTestCSV(t, path, [][]string{{"value"}, {strings.Repeat("x", 64)}}) + reader, err := openQualificationBoundedCSV(path, qualificationCSVLimits{ + compressedBytes: 1024, decompressedBytes: 32, headerBytes: []int{5}, fieldBytes: []int{64}, + }, 1) + if err != nil { + t.Fatal(err) + } + defer reader.Close() + if _, err := reader.Read(); err == nil || !strings.Contains(err.Error(), "decompressed byte bound") { + t.Fatalf("decompressed overflow rejection = %v", err) + } + }) +} + +func TestQualificationWireEvidenceRejectsIncompleteOrMalformedBundles(t *testing.T) { + profile := qualificationImpairmentProfiles()[5] + media := qualificationMediaProfiles()[0] + sourceDirectory := t.TempDir() + observation, err := runQualificationImpairment(t, profile, media, 40, + filepath.Join(sourceDirectory, "impairment-constrained-1080p60-h264.csv.gz")) + if err != nil { + t.Fatal(err) + } + if err := validateQualificationWireBundle(sourceDirectory, observation, media, qualificationEvidenceSmoke); err != nil { + t.Fatal(err) + } + wireRows := readQualificationTestCSV(t, filepath.Join(sourceDirectory, observation.RawWireSamples)) + + metadataCases := []struct { + name string + mutate func(*qualificationImpairmentObservation) + }{ + {"missing-artifact", func(value *qualificationImpairmentObservation) { value.RawWireSamples = "" }}, + {"wrong-hash", func(value *qualificationImpairmentObservation) { value.RawWireSamplesSHA256 = strings.Repeat("0", 64) }}, + {"wrong-size", func(value *qualificationImpairmentObservation) { value.RawWireSamplesBytes++ }}, + {"wrong-count", func(value *qualificationImpairmentObservation) { value.RawWireRows++ }}, + {"manifest-mismatch", func(value *qualificationImpairmentObservation) { value.CapacityStepObservations[0].Convergence++ }}, + } + for _, testCase := range metadataCases { + t.Run(testCase.name, func(t *testing.T) { + mutated := observation + mutated.CapacityStepObservations = append([]qualificationCapacityStep(nil), observation.CapacityStepObservations...) + testCase.mutate(&mutated) + if err := validateQualificationWireBundle(sourceDirectory, mutated, media, qualificationEvidenceSmoke); err == nil { + t.Fatal("malformed qualification wire bundle accepted") + } + }) + } + + type rowMutation struct { + name string + mutate func([][]string, *qualificationImpairmentObservation) [][]string + } + rowCases := []rowMutation{ + {"duplicate-transition", func(rows [][]string, value *qualificationImpairmentObservation) [][]string { + for index := 1; index < len(rows); index++ { + if rows[index][0] == "transition" { + duplicate := append([]string(nil), rows[index]...) + rows = append(rows, nil) + copy(rows[index+1:], rows[index:]) + rows[index] = duplicate + value.RawWireRows++ + value.RawWireTransitionRows++ + break + } + } + return rows + }}, + {"missing-transition", func(rows [][]string, value *qualificationImpairmentObservation) [][]string { + for index := 1; index < len(rows); index++ { + if rows[index][0] == "transition" { + copy(rows[index:], rows[index+1:]) + rows[len(rows)-1] = nil + value.RawWireRows-- + value.RawWireTransitionRows-- + break + } + } + return rows + }}, + {"delivery-before-epoch", func(rows [][]string, _ *qualificationImpairmentObservation) [][]string { + for index := 1; index < len(rows); index++ { + if rows[index][0] == "delivery" { + rows[index][3] = "-1" + break + } + } + return rows + }}, + {"invalid-encoded-length", func(rows [][]string, _ *qualificationImpairmentObservation) [][]string { + for index := 1; index < len(rows); index++ { + if rows[index][0] == "delivery" { + rows[index][4] = "0" + break + } + } + return rows + }}, + {"logical-frame-substitution", func(rows [][]string, _ *qualificationImpairmentObservation) [][]string { + for index := 1; index < len(rows); index++ { + if rows[index][0] == "delivery" { + rows[index][4] = "1179" + break + } + } + return rows + }}, + {"invalid-empty-fields", func(rows [][]string, _ *qualificationImpairmentObservation) [][]string { + for index := 1; index < len(rows); index++ { + if rows[index][0] == "delivery" { + rows[index][1] = "25" + break + } + } + return rows + }}, + {"nonmonotonic", func(rows [][]string, _ *qualificationImpairmentObservation) [][]string { + first := int64(-1) + for index := 1; index < len(rows); index++ { + if rows[index][0] != "delivery" { + continue + } + if first < 0 { + first, _ = strconv.ParseInt(rows[index][3], 10, 64) + continue + } + rows[index][3] = strconv.FormatInt(max(first-1, 0), 10) + break + } + return rows + }}, + } + for _, testCase := range rowCases { + t.Run(testCase.name, func(t *testing.T) { + rows := make([][]string, len(wireRows)) + for index := range wireRows { + rows[index] = append([]string(nil), wireRows[index]...) + } + mutated := observation + mutated.CapacityStepObservations = append([]qualificationCapacityStep(nil), observation.CapacityStepObservations...) + rows = testCase.mutate(rows, &mutated) + trimmed := rows[:0] + for _, row := range rows { + if row != nil { + trimmed = append(trimmed, row) + } + } + directory := t.TempDir() + path := filepath.Join(directory, mutated.RawWireSamples) + writeQualificationTestCSV(t, path, trimmed) + mutated.RawWireSamplesSHA256, mutated.RawWireSamplesBytes, _ = qualificationFileSHA256(path) + if err := validateQualificationWireBundle(directory, mutated, media, qualificationEvidenceSmoke); err == nil { + t.Fatal("malformed qualification wire bundle accepted") + } + }) } } @@ -750,6 +1680,14 @@ func TestQualificationCapacityStepsMeasurePublicWireDatagrams(t *testing.T) { } wireDeliveries[index] = sample.delivery } + directory := t.TempDir() + wirePath := filepath.Join(directory, "impairment-constrained-1080p60-h264-wire.csv.gz") + bundle := qualificationImpairmentObservation{ + Sent: packetCount, + ConfiguredCapacitySteps: []int{25, 50}, + QUICDatagramsSent: len(wireDeliveries), + RawWireTimebase: qualificationWireTimebase, + } for _, reduction := range []int{25, 50} { wireBytesPerSecond := qualificationMediaPacerKbps(profile, reduction) * 1000 / 8 @@ -765,8 +1703,10 @@ func TestQualificationCapacityStepsMeasurePublicWireDatagrams(t *testing.T) { if reduction == 25 && legacy != 11*time.Second { t.Fatalf("25%% complete-frame classifier convergence = %s, want 11s sentinel reproduction", legacy) } - observation := qualificationCapacityStepObservation(wireDeliveries, stepAt[reduction], profile, reduction) - if observation.Convergence > 10*time.Second { + stepObservation := qualificationCapacityStepObservation(wireDeliveries, stepAt[reduction], profile, reduction) + stepObservation.TransitionAfter = stepAt[reduction].Sub(started) + stepObservation.RecomputationSource = filepath.Base(wirePath) + if stepObservation.Convergence > 10*time.Second { for offset := time.Duration(0); offset < 2*time.Second; offset += 250 * time.Millisecond { var bucket int64 for _, delivery := range wireAfterStep { @@ -776,16 +1716,31 @@ func TestQualificationCapacityStepsMeasurePublicWireDatagrams(t *testing.T) { } t.Logf("%d%% wire bucket %s = %d bytes (%d B/s)", reduction, offset, bucket, bucket*4) } - t.Fatalf("%d%% wire convergence = %s sentinel with wire maximum %d under cap %d", reduction, observation.Convergence, maximum, wireBytesPerSecond*5*105/100) + t.Fatalf("%d%% wire convergence = %s sentinel with wire maximum %d under cap %d", reduction, stepObservation.Convergence, maximum, wireBytesPerSecond*5*105/100) } t.Logf("%d%%: payload_target=%d wire_target=%d legacy=%s wire_convergence=%s wire_max_5s=%d wire_cap_105=%d", - reduction, payloadBytesPerSecond, wireBytesPerSecond, legacy, observation.Convergence, maximum, wireBytesPerSecond*5*105/100) + reduction, payloadBytesPerSecond, wireBytesPerSecond, legacy, stepObservation.Convergence, maximum, wireBytesPerSecond*5*105/100) + bundle.CapacityStepObservations = append(bundle.CapacityStepObservations, stepObservation) + } + wireEvidence, err := writeQualificationWireSamples(wirePath, started, []int{25, 50}, stepAt, wireDeliveries, len(wireDeliveries)+2) + if err != nil { + t.Fatal(err) + } + bundle.RawWireSamples = wireEvidence.Name + bundle.RawWireSamplesSHA256 = wireEvidence.SHA256 + bundle.RawWireSamplesBytes = wireEvidence.Bytes + bundle.RawWireRows = wireEvidence.Rows + bundle.RawWireDeliveryRows = wireEvidence.DeliveryRows + bundle.RawWireTransitionRows = wireEvidence.TransitionRows + if err := validateQualificationWireBundle(directory, bundle, profile, qualificationEvidenceNormative); err != nil { + t.Fatal(err) } } func TestQualificationUsesPublicQUICAndProductionPacer(t *testing.T) { qualificationTraverseProfiles(t, qualificationMediaProfiles()) - evidence, err := qualificationPacerEvidence(t, filepath.Join(t.TempDir(), "fairness.csv.gz"), 2*time.Second, 4*time.Second) + directory := t.TempDir() + evidence, err := qualificationPacerEvidence(t, filepath.Join(directory, "fairness.csv.gz"), 2*time.Second, 4*time.Second) if err != nil { t.Fatal(err) } @@ -798,6 +1753,9 @@ func TestQualificationUsesPublicQUICAndProductionPacer(t *testing.T) { t.Fatalf("capacity step = %#v", step) } } + if err := validateQualificationFairnessBundle(directory, evidence); err != nil { + t.Fatal(err) + } } func TestQualificationSmokeTraversesNativeApolloRecoveryQueuePacerAndQUIC(t *testing.T) { diff --git a/gateway/qualification_harness_test.go b/gateway/qualification_harness_test.go index a2f324a..966e8ec 100644 --- a/gateway/qualification_harness_test.go +++ b/gateway/qualification_harness_test.go @@ -10,6 +10,7 @@ import ( "crypto/sha256" "crypto/tls" "encoding/binary" + "encoding/csv" "encoding/hex" "encoding/json" "errors" @@ -38,7 +39,7 @@ import ( ) const ( - qualificationToolVersion = "versevdi-gateway-qualification/v9" + qualificationToolVersion = "versevdi-gateway-qualification/v10" qualificationImpairmentQueuePackets = nativeApolloVideoQueuePackets qualificationImpairmentMaxPackets = 100_000 qualificationImpairmentPacketCount = 10_000 @@ -50,6 +51,7 @@ const ( qualificationApolloVideoRateBitsPerSecond = 1_000_000_000 * 80 / 100 qualificationApolloVideoBatchBytes = 64 * 1024 qualificationApolloVideoBatchPackets = 64 + qualificationWireTimebase = "monotonic offsets from constrained run start" ) type qualificationMediaProfile struct { @@ -204,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 { @@ -227,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"` @@ -1507,7 +1527,10 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro beforeProviderDrops := path.session.mediaDrops.Load() beforeSourceUDP := path.sourceUDP.Load() beforePacer := path.server.pacer.reservations.Load() - wireDeliveries := make([]qualificationDeliverySample, 0, len(jobs)*2) + 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 { @@ -1528,6 +1551,10 @@ func runQualificationImpairment(t *testing.T, profile qualificationImpairmentPro seen := make([]bool, packetCount) for len(result.packets) < len(jobs) { 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 { @@ -1601,6 +1628,9 @@ 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 totalLatency, totalJitter, previousLatency time.Duration previousDelivered := -1 @@ -1713,12 +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 { 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 && (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", reduction, step.Convergence, step.MaximumFiveSecond) + 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 || @@ -2128,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 @@ -2189,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() @@ -2246,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) @@ -2263,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 { diff --git a/openspec/changes/truthful-fixed-frame-qualification/design.md b/openspec/changes/truthful-fixed-frame-qualification/design.md index 085c0c4..b51b6b7 100644 --- a/openspec/changes/truthful-fixed-frame-qualification/design.md +++ b/openspec/changes/truthful-fixed-frame-qualification/design.md @@ -8,7 +8,7 @@ Because the bounded fixture uses loopback rather than a physical 1 Gbps link, v8 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 v9 normative run remain open. +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. @@ -22,6 +22,12 @@ Qualification v9 observes every raw public QUIC datagram immediately after the i 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. @@ -49,6 +55,8 @@ The native provider therefore requests `2,048 * 1,072 = 2,195,456` bytes with `S - 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 diff --git a/openspec/changes/truthful-fixed-frame-qualification/proposal.md b/openspec/changes/truthful-fixed-frame-qualification/proposal.md index e0ce032..e1d1cd5 100644 --- a/openspec/changes/truthful-fixed-frame-qualification/proposal.md +++ b/openspec/changes/truthful-fixed-frame-qualification/proposal.md @@ -12,6 +12,8 @@ Private Linux run 127/job 481 at exact source `122080ab342d20585d9a45db0017337b9 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. @@ -20,6 +22,7 @@ Private Linux run 128/job 482 was the single push-triggered attempt at exact sou - 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 @@ -34,4 +37,4 @@ None. ## Impact -The qualification harness, its canonical specification, native Apollo video ingress in `gateway/apollo_native.go`, and the existing production fair pacer in `gateway/telemetry.go`. 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. +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. diff --git a/openspec/changes/truthful-fixed-frame-qualification/specs/gateway-qualification/spec.md b/openspec/changes/truthful-fixed-frame-qualification/specs/gateway-qualification/spec.md index 5015156..1a04ce6 100644 --- a/openspec/changes/truthful-fixed-frame-qualification/specs/gateway-qualification/spec.md +++ b/openspec/changes/truthful-fixed-frame-qualification/specs/gateway-qualification/spec.md @@ -9,7 +9,11 @@ Native Apollo video ingress SHALL request a 2,195,456-byte socket receive buffer 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. A capacity step SHALL include only delivery observations at or after its recorded transition, 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. +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 @@ -39,6 +43,22 @@ Capacity-step convergence and rolling-cap evidence SHALL use the monotonic recei - **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. diff --git a/openspec/changes/truthful-fixed-frame-qualification/tasks.md b/openspec/changes/truthful-fixed-frame-qualification/tasks.md index 3e910ed..1a1a636 100644 --- a/openspec/changes/truthful-fixed-frame-qualification/tasks.md +++ b/openspec/changes/truthful-fixed-frame-qualification/tasks.md @@ -23,8 +23,8 @@ - [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 v9 measurement descendant -- [ ] 5.5 Run one separately approved replacement v9 normative Section 7 qualification +- [ ] 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 @@ -42,6 +42,11 @@ ## 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 Record raw independent-client QUIC datagram lengths/times and derive v9 capacity convergence and caps from configured wire rate while keeping logical payload observations separate +- [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