fix(obs): sample complete measurement intervals

commit: 9597b551a842ce7eaa79fc1da59e763797d9696a
author: brice · 3mo ago · 2026-07-13 00:20 UTC
parent: dd4fb64
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +329 -45

M cmd/vev-perf-harness/main.go +191 -39


@@ -25,8 +25,10 @@ import (
25	25		)
26	26
27	27		const (
28		-	minimumDuration = 30 * time.Second
29		-	minimumRepetitions = 10
	28	+	minimumDuration = 30 * time.Second
	29	+	minimumRepetitions = 10
	30	+	minimumInIntervalEventSamples = 10
	31	+	measuredEventCadence = time.Second
30	32		)
31	33
32	34		type options struct {

@@ -85,27 +87,48 @@ type harnessMark struct {
85	87		Tick int64 `json:"tick"`
86	88		Valid bool `json:"valid"`
87	89		}
	90	+
	91	+	// measuredInterval belongs solely to the harness clock domain. A paired event
	92	+	// is eligible only when both owned boundaries fall inside this interval.
	93	+	type measuredInterval struct {
	94	+	Start int64
	95	+	End int64
	96	+	}
	97	+
	98	+	type eventSample struct {
	99	+	Sequence uint64 `json:"sequence"`
	100	+	Injected int64 `json:"injected_tick"`
	101	+	Flushed int64 `json:"flushed_tick"`
	102	+	Latency int64 `json:"latency_nanos"`
	103	+	}
88	104		type span struct {
89	105		Component, Name string
90	106		Samples []int64
91	107		}
92	108		type runResult struct {
93		-	Spans []span `json:"process_local_spans"`
94		-	Scenario string `json:"scenario"`
95		-	Run int `json:"run"`
96		-	Samples int `json:"samples"`
97		-	EndToEnd []int64 `json:"end_to_end_nanos"`
98		-	Processes []processMapping `json:"processes"`
	109	+	Spans []span `json:"process_local_spans"`
	110	+	Scenario string `json:"scenario"`
	111	+	Run int `json:"run"`
	112	+	Samples int `json:"samples"`
	113	+	EndToEnd []int64 `json:"end_to_end_nanos"`
	114	+	Event distribution `json:"event_end_to_end"`
	115	+	Cadence distribution `json:"event_cadence_nanos"`
	116	+	CadenceSamples []int64 `json:"-"`
	117	+	MaxGap int64 `json:"event_max_gap_nanos"`
	118	+	Processes []processMapping `json:"processes"`
99	119		}
100	120		type summary struct {
101		-	Schema uint16 `json:"schema"`
102		-	GitSHA string `json:"git_sha"`
103		-	Warmup string `json:"warmup"`
104		-	Duration string `json:"duration"`
105		-	Repetitions int `json:"repetitions"`
106		-	EndToEnd distribution `json:"harness_end_to_end"`
107		-	Spans []spanSummary `json:"process_local_spans"`
108		-	Runs int `json:"runs"`
	121	+	Schema uint16 `json:"schema"`
	122	+	GitSHA string `json:"git_sha"`
	123	+	Warmup string `json:"warmup"`
	124	+	Duration string `json:"duration"`
	125	+	Repetitions int `json:"repetitions"`
	126	+	EndToEnd distribution `json:"harness_end_to_end"`
	127	+	Cadence distribution `json:"harness_event_cadence_nanos"`
	128	+	MaxGap int64 `json:"harness_event_max_gap_nanos"`
	129	+	RunP50Dispersion distribution `json:"run_p50_dispersion"`
	130	+	Spans []spanSummary `json:"process_local_spans"`
	131	+	Runs int `json:"runs"`
109	132		}
110	133		type distribution struct {
111	134		Count int `json:"count"`

@@ -768,7 +791,6 @@ func run(args []string, h *harness) error {
768	791		}
769	792		defer raw.Close()
770	793		results := make([]runResult, 0, len(m.Scenarios)*opt.repetitions)
771		-	var all []int64
772	794		var localSpans []span
773	795		scenarios := m.Scenarios
774	796		if h.selectScenarios != nil {

@@ -784,18 +806,33 @@ func run(args []string, h *harness) error {
784	806		return fmt.Errorf("scenario %s run %d: %w", s.ID, r, err)
785	807		}
786	808		results = append(results, rr)
787		-	all = append(all, rr.EndToEnd...)
788	809		localSpans = append(localSpans, rr.Spans...)
789	810		}
790	811		}
	812	+	all, cadence, maxGap, runP50s := aggregateEventResults(results)
791	813		if len(all) == 0 {
792	814		return errors.New("no measured harness input/flush pairs")
793	815		}
794	816		if err := writeJSON(filepath.Join(opt.out, "runs.json"), results); err != nil {
795	817		return err
796	818		}
797		-	return writeJSON(filepath.Join(opt.out, "summary.json"), summary{Schema: 1, GitSHA: h.gitSHA(), Warmup: opt.warmup.String(), Duration: opt.duration.String(), Repetitions: opt.repetitions, EndToEnd: percentiles(all), Spans: summarizeSpans(localSpans), Runs: len(results)})
	819	+	return writeJSON(filepath.Join(opt.out, "summary.json"), summary{Schema: 1, GitSHA: h.gitSHA(), Warmup: opt.warmup.String(), Duration: opt.duration.String(), Repetitions: opt.repetitions, EndToEnd: percentiles(all), Cadence: percentiles(cadence), MaxGap: maxGap, RunP50Dispersion: percentiles(runP50s), Spans: summarizeSpans(localSpans), Runs: len(results)})
	820	+	}
	821	+
	822	+	// aggregateEventResults deliberately preserves raw event latencies for
	823	+	// end-to-end percentiles. Per-run p50s are a separate dispersion series.
	824	+	func aggregateEventResults(results []runResult) (all, cadence []int64, maxGap int64, runP50s []int64) {
	825	+	for _, result := range results {
	826	+	all = append(all, result.EndToEnd...)
	827	+	cadence = append(cadence, result.CadenceSamples...)
	828	+	runP50s = append(runP50s, result.Event.P50)
	829	+	if result.MaxGap > maxGap {
	830	+	maxGap = result.MaxGap
	831	+	}
	832	+	}
	833	+	return all, cadence, maxGap, runP50s
798	834		}
	835	+
799	836		func parseOptions(args []string) (options, error) {
800	837		var o options
801	838		fs := flag.NewFlagSet("vev-perf-harness", flag.ContinueOnError)

@@ -1036,35 +1073,57 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr
1036	1073		}
1037	1074		}
1038	1075		// The harness clock, rather than a process clock, owns the explicit warmup
1039		-	// interval. Fake clocks make this path instant in unit tests.
	1076	+	// and measured intervals. Warmup boundaries are intentionally never added to
	1077	+	// marks, and each measured event must complete before interval.End to count.
1040	1078		h.clock.Sleep(o.warmup)
	1079	+	interval := measuredInterval{Start: h.clock.Now()}
	1080	+	interval.End = interval.Start + o.duration.Nanoseconds()
1041	1081		marks := []harnessMark{}
1042	1082		seq := uint64(0)
1043		-	for _, p := range processes {
1044		-	if p.mapping.Role != "client" {
1045		-	continue
1046		-	}
1047		-	seq++
1048		-	sequence := seq
1049		-	injected := func() error {
1050		-	marks = append(marks, harnessMark{s.ID, run, sequence, "input_injected", h.clock.Now(), true})
1051		-	return nil
	1083	+	nextInjection := interval.Start
	1084	+	for {
	1085	+	now := h.clock.Now()
	1086	+	if now < nextInjection {
	1087	+	h.clock.Sleep(time.Duration(nextInjection - now))
	1088	+	now = h.clock.Now()
1052	1089		}
1053		-	flushed := func() error {
1054		-	marks = append(marks, harnessMark{s.ID, run, sequence, "terminal_flushed", h.clock.Now(), true})
1055		-	return nil
	1090	+	if now >= interval.End {
	1091	+	break
1056	1092		}
1057		-	if err = p.process.Measure(workloadInput(s, run, fmt.Sprintf("measured-%d", sequence)), injected, flushed); err != nil {
1058		-	return res, err
	1093	+	for _, p := range processes {
	1094	+	if p.mapping.Role != "client" {
	1095	+	continue
	1096	+	}
	1097	+	seq++
	1098	+	sequence := seq
	1099	+	injected := func() error {
	1100	+	marks = append(marks, harnessMark{s.ID, run, sequence, "input_injected", h.clock.Now(), true})
	1101	+	return nil
	1102	+	}
	1103	+	flushed := func() error {
	1104	+	marks = append(marks, harnessMark{s.ID, run, sequence, "terminal_flushed", h.clock.Now(), true})
	1105	+	return nil
	1106	+	}
	1107	+	if err = p.process.Measure(workloadInput(s, run, fmt.Sprintf("measured-%d", sequence)), injected, flushed); err != nil {
	1108	+	return res, err
	1109	+	}
1059	1110		}
	1111	+	// Cap injection rate rather than bursting events after a slow terminal
	1112	+	// flush. Any resulting spacing is retained as raw cadence evidence.
	1113	+	nextInjection = h.clock.Now() + measuredEventCadence.Nanoseconds()
	1114	+	}
	1115	+	if now := h.clock.Now(); now < interval.End {
	1116	+	h.clock.Sleep(time.Duration(interval.End - now))
1060	1117		}
1061		-	// A measured run is never shorter than the requested duration in the
1062		-	// production clock domain. The terminal boundary stays inside that interval.
1063		-	h.clock.Sleep(o.duration)
1064		-	samples, e := pairedSamples(marks)
	1118	+	events, e := measuredEventSamples(marks, interval)
1065	1119		if e != nil {
1066	1120		return res, e
1067	1121		}
	1122	+	if err := requireMinimumEventSamples(events); err != nil {
	1123	+	return res, err
	1124	+	}
	1125	+	samples := eventLatencies(events)
	1126	+	cadenceSamples, maxGap := eventCadenceSamples(events)
1068	1127		// Process teardown can unblock a receive and cause its failed end mark to
1069	1128		// be serialized. Merge only the complete, post-cleanup per-process traces.
1070	1129		if cleanupErr := closeRoles(); cleanupErr != nil {

@@ -1074,13 +1133,20 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr
1074	1133		if e != nil {
1075	1134		return res, e
1076	1135		}
	1136	+	inInterval := make(map[uint64]bool, len(events))
	1137	+	for _, event := range events {
	1138	+	inInterval[event.Sequence] = true
	1139	+	}
1077	1140		for _, m := range marks {
	1141	+	if !inInterval[m.Sequence] {
	1142	+	continue
	1143	+	}
1078	1144		if _, e := fmt.Fprintf(raw, "%s\n", mustJSON(m)); e != nil {
1079	1145		return res, e
1080	1146		}
1081	1147		}
1082	1148		success = true
1083		-	return runResult{Spans: spans, Scenario: s.ID, Run: run, Samples: len(samples), EndToEnd: samples, Processes: maps}, nil
	1149	+	return runResult{Spans: spans, Scenario: s.ID, Run: run, Samples: len(samples), EndToEnd: samples, Event: percentiles(samples), Cadence: percentiles(cadenceSamples), CadenceSamples: cadenceSamples, MaxGap: maxGap, Processes: maps}, nil
1084	1150		}
1085	1151
1086	1152		func roleArgs(s scenario, m processMapping) roleCommand {

@@ -1179,6 +1245,92 @@ func workloadInput(s scenario, run int, phase string) []byte {
1179	1245		return []byte(fmt.Sprintf(body+"; printf '%s\\n'\n", s.ID, marker))
1180	1246		}
1181	1247
	1248	+	// measuredEventSamples validates every owned pair, then retains only events
	1249	+	// whose injection and successful terminal flush are both in the interval.
	1250	+	// Straddling pairs remain raw diagnostic facts but are never percentile input.
	1251	+	func measuredEventSamples(marks []harnessMark, interval measuredInterval) ([]eventSample, error) {
	1252	+	if interval.End < interval.Start {
	1253	+	return nil, errors.New("negative measured interval")
	1254	+	}
	1255	+	starts := map[uint64]int64{}
	1256	+	events := make([]eventSample, 0, len(marks)/2)
	1257	+	for _, m := range marks {
	1258	+	if !m.Valid {
	1259	+	return nil, errors.New("invalid harness boundary")
	1260	+	}
	1261	+	switch m.Kind {
	1262	+	case "input_injected":
	1263	+	if _, exists := starts[m.Sequence]; exists {
	1264	+	return nil, errors.New("duplicate input boundary")
	1265	+	}
	1266	+	starts[m.Sequence] = m.Tick
	1267	+	case "terminal_flushed":
	1268	+	start, ok := starts[m.Sequence]
	1269	+	if !ok {
	1270	+	return nil, errors.New("terminal flush without input pair")
	1271	+	}
	1272	+	if m.Tick < start {
	1273	+	return nil, errors.New("negative harness duration")
	1274	+	}
	1275	+	delete(starts, m.Sequence)
	1276	+	if start >= interval.Start && m.Tick <= interval.End {
	1277	+	events = append(events, eventSample{Sequence: m.Sequence, Injected: start, Flushed: m.Tick, Latency: m.Tick - start})
	1278	+	}
	1279	+	default:
	1280	+	return nil, errors.New("unknown harness mark")
	1281	+	}
	1282	+	}
	1283	+	if len(starts) != 0 {
	1284	+	return nil, errors.New("missing terminal flush pair")
	1285	+	}
	1286	+	sort.Slice(events, func(i, j int) bool {
	1287	+	if events[i].Injected == events[j].Injected {
	1288	+	return events[i].Sequence < events[j].Sequence
	1289	+	}
	1290	+	return events[i].Injected < events[j].Injected
	1291	+	})
	1292	+	return events, nil
	1293	+	}
	1294	+
	1295	+	func requireMinimumEventSamples(events []eventSample) error {
	1296	+	if len(events) < minimumInIntervalEventSamples {
	1297	+	return fmt.Errorf("insufficient in-interval event samples: got %d, need at least %d", len(events), minimumInIntervalEventSamples)
	1298	+	}
	1299	+	return nil
	1300	+	}
	1301	+
	1302	+	func eventLatencies(events []eventSample) []int64 {
	1303	+	out := make([]int64, len(events))
	1304	+	for i, event := range events {
	1305	+	out[i] = event.Latency
	1306	+	}
	1307	+	return out
	1308	+	}
	1309	+
	1310	+	func eventCadenceSamples(events []eventSample) ([]int64, int64) {
	1311	+	if len(events) < 2 {
	1312	+	return nil, 0
	1313	+	}
	1314	+	gaps := make([]int64, 0, len(events)-1)
	1315	+	var maxGap int64
	1316	+	for i := 1; i < len(events); i++ {
	1317	+	gap := events[i].Injected - events[i-1].Injected
	1318	+	gaps = append(gaps, gap)
	1319	+	if gap > maxGap {
	1320	+	maxGap = gap
	1321	+	}
	1322	+	}
	1323	+	return gaps, maxGap
	1324	+	}
	1325	+
	1326	+	func eventCadence(events []eventSample) (distribution, int64) {
	1327	+	gaps, maxGap := eventCadenceSamples(events)
	1328	+	if len(gaps) == 0 {
	1329	+	return distribution{}, maxGap
	1330	+	}
	1331	+	return percentiles(gaps), maxGap
	1332	+	}
	1333	+
1182	1334		func pairedSamples(marks []harnessMark) ([]int64, error) {
1183	1335		starts := map[uint64]int64{}
1184	1336		var out []int64


@@ -25,8 +25,10 @@ import (
25		)	25		)
26			26
27		const (	27		const (
28	-	minimumDuration = 30 * time.Second	28	+	minimumDuration = 30 * time.Second
29	-	minimumRepetitions = 10	29	+	minimumRepetitions = 10
			30	+	minimumInIntervalEventSamples = 10
			31	+	measuredEventCadence = time.Second
30		)	32		)
31			33
32		type options struct {	34		type options struct {

@@ -85,27 +87,48 @@ type harnessMark struct {
85		Tick int64 `json:"tick"`	87		Tick int64 `json:"tick"`
86		Valid bool `json:"valid"`	88		Valid bool `json:"valid"`
87		}	89		}
			90	+
			91	+	// measuredInterval belongs solely to the harness clock domain. A paired event
			92	+	// is eligible only when both owned boundaries fall inside this interval.
			93	+	type measuredInterval struct {
			94	+	Start int64
			95	+	End int64
			96	+	}
			97	+
			98	+	type eventSample struct {
			99	+	Sequence uint64 `json:"sequence"`
			100	+	Injected int64 `json:"injected_tick"`
			101	+	Flushed int64 `json:"flushed_tick"`
			102	+	Latency int64 `json:"latency_nanos"`
			103	+	}
88		type span struct {	104		type span struct {
89		Component, Name string	105		Component, Name string
90		Samples []int64	106		Samples []int64
91		}	107		}
92		type runResult struct {	108		type runResult struct {
93	-	Spans []span `json:"process_local_spans"`	109	+	Spans []span `json:"process_local_spans"`
94	-	Scenario string `json:"scenario"`	110	+	Scenario string `json:"scenario"`
95	-	Run int `json:"run"`	111	+	Run int `json:"run"`
96	-	Samples int `json:"samples"`	112	+	Samples int `json:"samples"`
97	-	EndToEnd []int64 `json:"end_to_end_nanos"`	113	+	EndToEnd []int64 `json:"end_to_end_nanos"`
98	-	Processes []processMapping `json:"processes"`	114	+	Event distribution `json:"event_end_to_end"`
			115	+	Cadence distribution `json:"event_cadence_nanos"`
			116	+	CadenceSamples []int64 `json:"-"`
			117	+	MaxGap int64 `json:"event_max_gap_nanos"`
			118	+	Processes []processMapping `json:"processes"`
99		}	119		}
100		type summary struct {	120		type summary struct {
101	-	Schema uint16 `json:"schema"`	121	+	Schema uint16 `json:"schema"`
102	-	GitSHA string `json:"git_sha"`	122	+	GitSHA string `json:"git_sha"`
103	-	Warmup string `json:"warmup"`	123	+	Warmup string `json:"warmup"`
104	-	Duration string `json:"duration"`	124	+	Duration string `json:"duration"`
105	-	Repetitions int `json:"repetitions"`	125	+	Repetitions int `json:"repetitions"`
106	-	EndToEnd distribution `json:"harness_end_to_end"`	126	+	EndToEnd distribution `json:"harness_end_to_end"`
107	-	Spans []spanSummary `json:"process_local_spans"`	127	+	Cadence distribution `json:"harness_event_cadence_nanos"`
108	-	Runs int `json:"runs"`	128	+	MaxGap int64 `json:"harness_event_max_gap_nanos"`
			129	+	RunP50Dispersion distribution `json:"run_p50_dispersion"`
			130	+	Spans []spanSummary `json:"process_local_spans"`
			131	+	Runs int `json:"runs"`
109		}	132		}
110		type distribution struct {	133		type distribution struct {
111		Count int `json:"count"`	134		Count int `json:"count"`

@@ -768,7 +791,6 @@ func run(args []string, h *harness) error {
768		}	791		}
769		defer raw.Close()	792		defer raw.Close()
770		results := make([]runResult, 0, len(m.Scenarios)*opt.repetitions)	793		results := make([]runResult, 0, len(m.Scenarios)*opt.repetitions)
771	-	var all []int64
772		var localSpans []span	794		var localSpans []span
773		scenarios := m.Scenarios	795		scenarios := m.Scenarios
774		if h.selectScenarios != nil {	796		if h.selectScenarios != nil {

@@ -784,18 +806,33 @@ func run(args []string, h *harness) error {
784		return fmt.Errorf("scenario %s run %d: %w", s.ID, r, err)	806		return fmt.Errorf("scenario %s run %d: %w", s.ID, r, err)
785		}	807		}
786		results = append(results, rr)	808		results = append(results, rr)
787	-	all = append(all, rr.EndToEnd...)
788		localSpans = append(localSpans, rr.Spans...)	809		localSpans = append(localSpans, rr.Spans...)
789		}	810		}
790		}	811		}
			812	+	all, cadence, maxGap, runP50s := aggregateEventResults(results)
791		if len(all) == 0 {	813		if len(all) == 0 {
792		return errors.New("no measured harness input/flush pairs")	814		return errors.New("no measured harness input/flush pairs")
793		}	815		}
794		if err := writeJSON(filepath.Join(opt.out, "runs.json"), results); err != nil {	816		if err := writeJSON(filepath.Join(opt.out, "runs.json"), results); err != nil {
795		return err	817		return err
796		}	818		}
797	-	return writeJSON(filepath.Join(opt.out, "summary.json"), summary{Schema: 1, GitSHA: h.gitSHA(), Warmup: opt.warmup.String(), Duration: opt.duration.String(), Repetitions: opt.repetitions, EndToEnd: percentiles(all), Spans: summarizeSpans(localSpans), Runs: len(results)})	819	+	return writeJSON(filepath.Join(opt.out, "summary.json"), summary{Schema: 1, GitSHA: h.gitSHA(), Warmup: opt.warmup.String(), Duration: opt.duration.String(), Repetitions: opt.repetitions, EndToEnd: percentiles(all), Cadence: percentiles(cadence), MaxGap: maxGap, RunP50Dispersion: percentiles(runP50s), Spans: summarizeSpans(localSpans), Runs: len(results)})
			820	+	}
			821	+
			822	+	// aggregateEventResults deliberately preserves raw event latencies for
			823	+	// end-to-end percentiles. Per-run p50s are a separate dispersion series.
			824	+	func aggregateEventResults(results []runResult) (all, cadence []int64, maxGap int64, runP50s []int64) {
			825	+	for _, result := range results {
			826	+	all = append(all, result.EndToEnd...)
			827	+	cadence = append(cadence, result.CadenceSamples...)
			828	+	runP50s = append(runP50s, result.Event.P50)
			829	+	if result.MaxGap > maxGap {
			830	+	maxGap = result.MaxGap
			831	+	}
			832	+	}
			833	+	return all, cadence, maxGap, runP50s
798		}	834		}
			835	+
799		func parseOptions(args []string) (options, error) {	836		func parseOptions(args []string) (options, error) {
800		var o options	837		var o options
801		fs := flag.NewFlagSet("vev-perf-harness", flag.ContinueOnError)	838		fs := flag.NewFlagSet("vev-perf-harness", flag.ContinueOnError)

@@ -1036,35 +1073,57 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr
1036		}	1073		}
1037		}	1074		}
1038		// The harness clock, rather than a process clock, owns the explicit warmup	1075		// The harness clock, rather than a process clock, owns the explicit warmup
1039	-	// interval. Fake clocks make this path instant in unit tests.	1076	+	// and measured intervals. Warmup boundaries are intentionally never added to
			1077	+	// marks, and each measured event must complete before interval.End to count.
1040		h.clock.Sleep(o.warmup)	1078		h.clock.Sleep(o.warmup)
			1079	+	interval := measuredInterval{Start: h.clock.Now()}
			1080	+	interval.End = interval.Start + o.duration.Nanoseconds()
1041		marks := []harnessMark{}	1081		marks := []harnessMark{}
1042		seq := uint64(0)	1082		seq := uint64(0)
1043	-	for _, p := range processes {	1083	+	nextInjection := interval.Start
1044	-	if p.mapping.Role != "client" {	1084	+	for {
1045	-	continue	1085	+	now := h.clock.Now()
1046	-	}	1086	+	if now < nextInjection {
1047	-	seq++	1087	+	h.clock.Sleep(time.Duration(nextInjection - now))
1048	-	sequence := seq	1088	+	now = h.clock.Now()
1049	-	injected := func() error {
1050	-	marks = append(marks, harnessMark{s.ID, run, sequence, "input_injected", h.clock.Now(), true})
1051	-	return nil
1052		}	1089		}
1053	-	flushed := func() error {	1090	+	if now >= interval.End {
1054	-	marks = append(marks, harnessMark{s.ID, run, sequence, "terminal_flushed", h.clock.Now(), true})	1091	+	break
1055	-	return nil
1056		}	1092		}
1057	-	if err = p.process.Measure(workloadInput(s, run, fmt.Sprintf("measured-%d", sequence)), injected, flushed); err != nil {	1093	+	for _, p := range processes {
1058	-	return res, err	1094	+	if p.mapping.Role != "client" {
			1095	+	continue
			1096	+	}
			1097	+	seq++
			1098	+	sequence := seq
			1099	+	injected := func() error {
			1100	+	marks = append(marks, harnessMark{s.ID, run, sequence, "input_injected", h.clock.Now(), true})
			1101	+	return nil
			1102	+	}
			1103	+	flushed := func() error {
			1104	+	marks = append(marks, harnessMark{s.ID, run, sequence, "terminal_flushed", h.clock.Now(), true})
			1105	+	return nil
			1106	+	}
			1107	+	if err = p.process.Measure(workloadInput(s, run, fmt.Sprintf("measured-%d", sequence)), injected, flushed); err != nil {
			1108	+	return res, err
			1109	+	}
1059		}	1110		}
			1111	+	// Cap injection rate rather than bursting events after a slow terminal
			1112	+	// flush. Any resulting spacing is retained as raw cadence evidence.
			1113	+	nextInjection = h.clock.Now() + measuredEventCadence.Nanoseconds()
			1114	+	}
			1115	+	if now := h.clock.Now(); now < interval.End {
			1116	+	h.clock.Sleep(time.Duration(interval.End - now))
1060		}	1117		}
1061	-	// A measured run is never shorter than the requested duration in the	1118	+	events, e := measuredEventSamples(marks, interval)
1062	-	// production clock domain. The terminal boundary stays inside that interval.
1063	-	h.clock.Sleep(o.duration)
1064	-	samples, e := pairedSamples(marks)
1065		if e != nil {	1119		if e != nil {
1066		return res, e	1120		return res, e
1067		}	1121		}
			1122	+	if err := requireMinimumEventSamples(events); err != nil {
			1123	+	return res, err
			1124	+	}
			1125	+	samples := eventLatencies(events)
			1126	+	cadenceSamples, maxGap := eventCadenceSamples(events)
1068		// Process teardown can unblock a receive and cause its failed end mark to	1127		// Process teardown can unblock a receive and cause its failed end mark to
1069		// be serialized. Merge only the complete, post-cleanup per-process traces.	1128		// be serialized. Merge only the complete, post-cleanup per-process traces.
1070		if cleanupErr := closeRoles(); cleanupErr != nil {	1129		if cleanupErr := closeRoles(); cleanupErr != nil {

@@ -1074,13 +1133,20 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr
1074		if e != nil {	1133		if e != nil {
1075		return res, e	1134		return res, e
1076		}	1135		}
			1136	+	inInterval := make(map[uint64]bool, len(events))
			1137	+	for _, event := range events {
			1138	+	inInterval[event.Sequence] = true
			1139	+	}
1077		for _, m := range marks {	1140		for _, m := range marks {
			1141	+	if !inInterval[m.Sequence] {
			1142	+	continue
			1143	+	}
1078		if _, e := fmt.Fprintf(raw, "%s\n", mustJSON(m)); e != nil {	1144		if _, e := fmt.Fprintf(raw, "%s\n", mustJSON(m)); e != nil {
1079		return res, e	1145		return res, e
1080		}	1146		}
1081		}	1147		}
1082		success = true	1148		success = true
1083	-	return runResult{Spans: spans, Scenario: s.ID, Run: run, Samples: len(samples), EndToEnd: samples, Processes: maps}, nil	1149	+	return runResult{Spans: spans, Scenario: s.ID, Run: run, Samples: len(samples), EndToEnd: samples, Event: percentiles(samples), Cadence: percentiles(cadenceSamples), CadenceSamples: cadenceSamples, MaxGap: maxGap, Processes: maps}, nil
1084		}	1150		}
1085			1151
1086		func roleArgs(s scenario, m processMapping) roleCommand {	1152		func roleArgs(s scenario, m processMapping) roleCommand {

@@ -1179,6 +1245,92 @@ func workloadInput(s scenario, run int, phase string) []byte {
1179		return []byte(fmt.Sprintf(body+"; printf '%s\\n'\n", s.ID, marker))	1245		return []byte(fmt.Sprintf(body+"; printf '%s\\n'\n", s.ID, marker))
1180		}	1246		}
1181			1247
			1248	+	// measuredEventSamples validates every owned pair, then retains only events
			1249	+	// whose injection and successful terminal flush are both in the interval.
			1250	+	// Straddling pairs remain raw diagnostic facts but are never percentile input.
			1251	+	func measuredEventSamples(marks []harnessMark, interval measuredInterval) ([]eventSample, error) {
			1252	+	if interval.End < interval.Start {
			1253	+	return nil, errors.New("negative measured interval")
			1254	+	}
			1255	+	starts := map[uint64]int64{}
			1256	+	events := make([]eventSample, 0, len(marks)/2)
			1257	+	for _, m := range marks {
			1258	+	if !m.Valid {
			1259	+	return nil, errors.New("invalid harness boundary")
			1260	+	}
			1261	+	switch m.Kind {
			1262	+	case "input_injected":
			1263	+	if _, exists := starts[m.Sequence]; exists {
			1264	+	return nil, errors.New("duplicate input boundary")
			1265	+	}
			1266	+	starts[m.Sequence] = m.Tick
			1267	+	case "terminal_flushed":
			1268	+	start, ok := starts[m.Sequence]
			1269	+	if !ok {
			1270	+	return nil, errors.New("terminal flush without input pair")
			1271	+	}
			1272	+	if m.Tick < start {
			1273	+	return nil, errors.New("negative harness duration")
			1274	+	}
			1275	+	delete(starts, m.Sequence)
			1276	+	if start >= interval.Start && m.Tick <= interval.End {
			1277	+	events = append(events, eventSample{Sequence: m.Sequence, Injected: start, Flushed: m.Tick, Latency: m.Tick - start})
			1278	+	}
			1279	+	default:
			1280	+	return nil, errors.New("unknown harness mark")
			1281	+	}
			1282	+	}
			1283	+	if len(starts) != 0 {
			1284	+	return nil, errors.New("missing terminal flush pair")
			1285	+	}
			1286	+	sort.Slice(events, func(i, j int) bool {
			1287	+	if events[i].Injected == events[j].Injected {
			1288	+	return events[i].Sequence < events[j].Sequence
			1289	+	}
			1290	+	return events[i].Injected < events[j].Injected
			1291	+	})
			1292	+	return events, nil
			1293	+	}
			1294	+
			1295	+	func requireMinimumEventSamples(events []eventSample) error {
			1296	+	if len(events) < minimumInIntervalEventSamples {
			1297	+	return fmt.Errorf("insufficient in-interval event samples: got %d, need at least %d", len(events), minimumInIntervalEventSamples)
			1298	+	}
			1299	+	return nil
			1300	+	}
			1301	+
			1302	+	func eventLatencies(events []eventSample) []int64 {
			1303	+	out := make([]int64, len(events))
			1304	+	for i, event := range events {
			1305	+	out[i] = event.Latency
			1306	+	}
			1307	+	return out
			1308	+	}
			1309	+
			1310	+	func eventCadenceSamples(events []eventSample) ([]int64, int64) {
			1311	+	if len(events) < 2 {
			1312	+	return nil, 0
			1313	+	}
			1314	+	gaps := make([]int64, 0, len(events)-1)
			1315	+	var maxGap int64
			1316	+	for i := 1; i < len(events); i++ {
			1317	+	gap := events[i].Injected - events[i-1].Injected
			1318	+	gaps = append(gaps, gap)
			1319	+	if gap > maxGap {
			1320	+	maxGap = gap
			1321	+	}
			1322	+	}
			1323	+	return gaps, maxGap
			1324	+	}
			1325	+
			1326	+	func eventCadence(events []eventSample) (distribution, int64) {
			1327	+	gaps, maxGap := eventCadenceSamples(events)
			1328	+	if len(gaps) == 0 {
			1329	+	return distribution{}, maxGap
			1330	+	}
			1331	+	return percentiles(gaps), maxGap
			1332	+	}
			1333	+
1182		func pairedSamples(marks []harnessMark) ([]int64, error) {	1334		func pairedSamples(marks []harnessMark) ([]int64, error) {
1183		starts := map[uint64]int64{}	1335		starts := map[uint64]int64{}
1184		var out []int64	1336		var out []int64

M cmd/vev-perf-harness/main_test.go +138 -6


@@ -16,8 +16,10 @@ import (
16	16
17	17		type fakeClock struct{ tick int64 }
18	18
19		-	func (c *fakeClock) Now() int64 { c.tick += 10; return c.tick }
20		-	func (*fakeClock) Sleep(time.Duration) {}
	19	+	func (c *fakeClock) Now() int64 { c.tick += 10; return c.tick }
	20	+	func (c *fakeClock) Sleep(d time.Duration) {
	21	+	c.tick += d.Nanoseconds()
	22	+	}
21	23
22	24		type fakeUDPNetem struct {
23	25		port int

@@ -76,6 +78,27 @@ func (l *fakeLauncher) Launch(m processMapping, command roleCommand) (launchedPr
76	78		return p, nil
77	79		}
78	80
	81	+	type lateFlushProcess struct{ clock *fakeClock }
	82	+
	83	+	func (*lateFlushProcess) Warmup([]byte) error { return nil }
	84	+	func (p *lateFlushProcess) Measure(_ []byte, injected, flushed func() error) error {
	85	+	if err := injected(); err != nil {
	86	+	return err
	87	+	}
	88	+	p.clock.tick += minimumDuration.Nanoseconds()
	89	+	return flushed()
	90	+	}
	91	+	func (*lateFlushProcess) Close() error { return nil }
	92	+
	93	+	type insufficientEventLauncher struct{ clock *fakeClock }
	94	+
	95	+	func (l insufficientEventLauncher) Launch(m processMapping, _ roleCommand) (launchedProcess, error) {
	96	+	if m.Role == "client" {
	97	+	return &lateFlushProcess{clock: l.clock}, nil
	98	+	}
	99	+	return &fakeProcess{}, nil
	100	+	}
	101	+
79	102		func TestCLIProcessCloseIsBoundedWhenPublicRoleDoesNotExit(t *testing.T) {
80	103		deadline := make(chan time.Time)
81	104		close(deadline)

@@ -138,7 +161,7 @@ func TestHarnessCanonicalLocalSmokeRealRoles(t *testing.T) {
138	161		if err != nil {
139	162		t.Fatal(err)
140	163		}
141		-	if result.Samples != 1 \|\| len(result.Processes) != 2 {
	164	+	if result.Samples < minimumInIntervalEventSamples \|\| len(result.Processes) != 2 {
142	165		t.Fatalf("canonical local smoke result=%+v", result)
143	166		}
144	167		}

@@ -241,7 +264,7 @@ func TestHarnessCreatesExclusiveTraceManifestAndEvidence(t *testing.T) {
241	264		if !launcher.manifestPresent {
242	265		t.Fatal("launched before complete run manifest")
243	266		}
244		-	if len(r.EndToEnd) != 1 \|\| r.Samples != 1 {
	267	+	if len(r.EndToEnd) < minimumInIntervalEventSamples \|\| r.Samples < minimumInIntervalEventSamples {
245	268		t.Fatalf("samples=%+v", r)
246	269		}
247	270		seen := map[string]bool{}

@@ -265,7 +288,7 @@ func TestHarnessCreatesExclusiveTraceManifestAndEvidence(t *testing.T) {
265	288		if len(got.Processes) != 2 {
266	289		t.Fatalf("manifest=%+v", got)
267	290		}
268		-	if len(launcher.process) != 2 \|\| !launcher.process[0].closed \|\| len(launcher.process[0].warmups) != 0 \|\| len(launcher.process[1].warmups) != 1 \|\| len(launcher.process[1].measures) != 1 {
	291	+	if len(launcher.process) != 2 \|\| !launcher.process[0].closed \|\| len(launcher.process[0].warmups) != 0 \|\| len(launcher.process[1].warmups) != 1 \|\| len(launcher.process[1].measures) < minimumInIntervalEventSamples {
269	292		t.Fatalf("process lifecycle not recorded: %+v", launcher.process)
270	293		}
271	294		}

@@ -676,6 +699,115 @@ func TestHarnessRejectsInvalidBoundaryPairs(t *testing.T) {
676	699		}
677	700		}
678	701
	702	+	func TestHarnessCollectsOnlyCompleteMeasuredIntervalEvents(t *testing.T) {
	703	+	interval := measuredInterval{Start: 100, End: 200}
	704	+	marks := []harnessMark{
	705	+	{Sequence: 1, Kind: "input_injected", Tick: 10, Valid: true},
	706	+	{Sequence: 1, Kind: "terminal_flushed", Tick: 20, Valid: true}, // warmup
	707	+	{Sequence: 2, Kind: "input_injected", Tick: 100, Valid: true},
	708	+	{Sequence: 2, Kind: "terminal_flushed", Tick: 120, Valid: true},
	709	+	{Sequence: 3, Kind: "input_injected", Tick: 150, Valid: true},
	710	+	{Sequence: 3, Kind: "terminal_flushed", Tick: 201, Valid: true}, // exits interval
	711	+	{Sequence: 4, Kind: "input_injected", Tick: 99, Valid: true},
	712	+	{Sequence: 4, Kind: "terminal_flushed", Tick: 110, Valid: true}, // enters interval
	713	+	}
	714	+	events, err := measuredEventSamples(marks, interval)
	715	+	if err != nil {
	716	+	t.Fatal(err)
	717	+	}
	718	+	if len(events) != 1 \|\| events[0].Sequence != 2 \|\| events[0].Latency != 20 {
	719	+	t.Fatalf("measured events=%+v, want only complete sequence 2", events)
	720	+	}
	721	+	}
	722	+
	723	+	func TestHarnessRejectsInsufficientMeasuredEventSamples(t *testing.T) {
	724	+	marks := make([]harnessMark, 0, 2*(minimumInIntervalEventSamples-1))
	725	+	for i := 0; i < minimumInIntervalEventSamples-1; i++ {
	726	+	marks = append(marks,
	727	+	harnessMark{Sequence: uint64(i + 1), Kind: "input_injected", Tick: int64(i * 100), Valid: true},
	728	+	harnessMark{Sequence: uint64(i + 1), Kind: "terminal_flushed", Tick: int64(i*100 + 10), Valid: true},
	729	+	)
	730	+	}
	731	+	events, err := measuredEventSamples(marks, measuredInterval{Start: 0, End: int64(minimumInIntervalEventSamples * 100)})
	732	+	if err != nil {
	733	+	t.Fatal(err)
	734	+	}
	735	+	if err := requireMinimumEventSamples(events); err == nil {
	736	+	t.Fatal("insufficient measured event samples accepted")
	737	+	}
	738	+	}
	739	+
	740	+	func TestHarnessRunRejectsStraddlingEventsAsInsufficient(t *testing.T) {
	741	+	dir := t.TempDir()
	742	+	raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE\|os.O_EXCL\|os.O_WRONLY, 0o600)
	743	+	if err != nil {
	744	+	t.Fatal(err)
	745	+	}
	746	+	defer raw.Close()
	747	+	clock := &fakeClock{}
	748	+	h := defaultHarness()
	749	+	h.clock, h.launcher = clock, insufficientEventLauncher{clock: clock}
	750	+	_, err = h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "late", Roles: []string{"daemon", "client"}}, 1, raw)
	751	+	if err == nil \|\| !strings.Contains(err.Error(), "insufficient in-interval event samples") {
	752	+	t.Fatalf("straddling run error=%v, want insufficient in-interval sample rejection", err)
	753	+	}
	754	+	}
	755	+
	756	+	func TestHarnessCalculatesEventCadenceAndMaxGapFromRawPairs(t *testing.T) {
	757	+	events := []eventSample{
	758	+	{Sequence: 1, Injected: 100, Flushed: 110, Latency: 10},
	759	+	{Sequence: 2, Injected: 130, Flushed: 140, Latency: 10},
	760	+	{Sequence: 3, Injected: 180, Flushed: 190, Latency: 10},
	761	+	}
	762	+	cadence, maxGap := eventCadence(events)
	763	+	if cadence.Count != 2 \|\| cadence.P50 != 30 \|\| cadence.P95 != 30 \|\| cadence.P99 != 30 \|\| maxGap != 50 {
	764	+	t.Fatalf("cadence=%+v maxGap=%d", cadence, maxGap)
	765	+	}
	766	+	}
	767	+
	768	+	func TestHarnessAggregatesRawEventsSeparatelyFromRunDispersion(t *testing.T) {
	769	+	first := append([]int64(nil), make([]int64, minimumInIntervalEventSamples+5)...)
	770	+	for i := range first {
	771	+	first[i] = 1
	772	+	}
	773	+	second := []int64{100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 1_000, 1_000}
	774	+	all, _, _, runP50s := aggregateEventResults([]runResult{
	775	+	{EndToEnd: first, Event: distribution{P50: 1}},
	776	+	{EndToEnd: second, Event: distribution{P50: 100}},
	777	+	})
	778	+	if got, dispersion := percentiles(all), percentiles(runP50s); got.Count != 30 \|\| got.P99 != 1_000 \|\| dispersion.Count != 2 \|\| dispersion.P99 != 1 {
	779	+	t.Fatalf("event aggregation=%+v run dispersion=%+v", got, dispersion)
	780	+	}
	781	+	}
	782	+
	783	+	func TestHarnessSchedulesMeasuredEventsAcrossFullInterval(t *testing.T) {
	784	+	dir := t.TempDir()
	785	+	raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE\|os.O_EXCL\|os.O_WRONLY, 0o600)
	786	+	if err != nil {
	787	+	t.Fatal(err)
	788	+	}
	789	+	defer raw.Close()
	790	+	h := defaultHarness()
	791	+	h.clock, h.launcher = &fakeClock{}, &fakeLauncher{}
	792	+	result, err := h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "schedule", Roles: []string{"daemon", "client"}}, 1, raw)
	793	+	if err != nil {
	794	+	t.Fatal(err)
	795	+	}
	796	+	if result.Samples < minimumInIntervalEventSamples \|\| result.Cadence.Count != result.Samples-1 \|\| result.MaxGap <= 0 {
	797	+	t.Fatalf("result did not include repeated measured events: %+v", result)
	798	+	}
	799	+	if err := raw.Close(); err != nil {
	800	+	t.Fatal(err)
	801	+	}
	802	+	records, err := readJSONL(filepath.Join(dir, "raw.jsonl"))
	803	+	if err != nil {
	804	+	t.Fatal(err)
	805	+	}
	806	+	if len(records) != result.Samples*2 {
	807	+	t.Fatalf("raw records=%d, want %d complete measured boundaries", len(records), result.Samples*2)
	808	+	}
	809	+	}
	810	+
679	811		func TestHarnessPreservesWorkloadAndCleanupErrors(t *testing.T) {
680	812		dir := t.TempDir()
681	813		raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE\|os.O_EXCL\|os.O_WRONLY, 0o600)

@@ -895,7 +1027,7 @@ func TestHarnessWritesRequiredEvidenceWithSufficientSpans(t *testing.T) {
895	1027		if err := json.Unmarshal(b, &got); err != nil {
896	1028		t.Fatal(err)
897	1029		}
898		-	if got.EndToEnd.Count != 10 \|\| got.Repetitions != 10 \|\| len(got.Spans) == 0 {
	1030	+	if got.EndToEnd.Count < 10*minimumInIntervalEventSamples \|\| got.Repetitions != 10 \|\| got.RunP50Dispersion.Count != 10 \|\| got.Cadence.Count == 0 \|\| got.MaxGap == 0 \|\| len(got.Spans) == 0 {
899	1031		t.Fatalf("summary=%+v", got)
900	1032		}
901	1033		for _, s := range got.Spans {


@@ -16,8 +16,10 @@ import (
16			16
17		type fakeClock struct{ tick int64 }	17		type fakeClock struct{ tick int64 }
18			18
19	-	func (c *fakeClock) Now() int64 { c.tick += 10; return c.tick }	19	+	func (c *fakeClock) Now() int64 { c.tick += 10; return c.tick }
20	-	func (*fakeClock) Sleep(time.Duration) {}	20	+	func (c *fakeClock) Sleep(d time.Duration) {
			21	+	c.tick += d.Nanoseconds()
			22	+	}
21			23
22		type fakeUDPNetem struct {	24		type fakeUDPNetem struct {
23		port int	25		port int

@@ -76,6 +78,27 @@ func (l *fakeLauncher) Launch(m processMapping, command roleCommand) (launchedPr
76		return p, nil	78		return p, nil
77		}	79		}
78			80
			81	+	type lateFlushProcess struct{ clock *fakeClock }
			82	+
			83	+	func (*lateFlushProcess) Warmup([]byte) error { return nil }
			84	+	func (p *lateFlushProcess) Measure(_ []byte, injected, flushed func() error) error {
			85	+	if err := injected(); err != nil {
			86	+	return err
			87	+	}
			88	+	p.clock.tick += minimumDuration.Nanoseconds()
			89	+	return flushed()
			90	+	}
			91	+	func (*lateFlushProcess) Close() error { return nil }
			92	+
			93	+	type insufficientEventLauncher struct{ clock *fakeClock }
			94	+
			95	+	func (l insufficientEventLauncher) Launch(m processMapping, _ roleCommand) (launchedProcess, error) {
			96	+	if m.Role == "client" {
			97	+	return &lateFlushProcess{clock: l.clock}, nil
			98	+	}
			99	+	return &fakeProcess{}, nil
			100	+	}
			101	+
79		func TestCLIProcessCloseIsBoundedWhenPublicRoleDoesNotExit(t *testing.T) {	102		func TestCLIProcessCloseIsBoundedWhenPublicRoleDoesNotExit(t *testing.T) {
80		deadline := make(chan time.Time)	103		deadline := make(chan time.Time)
81		close(deadline)	104		close(deadline)

@@ -138,7 +161,7 @@ func TestHarnessCanonicalLocalSmokeRealRoles(t *testing.T) {
138		if err != nil {	161		if err != nil {
139		t.Fatal(err)	162		t.Fatal(err)
140		}	163		}
141	-	if result.Samples != 1 \|\| len(result.Processes) != 2 {	164	+	if result.Samples < minimumInIntervalEventSamples \|\| len(result.Processes) != 2 {
142		t.Fatalf("canonical local smoke result=%+v", result)	165		t.Fatalf("canonical local smoke result=%+v", result)
143		}	166		}
144		}	167		}

@@ -241,7 +264,7 @@ func TestHarnessCreatesExclusiveTraceManifestAndEvidence(t *testing.T) {
241		if !launcher.manifestPresent {	264		if !launcher.manifestPresent {
242		t.Fatal("launched before complete run manifest")	265		t.Fatal("launched before complete run manifest")
243		}	266		}
244	-	if len(r.EndToEnd) != 1 \|\| r.Samples != 1 {	267	+	if len(r.EndToEnd) < minimumInIntervalEventSamples \|\| r.Samples < minimumInIntervalEventSamples {
245		t.Fatalf("samples=%+v", r)	268		t.Fatalf("samples=%+v", r)
246		}	269		}
247		seen := map[string]bool{}	270		seen := map[string]bool{}

@@ -265,7 +288,7 @@ func TestHarnessCreatesExclusiveTraceManifestAndEvidence(t *testing.T) {
265		if len(got.Processes) != 2 {	288		if len(got.Processes) != 2 {
266		t.Fatalf("manifest=%+v", got)	289		t.Fatalf("manifest=%+v", got)
267		}	290		}
268	-	if len(launcher.process) != 2 \|\| !launcher.process[0].closed \|\| len(launcher.process[0].warmups) != 0 \|\| len(launcher.process[1].warmups) != 1 \|\| len(launcher.process[1].measures) != 1 {	291	+	if len(launcher.process) != 2 \|\| !launcher.process[0].closed \|\| len(launcher.process[0].warmups) != 0 \|\| len(launcher.process[1].warmups) != 1 \|\| len(launcher.process[1].measures) < minimumInIntervalEventSamples {
269		t.Fatalf("process lifecycle not recorded: %+v", launcher.process)	292		t.Fatalf("process lifecycle not recorded: %+v", launcher.process)
270		}	293		}
271		}	294		}

@@ -676,6 +699,115 @@ func TestHarnessRejectsInvalidBoundaryPairs(t *testing.T) {
676		}	699		}
677		}	700		}
678			701
			702	+	func TestHarnessCollectsOnlyCompleteMeasuredIntervalEvents(t *testing.T) {
			703	+	interval := measuredInterval{Start: 100, End: 200}
			704	+	marks := []harnessMark{
			705	+	{Sequence: 1, Kind: "input_injected", Tick: 10, Valid: true},
			706	+	{Sequence: 1, Kind: "terminal_flushed", Tick: 20, Valid: true}, // warmup
			707	+	{Sequence: 2, Kind: "input_injected", Tick: 100, Valid: true},
			708	+	{Sequence: 2, Kind: "terminal_flushed", Tick: 120, Valid: true},
			709	+	{Sequence: 3, Kind: "input_injected", Tick: 150, Valid: true},
			710	+	{Sequence: 3, Kind: "terminal_flushed", Tick: 201, Valid: true}, // exits interval
			711	+	{Sequence: 4, Kind: "input_injected", Tick: 99, Valid: true},
			712	+	{Sequence: 4, Kind: "terminal_flushed", Tick: 110, Valid: true}, // enters interval
			713	+	}
			714	+	events, err := measuredEventSamples(marks, interval)
			715	+	if err != nil {
			716	+	t.Fatal(err)
			717	+	}
			718	+	if len(events) != 1 \|\| events[0].Sequence != 2 \|\| events[0].Latency != 20 {
			719	+	t.Fatalf("measured events=%+v, want only complete sequence 2", events)
			720	+	}
			721	+	}
			722	+
			723	+	func TestHarnessRejectsInsufficientMeasuredEventSamples(t *testing.T) {
			724	+	marks := make([]harnessMark, 0, 2*(minimumInIntervalEventSamples-1))
			725	+	for i := 0; i < minimumInIntervalEventSamples-1; i++ {
			726	+	marks = append(marks,
			727	+	harnessMark{Sequence: uint64(i + 1), Kind: "input_injected", Tick: int64(i * 100), Valid: true},
			728	+	harnessMark{Sequence: uint64(i + 1), Kind: "terminal_flushed", Tick: int64(i*100 + 10), Valid: true},
			729	+	)
			730	+	}
			731	+	events, err := measuredEventSamples(marks, measuredInterval{Start: 0, End: int64(minimumInIntervalEventSamples * 100)})
			732	+	if err != nil {
			733	+	t.Fatal(err)
			734	+	}
			735	+	if err := requireMinimumEventSamples(events); err == nil {
			736	+	t.Fatal("insufficient measured event samples accepted")
			737	+	}
			738	+	}
			739	+
			740	+	func TestHarnessRunRejectsStraddlingEventsAsInsufficient(t *testing.T) {
			741	+	dir := t.TempDir()
			742	+	raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE\|os.O_EXCL\|os.O_WRONLY, 0o600)
			743	+	if err != nil {
			744	+	t.Fatal(err)
			745	+	}
			746	+	defer raw.Close()
			747	+	clock := &fakeClock{}
			748	+	h := defaultHarness()
			749	+	h.clock, h.launcher = clock, insufficientEventLauncher{clock: clock}
			750	+	_, err = h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "late", Roles: []string{"daemon", "client"}}, 1, raw)
			751	+	if err == nil \|\| !strings.Contains(err.Error(), "insufficient in-interval event samples") {
			752	+	t.Fatalf("straddling run error=%v, want insufficient in-interval sample rejection", err)
			753	+	}
			754	+	}
			755	+
			756	+	func TestHarnessCalculatesEventCadenceAndMaxGapFromRawPairs(t *testing.T) {
			757	+	events := []eventSample{
			758	+	{Sequence: 1, Injected: 100, Flushed: 110, Latency: 10},
			759	+	{Sequence: 2, Injected: 130, Flushed: 140, Latency: 10},
			760	+	{Sequence: 3, Injected: 180, Flushed: 190, Latency: 10},
			761	+	}
			762	+	cadence, maxGap := eventCadence(events)
			763	+	if cadence.Count != 2 \|\| cadence.P50 != 30 \|\| cadence.P95 != 30 \|\| cadence.P99 != 30 \|\| maxGap != 50 {
			764	+	t.Fatalf("cadence=%+v maxGap=%d", cadence, maxGap)
			765	+	}
			766	+	}
			767	+
			768	+	func TestHarnessAggregatesRawEventsSeparatelyFromRunDispersion(t *testing.T) {
			769	+	first := append([]int64(nil), make([]int64, minimumInIntervalEventSamples+5)...)
			770	+	for i := range first {
			771	+	first[i] = 1
			772	+	}
			773	+	second := []int64{100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 1_000, 1_000}
			774	+	all, _, _, runP50s := aggregateEventResults([]runResult{
			775	+	{EndToEnd: first, Event: distribution{P50: 1}},
			776	+	{EndToEnd: second, Event: distribution{P50: 100}},
			777	+	})
			778	+	if got, dispersion := percentiles(all), percentiles(runP50s); got.Count != 30 \|\| got.P99 != 1_000 \|\| dispersion.Count != 2 \|\| dispersion.P99 != 1 {
			779	+	t.Fatalf("event aggregation=%+v run dispersion=%+v", got, dispersion)
			780	+	}
			781	+	}
			782	+
			783	+	func TestHarnessSchedulesMeasuredEventsAcrossFullInterval(t *testing.T) {
			784	+	dir := t.TempDir()
			785	+	raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE\|os.O_EXCL\|os.O_WRONLY, 0o600)
			786	+	if err != nil {
			787	+	t.Fatal(err)
			788	+	}
			789	+	defer raw.Close()
			790	+	h := defaultHarness()
			791	+	h.clock, h.launcher = &fakeClock{}, &fakeLauncher{}
			792	+	result, err := h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "schedule", Roles: []string{"daemon", "client"}}, 1, raw)
			793	+	if err != nil {
			794	+	t.Fatal(err)
			795	+	}
			796	+	if result.Samples < minimumInIntervalEventSamples \|\| result.Cadence.Count != result.Samples-1 \|\| result.MaxGap <= 0 {
			797	+	t.Fatalf("result did not include repeated measured events: %+v", result)
			798	+	}
			799	+	if err := raw.Close(); err != nil {
			800	+	t.Fatal(err)
			801	+	}
			802	+	records, err := readJSONL(filepath.Join(dir, "raw.jsonl"))
			803	+	if err != nil {
			804	+	t.Fatal(err)
			805	+	}
			806	+	if len(records) != result.Samples*2 {
			807	+	t.Fatalf("raw records=%d, want %d complete measured boundaries", len(records), result.Samples*2)
			808	+	}
			809	+	}
			810	+
679		func TestHarnessPreservesWorkloadAndCleanupErrors(t *testing.T) {	811		func TestHarnessPreservesWorkloadAndCleanupErrors(t *testing.T) {
680		dir := t.TempDir()	812		dir := t.TempDir()
681		raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE\|os.O_EXCL\|os.O_WRONLY, 0o600)	813		raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE\|os.O_EXCL\|os.O_WRONLY, 0o600)

@@ -895,7 +1027,7 @@ func TestHarnessWritesRequiredEvidenceWithSufficientSpans(t *testing.T) {
895		if err := json.Unmarshal(b, &got); err != nil {	1027		if err := json.Unmarshal(b, &got); err != nil {
896		t.Fatal(err)	1028		t.Fatal(err)
897		}	1029		}
898	-	if got.EndToEnd.Count != 10 \|\| got.Repetitions != 10 \|\| len(got.Spans) == 0 {	1030	+	if got.EndToEnd.Count < 10*minimumInIntervalEventSamples \|\| got.Repetitions != 10 \|\| got.RunP50Dispersion.Count != 10 \|\| got.Cadence.Count == 0 \|\| got.MaxGap == 0 \|\| len(got.Spans) == 0 {
899		t.Fatalf("summary=%+v", got)	1031		t.Fatalf("summary=%+v", got)
900		}	1032		}
901		for _, s := range got.Spans {	1033		for _, s := range got.Spans {