test(obs): validate canonical local evidence

commit: 42b3f98bd5ac93dff62932d526bccf02bb1efb5c
author: brice · 3mo ago · 2026-07-13 00:59 UTC
committer: brice · 3mo ago · 2026-07-13 01:11 UTC
parent: 0021186
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +198 -10

M cmd/vev-perf-harness/main.go +117 -3


@@ -5,6 +5,7 @@ package main
5	5		import (
6	6		"bufio"
7	7		"bytes"
	8	+	"context"
8	9		"encoding/json"
9	10		"errors"
10	11		"flag"

@@ -13,6 +14,7 @@ import (
13	14		"os"
14	15		"os/exec"
15	16		"path/filepath"
	17	+	"runtime/debug"
16	18		"sort"
17	19		"strconv"
18	20		"strings"

@@ -33,6 +35,7 @@ const (
33	35
34	36		type options struct {
35	37		vevBin, manifest, out string
	38	+	scenario string
36	39		warmup, duration time.Duration
37	40		repetitions int
38	41		}

@@ -247,7 +250,9 @@ type cliProcess struct {
247	250		waitTimeout func() <-chan time.Time
248	251		forceCleanup func()
249	252		forceKill func()
	253	+	shutdown func() error
250	254		terminalReady bool
	255	+	readyPath string
251	256		}
252	257
253	258		func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProcess, error) {

@@ -285,7 +290,7 @@ func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProces
285	290		if err := safedir.EnsurePrivate(filepath.Join(runDir, "state")); err != nil {
286	291		return nil, err
287	292		}
288		-	p := &cliProcess{cmd: cmd, waitErr: make(chan error, 1)}
	293	+	p := &cliProcess{cmd: cmd, waitErr: make(chan error, 1), readyPath: filepath.Join(runDir, "runtime", "vev", "daemon.sock")}
289	294		if m.Role == "client" {
290	295		master, slave, err := openPTY()
291	296		if err != nil {

@@ -324,6 +329,18 @@ func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProces
324	329		}
325	330		_ = devNull.Close()
326	331		}
	332	+	if m.Role == "daemon" {
	333	+	// Stop the foreground daemon through its public CLI before its bounded
	334	+	// process-group fallback. This lets blocked carriage operations close and
	335	+	// serialize their failed end marks instead of being cut off by SIGKILL.
	336	+	p.shutdown = func() error {
	337	+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	338	+	defer cancel()
	339	+	shutdown := exec.CommandContext(ctx, l.bin, "kill", "--daemon")
	340	+	shutdown.Env = withoutEnv(cmd.Env, "VEV_PERF_TRACE", "VEV_PERF_PROCESS_ID", "VEV_PERF_SCENARIO", "VEV_PERF_RUN")
	341	+	return shutdown.Run()
	342	+	}
	343	+	}
327	344		go func() { p.waitErr <- cmd.Wait() }()
328	345		return p, nil
329	346		}

@@ -482,6 +499,38 @@ func (p *cliProcess) copyTerminal() {
482	499		}
483	500		}
484	501
	502	+	// WaitReady confirms the explicitly launched public daemon owns its unique
	503	+	// socket before a client is launched. Without this barrier, the normal client
	504	+	// auto-spawn behavior can create an unmanifested daemon process during startup.
	505	+	func (p *cliProcess) WaitReady() error {
	506	+	if p.readyPath == "" {
	507	+	return errors.New("daemon readiness path is empty")
	508	+	}
	509	+	ticker := time.NewTicker(10 * time.Millisecond)
	510	+	defer ticker.Stop()
	511	+	timer := time.NewTimer(10 * time.Second)
	512	+	defer timer.Stop()
	513	+	for {
	514	+	if info, err := os.Stat(p.readyPath); err == nil {
	515	+	if info.Mode()&os.ModeSocket == 0 {
	516	+	return fmt.Errorf("daemon readiness path %s is not a socket", p.readyPath)
	517	+	}
	518	+	return nil
	519	+	}
	520	+	select {
	521	+	case err := <-p.waitErr:
	522	+	p.waitErr <- err
	523	+	if err == nil {
	524	+	return errors.New("daemon exited before readiness")
	525	+	}
	526	+	return fmt.Errorf("daemon exited before readiness: %w", err)
	527	+	case <-ticker.C:
	528	+	case <-timer.C:
	529	+	return fmt.Errorf("timed out waiting for daemon socket %s", p.readyPath)
	530	+	}
	531	+	}
	532	+	}
	533	+
485	534		func (p *cliProcess) Warmup(input []byte) error {
486	535		if p.pty == nil \|\| len(input) == 0 {
487	536		return nil

@@ -649,6 +698,17 @@ func (p *cliProcess) Close() error {
649	698		// before closing the PTY so its transport can finish its receive span;
650	699		// this avoids turning ordinary harness teardown into an aborted trace.
651	700		exited := false
	701	+	if p.shutdown != nil && p.waitErr != nil {
	702	+	select {
	703	+	case <-p.waitErr:
	704	+	exited = true
	705	+	default:
	706	+	// The client may have just ended the last session and raced this
	707	+	// best-effort public shutdown. The normal bounded reaping path below
	708	+	// remains authoritative in that case.
	709	+	_ = p.shutdown()
	710	+	}
	711	+	}
652	712		if p.pty != nil && p.waitErr != nil {
653	713		_, _ = p.pty.Write([]byte("exit\n"))
654	714		select {

@@ -692,6 +752,12 @@ func (p *cliProcess) closeDeadline() <-chan time.Time {
692	752		if p.waitTimeout != nil {
693	753		return p.waitTimeout()
694	754		}
	755	+	if p.shutdown != nil {
	756	+	// A public kill asks the daemon to finish bounded detached notices and
	757	+	// close blocked carriage before it exits; do not turn that graceful
	758	+	// protocol into a SIGKILL at the two-second client deadline.
	759	+	return time.After(5 * time.Second)
	760	+	}
695	761		return time.After(2 * time.Second)
696	762		}
697	763

@@ -768,7 +834,25 @@ func main() {
768	834		}
769	835		}
770	836		func defaultHarness() *harness {
771		-	return &harness{clock: systemClock{time.Now()}, mkdir: safedir.EnsurePrivate, create: func(p string) (*os.File, error) { return os.OpenFile(p, os.O_WRONLY\|os.O_CREATE\|os.O_EXCL, 0o600) }, removeAll: os.RemoveAll, gitSHA: func() string { return "unknown" }}
	837	+	return &harness{clock: systemClock{time.Now()}, mkdir: safedir.EnsurePrivate, create: func(p string) (*os.File, error) { return os.OpenFile(p, os.O_WRONLY\|os.O_CREATE\|os.O_EXCL, 0o600) }, removeAll: os.RemoveAll, gitSHA: recordedGitSHA}
	838	+	}
	839	+
	840	+	// recordedGitSHA prefers build provenance so an installed public harness does
	841	+	// not depend on its launch directory being a checkout. The git fallback keeps
	842	+	// go run evidence attributable as well.
	843	+	func recordedGitSHA() string {
	844	+	if info, ok := debug.ReadBuildInfo(); ok {
	845	+	for _, setting := range info.Settings {
	846	+	if setting.Key == "vcs.revision" && setting.Value != "" {
	847	+	return setting.Value
	848	+	}
	849	+	}
	850	+	}
	851	+	out, err := exec.Command("git", "rev-parse", "HEAD").Output()
	852	+	if err == nil && strings.TrimSpace(string(out)) != "" {
	853	+	return strings.TrimSpace(string(out))
	854	+	}
	855	+	return "unknown"
772	856		}
773	857		func run(args []string, h *harness) error {
774	858		opt, err := parseOptions(args)

@@ -793,6 +877,12 @@ func run(args []string, h *harness) error {
793	877		results := make([]runResult, 0, len(m.Scenarios)*opt.repetitions)
794	878		var localSpans []span
795	879		scenarios := m.Scenarios
	880	+	if opt.scenario != "" {
	881	+	scenarios, err = selectScenario(m, opt.scenario)
	882	+	if err != nil {
	883	+	return err
	884	+	}
	885	+	}
796	886		if h.selectScenarios != nil {
797	887		scenarios = h.selectScenarios(m)
798	888		}

@@ -840,6 +930,7 @@ func parseOptions(args []string) (options, error) {
840	930		fs.StringVar(&o.vevBin, "vev-bin", "", "public vev binary")
841	931		fs.StringVar(&o.manifest, "manifest", "", "manifest")
842	932		fs.StringVar(&o.out, "out", "", "output directory")
	933	+	fs.StringVar(&o.scenario, "scenario", "", "one canonical scenario ID (after full manifest validation)")
843	934		fs.DurationVar(&o.warmup, "warmup", 0, "warmup")
844	935		fs.DurationVar(&o.duration, "duration", 0, "measurement duration")
845	936		fs.IntVar(&o.repetitions, "repetitions", 0, "repetitions")

@@ -948,6 +1039,22 @@ func validateManifest(m manifest) error {
948	1039		}
949	1040		return nil
950	1041		}
	1042	+
	1043	+	// selectScenario is intentionally called only after validateManifest. A bounded
	1044	+	// local evidence run therefore cannot mask an incomplete canonical matrix.
	1045	+	func selectScenario(m manifest, id string) ([]scenario, error) {
	1046	+	for _, s := range m.Scenarios {
	1047	+	if s.ID != id {
	1048	+	continue
	1049	+	}
	1050	+	if s.InapplicableReason != "" {
	1051	+	return nil, fmt.Errorf("scenario %s is inapplicable: %s", id, s.InapplicableReason)
	1052	+	}
	1053	+	return []scenario{s}, nil
	1054	+	}
	1055	+	return nil, fmt.Errorf("unknown canonical scenario %s", id)
	1056	+	}
	1057	+
951	1058		func requiredRoles(transportID string) []string {
952	1059		switch transportID {
953	1060		case "local":

@@ -1063,6 +1170,13 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr
1063	1170		return res, e
1064	1171		}
1065	1172		processes = append(processes, launchedRole{mapping: pm, process: p})
	1173	+	if pm.Role == "daemon" {
	1174	+	if ready, ok := p.(interface{ WaitReady() error }); ok {
	1175	+	if err = ready.WaitReady(); err != nil {
	1176	+	return res, err
	1177	+	}
	1178	+	}
	1179	+	}
1066	1180		}
1067	1181		for _, p := range processes {
1068	1182		if p.mapping.Role != "client" {

@@ -1382,7 +1496,7 @@ type traceRecord struct {
1382	1496		}
1383	1497		type spanPair struct{ start, end, name string }
1384	1498
1385		-	var spanPairs = []spanPair{{"diff_start", "diff_end", "diff_duration"}, {"queue_enqueued", "queue_dequeued", "queue_wait"}, {"ack_blocked_start", "ack_blocked_end", "ack_blocked_interval"}, {"adapter_send_start", "adapter_send_end", "adapter_send_duration"}, {"adapter_receive_start", "adapter_receive_end", "adapter_receive_duration"}}
	1499	+	var spanPairs = []spanPair{{"capture_start", "capture_end", "capture_duration"}, {"compose_start", "compose_end", "compose_duration"}, {"diff_start", "diff_end", "diff_duration"}, {"queue_enqueued", "queue_dequeued", "queue_wait"}, {"ack_blocked_start", "ack_blocked_end", "ack_blocked_interval"}, {"emit_start", "emit_end", "emit_duration"}, {"adapter_send_start", "adapter_send_end", "adapter_send_duration"}, {"adapter_receive_start", "adapter_receive_end", "adapter_receive_duration"}}
1386	1500
1387	1501		// mergeProcessTraces permits only records from one manifest process in a span.
1388	1502		// In particular, the correlation key includes process_id before ticks are read.


@@ -5,6 +5,7 @@ package main
5		import (	5		import (
6		"bufio"	6		"bufio"
7		"bytes"	7		"bytes"
			8	+	"context"
8		"encoding/json"	9		"encoding/json"
9		"errors"	10		"errors"
10		"flag"	11		"flag"

@@ -13,6 +14,7 @@ import (
13		"os"	14		"os"
14		"os/exec"	15		"os/exec"
15		"path/filepath"	16		"path/filepath"
			17	+	"runtime/debug"
16		"sort"	18		"sort"
17		"strconv"	19		"strconv"
18		"strings"	20		"strings"

@@ -33,6 +35,7 @@ const (
33			35
34		type options struct {	36		type options struct {
35		vevBin, manifest, out string	37		vevBin, manifest, out string
			38	+	scenario string
36		warmup, duration time.Duration	39		warmup, duration time.Duration
37		repetitions int	40		repetitions int
38		}	41		}

@@ -247,7 +250,9 @@ type cliProcess struct {
247		waitTimeout func() <-chan time.Time	250		waitTimeout func() <-chan time.Time
248		forceCleanup func()	251		forceCleanup func()
249		forceKill func()	252		forceKill func()
			253	+	shutdown func() error
250		terminalReady bool	254		terminalReady bool
			255	+	readyPath string
251		}	256		}
252			257
253		func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProcess, error) {	258		func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProcess, error) {

@@ -285,7 +290,7 @@ func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProces
285		if err := safedir.EnsurePrivate(filepath.Join(runDir, "state")); err != nil {	290		if err := safedir.EnsurePrivate(filepath.Join(runDir, "state")); err != nil {
286		return nil, err	291		return nil, err
287		}	292		}
288	-	p := &cliProcess{cmd: cmd, waitErr: make(chan error, 1)}	293	+	p := &cliProcess{cmd: cmd, waitErr: make(chan error, 1), readyPath: filepath.Join(runDir, "runtime", "vev", "daemon.sock")}
289		if m.Role == "client" {	294		if m.Role == "client" {
290		master, slave, err := openPTY()	295		master, slave, err := openPTY()
291		if err != nil {	296		if err != nil {

@@ -324,6 +329,18 @@ func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProces
324		}	329		}
325		_ = devNull.Close()	330		_ = devNull.Close()
326		}	331		}
			332	+	if m.Role == "daemon" {
			333	+	// Stop the foreground daemon through its public CLI before its bounded
			334	+	// process-group fallback. This lets blocked carriage operations close and
			335	+	// serialize their failed end marks instead of being cut off by SIGKILL.
			336	+	p.shutdown = func() error {
			337	+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
			338	+	defer cancel()
			339	+	shutdown := exec.CommandContext(ctx, l.bin, "kill", "--daemon")
			340	+	shutdown.Env = withoutEnv(cmd.Env, "VEV_PERF_TRACE", "VEV_PERF_PROCESS_ID", "VEV_PERF_SCENARIO", "VEV_PERF_RUN")
			341	+	return shutdown.Run()
			342	+	}
			343	+	}
327		go func() { p.waitErr <- cmd.Wait() }()	344		go func() { p.waitErr <- cmd.Wait() }()
328		return p, nil	345		return p, nil
329		}	346		}

@@ -482,6 +499,38 @@ func (p *cliProcess) copyTerminal() {
482		}	499		}
483		}	500		}
484			501
			502	+	// WaitReady confirms the explicitly launched public daemon owns its unique
			503	+	// socket before a client is launched. Without this barrier, the normal client
			504	+	// auto-spawn behavior can create an unmanifested daemon process during startup.
			505	+	func (p *cliProcess) WaitReady() error {
			506	+	if p.readyPath == "" {
			507	+	return errors.New("daemon readiness path is empty")
			508	+	}
			509	+	ticker := time.NewTicker(10 * time.Millisecond)
			510	+	defer ticker.Stop()
			511	+	timer := time.NewTimer(10 * time.Second)
			512	+	defer timer.Stop()
			513	+	for {
			514	+	if info, err := os.Stat(p.readyPath); err == nil {
			515	+	if info.Mode()&os.ModeSocket == 0 {
			516	+	return fmt.Errorf("daemon readiness path %s is not a socket", p.readyPath)
			517	+	}
			518	+	return nil
			519	+	}
			520	+	select {
			521	+	case err := <-p.waitErr:
			522	+	p.waitErr <- err
			523	+	if err == nil {
			524	+	return errors.New("daemon exited before readiness")
			525	+	}
			526	+	return fmt.Errorf("daemon exited before readiness: %w", err)
			527	+	case <-ticker.C:
			528	+	case <-timer.C:
			529	+	return fmt.Errorf("timed out waiting for daemon socket %s", p.readyPath)
			530	+	}
			531	+	}
			532	+	}
			533	+
485		func (p *cliProcess) Warmup(input []byte) error {	534		func (p *cliProcess) Warmup(input []byte) error {
486		if p.pty == nil \|\| len(input) == 0 {	535		if p.pty == nil \|\| len(input) == 0 {
487		return nil	536		return nil

@@ -649,6 +698,17 @@ func (p *cliProcess) Close() error {
649		// before closing the PTY so its transport can finish its receive span;	698		// before closing the PTY so its transport can finish its receive span;
650		// this avoids turning ordinary harness teardown into an aborted trace.	699		// this avoids turning ordinary harness teardown into an aborted trace.
651		exited := false	700		exited := false
			701	+	if p.shutdown != nil && p.waitErr != nil {
			702	+	select {
			703	+	case <-p.waitErr:
			704	+	exited = true
			705	+	default:
			706	+	// The client may have just ended the last session and raced this
			707	+	// best-effort public shutdown. The normal bounded reaping path below
			708	+	// remains authoritative in that case.
			709	+	_ = p.shutdown()
			710	+	}
			711	+	}
652		if p.pty != nil && p.waitErr != nil {	712		if p.pty != nil && p.waitErr != nil {
653		_, _ = p.pty.Write([]byte("exit\n"))	713		_, _ = p.pty.Write([]byte("exit\n"))
654		select {	714		select {

@@ -692,6 +752,12 @@ func (p *cliProcess) closeDeadline() <-chan time.Time {
692		if p.waitTimeout != nil {	752		if p.waitTimeout != nil {
693		return p.waitTimeout()	753		return p.waitTimeout()
694		}	754		}
			755	+	if p.shutdown != nil {
			756	+	// A public kill asks the daemon to finish bounded detached notices and
			757	+	// close blocked carriage before it exits; do not turn that graceful
			758	+	// protocol into a SIGKILL at the two-second client deadline.
			759	+	return time.After(5 * time.Second)
			760	+	}
695		return time.After(2 * time.Second)	761		return time.After(2 * time.Second)
696		}	762		}
697			763

@@ -768,7 +834,25 @@ func main() {
768		}	834		}
769		}	835		}
770		func defaultHarness() *harness {	836		func defaultHarness() *harness {
771	-	return &harness{clock: systemClock{time.Now()}, mkdir: safedir.EnsurePrivate, create: func(p string) (*os.File, error) { return os.OpenFile(p, os.O_WRONLY\|os.O_CREATE\|os.O_EXCL, 0o600) }, removeAll: os.RemoveAll, gitSHA: func() string { return "unknown" }}	837	+	return &harness{clock: systemClock{time.Now()}, mkdir: safedir.EnsurePrivate, create: func(p string) (*os.File, error) { return os.OpenFile(p, os.O_WRONLY\|os.O_CREATE\|os.O_EXCL, 0o600) }, removeAll: os.RemoveAll, gitSHA: recordedGitSHA}
			838	+	}
			839	+
			840	+	// recordedGitSHA prefers build provenance so an installed public harness does
			841	+	// not depend on its launch directory being a checkout. The git fallback keeps
			842	+	// go run evidence attributable as well.
			843	+	func recordedGitSHA() string {
			844	+	if info, ok := debug.ReadBuildInfo(); ok {
			845	+	for _, setting := range info.Settings {
			846	+	if setting.Key == "vcs.revision" && setting.Value != "" {
			847	+	return setting.Value
			848	+	}
			849	+	}
			850	+	}
			851	+	out, err := exec.Command("git", "rev-parse", "HEAD").Output()
			852	+	if err == nil && strings.TrimSpace(string(out)) != "" {
			853	+	return strings.TrimSpace(string(out))
			854	+	}
			855	+	return "unknown"
772		}	856		}
773		func run(args []string, h *harness) error {	857		func run(args []string, h *harness) error {
774		opt, err := parseOptions(args)	858		opt, err := parseOptions(args)

@@ -793,6 +877,12 @@ func run(args []string, h *harness) error {
793		results := make([]runResult, 0, len(m.Scenarios)*opt.repetitions)	877		results := make([]runResult, 0, len(m.Scenarios)*opt.repetitions)
794		var localSpans []span	878		var localSpans []span
795		scenarios := m.Scenarios	879		scenarios := m.Scenarios
			880	+	if opt.scenario != "" {
			881	+	scenarios, err = selectScenario(m, opt.scenario)
			882	+	if err != nil {
			883	+	return err
			884	+	}
			885	+	}
796		if h.selectScenarios != nil {	886		if h.selectScenarios != nil {
797		scenarios = h.selectScenarios(m)	887		scenarios = h.selectScenarios(m)
798		}	888		}

@@ -840,6 +930,7 @@ func parseOptions(args []string) (options, error) {
840		fs.StringVar(&o.vevBin, "vev-bin", "", "public vev binary")	930		fs.StringVar(&o.vevBin, "vev-bin", "", "public vev binary")
841		fs.StringVar(&o.manifest, "manifest", "", "manifest")	931		fs.StringVar(&o.manifest, "manifest", "", "manifest")
842		fs.StringVar(&o.out, "out", "", "output directory")	932		fs.StringVar(&o.out, "out", "", "output directory")
			933	+	fs.StringVar(&o.scenario, "scenario", "", "one canonical scenario ID (after full manifest validation)")
843		fs.DurationVar(&o.warmup, "warmup", 0, "warmup")	934		fs.DurationVar(&o.warmup, "warmup", 0, "warmup")
844		fs.DurationVar(&o.duration, "duration", 0, "measurement duration")	935		fs.DurationVar(&o.duration, "duration", 0, "measurement duration")
845		fs.IntVar(&o.repetitions, "repetitions", 0, "repetitions")	936		fs.IntVar(&o.repetitions, "repetitions", 0, "repetitions")

@@ -948,6 +1039,22 @@ func validateManifest(m manifest) error {
948		}	1039		}
949		return nil	1040		return nil
950		}	1041		}
			1042	+
			1043	+	// selectScenario is intentionally called only after validateManifest. A bounded
			1044	+	// local evidence run therefore cannot mask an incomplete canonical matrix.
			1045	+	func selectScenario(m manifest, id string) ([]scenario, error) {
			1046	+	for _, s := range m.Scenarios {
			1047	+	if s.ID != id {
			1048	+	continue
			1049	+	}
			1050	+	if s.InapplicableReason != "" {
			1051	+	return nil, fmt.Errorf("scenario %s is inapplicable: %s", id, s.InapplicableReason)
			1052	+	}
			1053	+	return []scenario{s}, nil
			1054	+	}
			1055	+	return nil, fmt.Errorf("unknown canonical scenario %s", id)
			1056	+	}
			1057	+
951		func requiredRoles(transportID string) []string {	1058		func requiredRoles(transportID string) []string {
952		switch transportID {	1059		switch transportID {
953		case "local":	1060		case "local":

@@ -1063,6 +1170,13 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr
1063		return res, e	1170		return res, e
1064		}	1171		}
1065		processes = append(processes, launchedRole{mapping: pm, process: p})	1172		processes = append(processes, launchedRole{mapping: pm, process: p})
			1173	+	if pm.Role == "daemon" {
			1174	+	if ready, ok := p.(interface{ WaitReady() error }); ok {
			1175	+	if err = ready.WaitReady(); err != nil {
			1176	+	return res, err
			1177	+	}
			1178	+	}
			1179	+	}
1066		}	1180		}
1067		for _, p := range processes {	1181		for _, p := range processes {
1068		if p.mapping.Role != "client" {	1182		if p.mapping.Role != "client" {

@@ -1382,7 +1496,7 @@ type traceRecord struct {
1382		}	1496		}
1383		type spanPair struct{ start, end, name string }	1497		type spanPair struct{ start, end, name string }
1384			1498
1385	-	var spanPairs = []spanPair{{"diff_start", "diff_end", "diff_duration"}, {"queue_enqueued", "queue_dequeued", "queue_wait"}, {"ack_blocked_start", "ack_blocked_end", "ack_blocked_interval"}, {"adapter_send_start", "adapter_send_end", "adapter_send_duration"}, {"adapter_receive_start", "adapter_receive_end", "adapter_receive_duration"}}	1499	+	var spanPairs = []spanPair{{"capture_start", "capture_end", "capture_duration"}, {"compose_start", "compose_end", "compose_duration"}, {"diff_start", "diff_end", "diff_duration"}, {"queue_enqueued", "queue_dequeued", "queue_wait"}, {"ack_blocked_start", "ack_blocked_end", "ack_blocked_interval"}, {"emit_start", "emit_end", "emit_duration"}, {"adapter_send_start", "adapter_send_end", "adapter_send_duration"}, {"adapter_receive_start", "adapter_receive_end", "adapter_receive_duration"}}
1386			1500
1387		// mergeProcessTraces permits only records from one manifest process in a span.	1501		// mergeProcessTraces permits only records from one manifest process in a span.
1388		// In particular, the correlation key includes process_id before ticks are read.	1502		// In particular, the correlation key includes process_id before ticks are read.

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


@@ -117,6 +117,40 @@ func TestCLIProcessCloseIsBoundedWhenPublicRoleDoesNotExit(t *testing.T) {
117	117		}
118	118		}
119	119
	120	+	func TestCLIProcessWaitReadyRequiresDaemonSocket(t *testing.T) {
	121	+	path := filepath.Join(t.TempDir(), "daemon.sock")
	122	+	listener, err := net.Listen("unix", path)
	123	+	if err != nil {
	124	+	t.Fatal(err)
	125	+	}
	126	+	defer listener.Close()
	127	+	if err := (&cliProcess{readyPath: path, waitErr: make(chan error, 1)}).WaitReady(); err != nil {
	128	+	t.Fatal(err)
	129	+	}
	130	+	}
	131	+
	132	+	func TestCLIProcessCloseUsesPublicDaemonShutdown(t *testing.T) {
	133	+	wait := make(chan error, 1)
	134	+	wait <- nil
	135	+	var shutdown int
	136	+	p := &cliProcess{waitErr: wait, shutdown: func() error { shutdown++; return nil }}
	137	+	if err := p.Close(); err != nil {
	138	+	t.Fatal(err)
	139	+	}
	140	+	if shutdown != 0 {
	141	+	t.Fatalf("shutdown calls = %d, want 0 for an exited daemon", shutdown)
	142	+	}
	143	+
	144	+	wait = make(chan error, 1)
	145	+	p = &cliProcess{waitErr: wait, shutdown: func() error { shutdown++; wait <- nil; return nil }}
	146	+	if err := p.Close(); err != nil {
	147	+	t.Fatal(err)
	148	+	}
	149	+	if shutdown != 1 {
	150	+	t.Fatalf("shutdown calls = %d, want 1", shutdown)
	151	+	}
	152	+	}
	153	+
120	154		func TestHarnessCanonicalLocalSmokeRealRoles(t *testing.T) {
121	155		if testing.Short() {
122	156		t.Skip("public CLI smoke")

@@ -166,7 +200,7 @@ func TestHarnessCanonicalLocalSmokeRealRoles(t *testing.T) {
166	200		}
167	201		}
168	202
169		-	func TestHarnessTestSelectionKeepsCanonicalManifestValidation(t *testing.T) {
	203	+	func TestHarnessScenarioSelectionKeepsCanonicalManifestValidation(t *testing.T) {
170	204		root, err := filepath.Abs(filepath.Join("..", ".."))
171	205		if err != nil {
172	206		t.Fatal(err)

@@ -176,15 +210,11 @@ func TestHarnessTestSelectionKeepsCanonicalManifestValidation(t *testing.T) {
176	210		if err != nil {
177	211		t.Fatal(err)
178	212		}
179		-	if err := validateManifest(m); err != nil {
180		-	t.Fatal(err)
181		-	}
182	213		selected := m.Scenarios[0]
183	214		h := defaultHarness()
184	215		h.clock, h.launcher = &fakeClock{}, &fakeLauncher{}
185		-	h.selectScenarios = func(manifest) []scenario { return []scenario{selected} }
186	216		out := filepath.Join(t.TempDir(), "out")
187		-	if err := run([]string{"--vev-bin", "ignored", "--manifest", manifestPath, "--out", out, "--warmup", "0s", "--duration", "30s", "--repetitions", "10"}, h); err != nil {
	217	+	if err := run([]string{"--vev-bin", "ignored", "--manifest", manifestPath, "--out", out, "--scenario", selected.ID, "--warmup", "0s", "--duration", "30s", "--repetitions", "10"}, h); err != nil {
188	218		t.Fatal(err)
189	219		}
190	220		var runs []runResult

@@ -198,6 +228,39 @@ func TestHarnessTestSelectionKeepsCanonicalManifestValidation(t *testing.T) {
198	228		if len(runs) != 10 {
199	229		t.Fatalf("selected canonical runs=%d, want 10", len(runs))
200	230		}
	231	+	for _, result := range runs {
	232	+	if result.Scenario != selected.ID {
	233	+	t.Fatalf("selected scenario = %q, want %q", result.Scenario, selected.ID)
	234	+	}
	235	+	}
	236	+	}
	237	+
	238	+	func TestHarnessScenarioSelectionRejectsIncompleteCanonicalManifest(t *testing.T) {
	239	+	m, err := readManifest(filepath.Join("..", "..", "testdata", "perf", "manifest.json"))
	240	+	if err != nil {
	241	+	t.Fatal(err)
	242	+	}
	243	+	selected := m.Scenarios[0]
	244	+	m.Scenarios = m.Scenarios[:len(m.Scenarios)-1]
	245	+	dir := t.TempDir()
	246	+	manifestPath := filepath.Join(dir, "manifest.json")
	247	+	if err := os.WriteFile(manifestPath, []byte(mustJSON(m)), 0o600); err != nil {
	248	+	t.Fatal(err)
	249	+	}
	250	+	launcher := &fakeLauncher{}
	251	+	h := defaultHarness()
	252	+	h.clock, h.launcher = &fakeClock{}, launcher
	253	+	out := filepath.Join(dir, "out")
	254	+	err = run([]string{"--vev-bin", "ignored", "--manifest", manifestPath, "--out", out, "--scenario", selected.ID, "--warmup", "0s", "--duration", "30s", "--repetitions", "10"}, h)
	255	+	if err == nil \|\| !strings.Contains(err.Error(), "missing scenario or inapplicable reason") {
	256	+	t.Fatalf("incomplete manifest selection error = %v", err)
	257	+	}
	258	+	if len(launcher.mappings) != 0 {
	259	+	t.Fatalf("launched with incomplete canonical manifest: %+v", launcher.mappings)
	260	+	}
	261	+	if _, statErr := os.Stat(out); !errors.Is(statErr, os.ErrNotExist) {
	262	+	t.Fatalf("output created before manifest validation: %v", statErr)
	263	+	}
201	264		}
202	265
203	266		func TestTraceEnvironmentMatchesProcessMapping(t *testing.T) {

@@ -883,6 +946,17 @@ func equalStrings(a, b []string) bool {
883	946		return true
884	947		}
885	948
	949	+	func TestHarnessIncludesAllRequiredProcessLocalSpans(t *testing.T) {
	950	+	got := make([]string, 0, len(spanPairs))
	951	+	for _, pair := range spanPairs {
	952	+	got = append(got, pair.name)
	953	+	}
	954	+	want := []string{"capture_duration", "compose_duration", "diff_duration", "queue_wait", "ack_blocked_interval", "emit_duration", "adapter_send_duration", "adapter_receive_duration"}
	955	+	if !equalStrings(got, want) {
	956	+	t.Fatalf("span pairs = %v, want %v", got, want)
	957	+	}
	958	+	}
	959	+
886	960		func TestHarnessRawRecordsAreDeterministicJSONL(t *testing.T) {
887	961		var b bytes.Buffer
888	962		for _, m := range []harnessMark{{Scenario: "s", Run: 1, Sequence: 1, Kind: "input_injected", Tick: 1, Valid: true}, {Scenario: "s", Run: 1, Sequence: 1, Kind: "terminal_flushed", Tick: 2, Valid: true}} {


@@ -117,6 +117,40 @@ func TestCLIProcessCloseIsBoundedWhenPublicRoleDoesNotExit(t *testing.T) {
117		}	117		}
118		}	118		}
119			119
			120	+	func TestCLIProcessWaitReadyRequiresDaemonSocket(t *testing.T) {
			121	+	path := filepath.Join(t.TempDir(), "daemon.sock")
			122	+	listener, err := net.Listen("unix", path)
			123	+	if err != nil {
			124	+	t.Fatal(err)
			125	+	}
			126	+	defer listener.Close()
			127	+	if err := (&cliProcess{readyPath: path, waitErr: make(chan error, 1)}).WaitReady(); err != nil {
			128	+	t.Fatal(err)
			129	+	}
			130	+	}
			131	+
			132	+	func TestCLIProcessCloseUsesPublicDaemonShutdown(t *testing.T) {
			133	+	wait := make(chan error, 1)
			134	+	wait <- nil
			135	+	var shutdown int
			136	+	p := &cliProcess{waitErr: wait, shutdown: func() error { shutdown++; return nil }}
			137	+	if err := p.Close(); err != nil {
			138	+	t.Fatal(err)
			139	+	}
			140	+	if shutdown != 0 {
			141	+	t.Fatalf("shutdown calls = %d, want 0 for an exited daemon", shutdown)
			142	+	}
			143	+
			144	+	wait = make(chan error, 1)
			145	+	p = &cliProcess{waitErr: wait, shutdown: func() error { shutdown++; wait <- nil; return nil }}
			146	+	if err := p.Close(); err != nil {
			147	+	t.Fatal(err)
			148	+	}
			149	+	if shutdown != 1 {
			150	+	t.Fatalf("shutdown calls = %d, want 1", shutdown)
			151	+	}
			152	+	}
			153	+
120		func TestHarnessCanonicalLocalSmokeRealRoles(t *testing.T) {	154		func TestHarnessCanonicalLocalSmokeRealRoles(t *testing.T) {
121		if testing.Short() {	155		if testing.Short() {
122		t.Skip("public CLI smoke")	156		t.Skip("public CLI smoke")

@@ -166,7 +200,7 @@ func TestHarnessCanonicalLocalSmokeRealRoles(t *testing.T) {
166		}	200		}
167		}	201		}
168			202
169	-	func TestHarnessTestSelectionKeepsCanonicalManifestValidation(t *testing.T) {	203	+	func TestHarnessScenarioSelectionKeepsCanonicalManifestValidation(t *testing.T) {
170		root, err := filepath.Abs(filepath.Join("..", ".."))	204		root, err := filepath.Abs(filepath.Join("..", ".."))
171		if err != nil {	205		if err != nil {
172		t.Fatal(err)	206		t.Fatal(err)

@@ -176,15 +210,11 @@ func TestHarnessTestSelectionKeepsCanonicalManifestValidation(t *testing.T) {
176		if err != nil {	210		if err != nil {
177		t.Fatal(err)	211		t.Fatal(err)
178		}	212		}
179	-	if err := validateManifest(m); err != nil {
180	-	t.Fatal(err)
181	-	}
182		selected := m.Scenarios[0]	213		selected := m.Scenarios[0]
183		h := defaultHarness()	214		h := defaultHarness()
184		h.clock, h.launcher = &fakeClock{}, &fakeLauncher{}	215		h.clock, h.launcher = &fakeClock{}, &fakeLauncher{}
185	-	h.selectScenarios = func(manifest) []scenario { return []scenario{selected} }
186		out := filepath.Join(t.TempDir(), "out")	216		out := filepath.Join(t.TempDir(), "out")
187	-	if err := run([]string{"--vev-bin", "ignored", "--manifest", manifestPath, "--out", out, "--warmup", "0s", "--duration", "30s", "--repetitions", "10"}, h); err != nil {	217	+	if err := run([]string{"--vev-bin", "ignored", "--manifest", manifestPath, "--out", out, "--scenario", selected.ID, "--warmup", "0s", "--duration", "30s", "--repetitions", "10"}, h); err != nil {
188		t.Fatal(err)	218		t.Fatal(err)
189		}	219		}
190		var runs []runResult	220		var runs []runResult

@@ -198,6 +228,39 @@ func TestHarnessTestSelectionKeepsCanonicalManifestValidation(t *testing.T) {
198		if len(runs) != 10 {	228		if len(runs) != 10 {
199		t.Fatalf("selected canonical runs=%d, want 10", len(runs))	229		t.Fatalf("selected canonical runs=%d, want 10", len(runs))
200		}	230		}
			231	+	for _, result := range runs {
			232	+	if result.Scenario != selected.ID {
			233	+	t.Fatalf("selected scenario = %q, want %q", result.Scenario, selected.ID)
			234	+	}
			235	+	}
			236	+	}
			237	+
			238	+	func TestHarnessScenarioSelectionRejectsIncompleteCanonicalManifest(t *testing.T) {
			239	+	m, err := readManifest(filepath.Join("..", "..", "testdata", "perf", "manifest.json"))
			240	+	if err != nil {
			241	+	t.Fatal(err)
			242	+	}
			243	+	selected := m.Scenarios[0]
			244	+	m.Scenarios = m.Scenarios[:len(m.Scenarios)-1]
			245	+	dir := t.TempDir()
			246	+	manifestPath := filepath.Join(dir, "manifest.json")
			247	+	if err := os.WriteFile(manifestPath, []byte(mustJSON(m)), 0o600); err != nil {
			248	+	t.Fatal(err)
			249	+	}
			250	+	launcher := &fakeLauncher{}
			251	+	h := defaultHarness()
			252	+	h.clock, h.launcher = &fakeClock{}, launcher
			253	+	out := filepath.Join(dir, "out")
			254	+	err = run([]string{"--vev-bin", "ignored", "--manifest", manifestPath, "--out", out, "--scenario", selected.ID, "--warmup", "0s", "--duration", "30s", "--repetitions", "10"}, h)
			255	+	if err == nil \|\| !strings.Contains(err.Error(), "missing scenario or inapplicable reason") {
			256	+	t.Fatalf("incomplete manifest selection error = %v", err)
			257	+	}
			258	+	if len(launcher.mappings) != 0 {
			259	+	t.Fatalf("launched with incomplete canonical manifest: %+v", launcher.mappings)
			260	+	}
			261	+	if _, statErr := os.Stat(out); !errors.Is(statErr, os.ErrNotExist) {
			262	+	t.Fatalf("output created before manifest validation: %v", statErr)
			263	+	}
201		}	264		}
202			265
203		func TestTraceEnvironmentMatchesProcessMapping(t *testing.T) {	266		func TestTraceEnvironmentMatchesProcessMapping(t *testing.T) {

@@ -883,6 +946,17 @@ func equalStrings(a, b []string) bool {
883		return true	946		return true
884		}	947		}
885			948
			949	+	func TestHarnessIncludesAllRequiredProcessLocalSpans(t *testing.T) {
			950	+	got := make([]string, 0, len(spanPairs))
			951	+	for _, pair := range spanPairs {
			952	+	got = append(got, pair.name)
			953	+	}
			954	+	want := []string{"capture_duration", "compose_duration", "diff_duration", "queue_wait", "ack_blocked_interval", "emit_duration", "adapter_send_duration", "adapter_receive_duration"}
			955	+	if !equalStrings(got, want) {
			956	+	t.Fatalf("span pairs = %v, want %v", got, want)
			957	+	}
			958	+	}
			959	+
886		func TestHarnessRawRecordsAreDeterministicJSONL(t *testing.T) {	960		func TestHarnessRawRecordsAreDeterministicJSONL(t *testing.T) {
887		var b bytes.Buffer	961		var b bytes.Buffer
888		for _, m := range []harnessMark{{Scenario: "s", Run: 1, Sequence: 1, Kind: "input_injected", Tick: 1, Valid: true}, {Scenario: "s", Run: 1, Sequence: 1, Kind: "terminal_flushed", Tick: 2, Valid: true}} {	962		for _, m := range []harnessMark{{Scenario: "s", Run: 1, Sequence: 1, Kind: "input_injected", Tick: 1, Valid: true}, {Scenario: "s", Run: 1, Sequence: 1, Kind: "terminal_flushed", Tick: 2, Valid: true}} {

M testdata/perf/README.md +1 -1


@@ -66,4 +66,4 @@ jq -e '(.topologies\|length)==4 and (.workloads\|length)==9 and (.transports\|lengt
66	66		go test ./cmd/vev-perf-harness
67	67		```
68	68
69		-	The full command above is the measurement procedure; its 10-second warmup, 30-second measured interval, and 10 repetitions are the minimum accepted settings.
	69	+	The full command above is the measurement procedure; its 10-second warmup, 30-second measured interval, and 10 repetitions are the minimum accepted settings. For a bounded local evidence run, add `--scenario 1x4-idle-local`; the harness validates the complete canonical manifest before it applies that selection, so a single-scenario run cannot hide missing matrix coverage.


@@ -66,4 +66,4 @@ jq -e '(.topologies\|length)==4 and (.workloads\|length)==9 and (.transports\|lengt
66		go test ./cmd/vev-perf-harness	66		go test ./cmd/vev-perf-harness
67		```	67		```
68			68
69	-	The full command above is the measurement procedure; its 10-second warmup, 30-second measured interval, and 10 repetitions are the minimum accepted settings.	69	+	The full command above is the measurement procedure; its 10-second warmup, 30-second measured interval, and 10 repetitions are the minimum accepted settings. For a bounded local evidence run, add `--scenario 1x4-idle-local`; the harness validates the complete canonical manifest before it applies that selection, so a single-scenario run cannot hide missing matrix coverage.