fix(obs): propagate harness correlation

commit: dd4fb64259d2ca15d4b639c9a6002e47b8f35e8a
author: brice · 3mo ago · 2026-07-13 00:12 UTC
parent: 3d59ccb
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +399 -46

M cmd/vev-perf-harness/main.go +96 -24


@@ -158,7 +158,7 @@ func closeLaunchedRoles(processes []launchedRole) error {
158	158		var failures []error
159	159		for i := len(processes) - 1; i >= 0; i-- {
160	160		if err := processes[i].process.Close(); err != nil {
161		-	failures = append(failures, err)
	161	+	failures = append(failures, fmt.Errorf("close %s role: %w", processes[i].mapping.Role, err))
162	162		}
163	163		}
164	164		return errors.Join(failures...)

@@ -223,6 +223,7 @@ type cliProcess struct {
223	223		waitErr chan error
224	224		waitTimeout func() <-chan time.Time
225	225		forceCleanup func()
	226	+	forceKill func()
226	227		terminalReady bool
227	228		}
228	229

@@ -240,8 +241,8 @@ func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProces
240	241		}
241	242		cmd := exec.Command(l.bin, role.Args...)
242	243		runDir := filepath.Dir(m.TracePath)
243		-	cmd.Env = append(withoutEnv(os.Environ(), "VEV", "VEV_PERF_TRACE", "VEV_PERF_PROCESS_ID", "VEV_REMOTE_TRANSPORT"), "VEV_PERF_TRACE="+m.TracePath, "VEV_PERF_PROCESS_ID="+m.ProcessID,
244		-	"XDG_RUNTIME_DIR="+filepath.Join(runDir, "runtime"),
	244	+	cmd.Env = append(withoutEnv(os.Environ(), "VEV", "VEV_PERF_TRACE", "VEV_PERF_PROCESS_ID", "VEV_PERF_SCENARIO", "VEV_PERF_RUN", "VEV_REMOTE_TRANSPORT"), traceEnvironment(m)...)
	245	+	cmd.Env = append(cmd.Env, "XDG_RUNTIME_DIR="+filepath.Join(runDir, "runtime"),
245	246		"XDG_STATE_HOME="+filepath.Join(runDir, "state"), "TERM=xterm-256color")
246	247		if m.Role == "client" {
247	248		l.mu.Lock()

@@ -286,6 +287,9 @@ func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProces
286	287		p.pty, p.output, p.chunks, p.done = master, output, make(chan []byte, 32), make(chan struct{})
287	288		go p.copyTerminal()
288	289		} else {
	290	+	// Every launched role owns a dedicated process group, so the bounded
	291	+	// cleanup path never signals the harness/test runner's group.
	292	+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
289	293		devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
290	294		if err != nil {
291	295		return nil, err

@@ -352,11 +356,11 @@ func (l *cliLauncher) preparePeer(m processMapping, role roleCommand) (launchedP
352	356		set -eu
353	357		case "$*" in
354	358		"_stdio")
355		-	exec env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _stdio %[6]q
	359	+	exec env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q VEV_PERF_SCENARIO=%[12]q VEV_PERF_RUN=%[13]d XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _stdio %[6]q
356	360		;;
357	361		"_udp-bootstrap")
358	362		rm -f %[7]q %[8]q %[9]q
359		-	env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _udp-proxy %[6]q >%[7]q 2>%[8]q &
	363	+	env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q VEV_PERF_SCENARIO=%[12]q VEV_PERF_RUN=%[13]d XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _udp-proxy %[6]q >%[7]q 2>%[8]q &
360	364		peer=$!
361	365		printf '%%s\n' "$peer" > %[11]q
362	366		i=0

@@ -372,7 +376,7 @@ case "$*" in
372	376		;;
373	377		*) echo 'vev harness ssh seam rejected non-vev command' >&2; exit 64 ;;
374	378		esac
375		-	`, m.TracePath, m.ProcessID, filepath.Join(runDir, "runtime"), filepath.Join(runDir, "state"), l.bin, role.Args[1], ready, stderr, target, netemPort, route.pidPath)
	379	+	`, m.TracePath, m.ProcessID, filepath.Join(runDir, "runtime"), filepath.Join(runDir, "state"), l.bin, role.Args[1], ready, stderr, target, netemPort, route.pidPath, m.Scenario, m.Run)
376	380		if err := os.WriteFile(shim, []byte(body), 0o700); err != nil {
377	381		if route.netem != nil {
378	382		_ = route.netem.Close()

@@ -406,6 +410,18 @@ func remoteMode(t transport) (string, error) {
406	410		}
407	411		}
408	412
	413	+	// traceEnvironment is the complete process-local identity supplied by the
	414	+	// persisted harness processMapping. App turns Scenario/Run into production
	415	+	// mark correlation; sequence/request/epoch remain process-local operation IDs.
	416	+	func traceEnvironment(m processMapping) []string {
	417	+	return []string{
	418	+	"VEV_PERF_TRACE=" + m.TracePath,
	419	+	"VEV_PERF_PROCESS_ID=" + m.ProcessID,
	420	+	"VEV_PERF_SCENARIO=" + m.Scenario,
	421	+	"VEV_PERF_RUN=" + strconv.Itoa(m.Run),
	422	+	}
	423	+	}
	424	+
409	425		func withoutEnv(env []string, names ...string) []string {
410	426		remove := make(map[string]bool, len(names))
411	427		for _, name := range names {

@@ -498,7 +514,7 @@ func (p *cliProcess) waitForTerminalReady() error {
498	514		select {
499	515		case chunk, ok := <-p.chunks:
500	516		if !ok {
501		-	return errors.New("client terminal closed before readiness output")
	517	+	return p.terminalClosedError("readiness output")
502	518		}
503	519		if len(chunk) == 0 {
504	520		continue

@@ -520,6 +536,21 @@ func (p *cliProcess) waitForTerminalReady() error {
520	536		}
521	537		}
522	538
	539	+	func (p *cliProcess) terminalClosedError(boundary string) error {
	540	+	if p.waitErr != nil {
	541	+	select {
	542	+	case err := <-p.waitErr:
	543	+	// Close owns reaping and may need the result after a failed boundary.
	544	+	p.waitErr <- err
	545	+	if err != nil {
	546	+	return fmt.Errorf("client terminal closed before %s: %w", boundary, err)
	547	+	}
	548	+	case <-time.After(10 * time.Millisecond):
	549	+	}
	550	+	}
	551	+	return fmt.Errorf("client terminal closed before %s", boundary)
	552	+	}
	553	+
523	554		func (p *cliProcess) waitForTerminalMarker(marker, input []byte) error {
524	555		var received []byte
525	556		echo := newPTYLocalEcho(input)

@@ -529,7 +560,7 @@ func (p *cliProcess) waitForTerminalMarker(marker, input []byte) error {
529	560		select {
530	561		case chunk, ok := <-p.chunks:
531	562		if !ok {
532		-	return errors.New("client terminal closed before application output")
	563	+	return p.terminalClosedError("application output")
533	564		}
534	565		received = append(received, echo.applicationOutput(chunk)...)
535	566		case err := <-p.waitErr:

@@ -591,23 +622,44 @@ func (p *cliProcess) Close() error {
591	622		if p.done != nil {
592	623		close(p.done)
593	624		}
594		-	// Closing the client PTY causes its stdin pump to detach, then close the
595		-	// transport and wait for its adapter receive end mark. Do this before a
596		-	// signal so process-local spans are complete in the trace.
	625	+	// A public terminal client owns a shell session. Ask that shell to exit
	626	+	// before closing the PTY so its transport can finish its receive span;
	627	+	// this avoids turning ordinary harness teardown into an aborted trace.
	628	+	exited := false
	629	+	if p.pty != nil && p.waitErr != nil {
	630	+	_, _ = p.pty.Write([]byte("exit\n"))
	631	+	select {
	632	+	case <-p.waitErr:
	633	+	exited = true
	634	+	case <-p.closeDeadline():
	635	+	}
	636	+	}
597	637		if p.pty != nil {
598	638		_ = p.pty.Close()
599	639		}
600		-	if p.waitErr != nil {
	640	+	if p.waitErr != nil && !exited {
601	641		select {
602	642		case <-p.waitErr:
603		-	// The client exited through its normal detach path.
604	643		case <-p.closeDeadline():
	644	+	// Teardown is deliberately bounded: a stuck public CLI or child
	645	+	// must not hold a local harness run indefinitely.
605	646		p.forceProcessGroupCleanup()
606		-	<-p.waitErr
	647	+	select {
	648	+	case <-p.waitErr:
	649	+	case <-p.closeDeadline():
	650	+	p.forceProcessGroupKill()
	651	+	select {
	652	+	case <-p.waitErr:
	653	+	case <-p.closeDeadline():
	654	+	result = errors.New("timed out reaping public CLI process")
	655	+	}
	656	+	}
607	657		}
608	658		}
609	659		if p.output != nil {
610		-	result = p.output.Close()
	660	+	if err := p.output.Close(); result == nil {
	661	+	result = err
	662	+	}
611	663		}
612	664		})
613	665		return result

@@ -632,6 +684,16 @@ func (p *cliProcess) forceProcessGroupCleanup() {
632	684		}
633	685		}
634	686
	687	+	func (p *cliProcess) forceProcessGroupKill() {
	688	+	if p.forceKill != nil {
	689	+	p.forceKill()
	690	+	return
	691	+	}
	692	+	if p.cmd != nil && p.cmd.Process != nil {
	693	+	_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGKILL)
	694	+	}
	695	+	}
	696	+
635	697		// openPTY is the small Linux PTY boundary needed to drive the public terminal
636	698		// client. No daemon or adapter implementation is imported by this command.
637	699		func openPTY() (os.File, os.File, error) {

@@ -667,12 +729,13 @@ func inputMarker(input []byte) []byte {
667	729		}
668	730
669	731		type harness struct {
670		-	clock clock
671		-	launcher launcher
672		-	mkdir func(string) error
673		-	create func(string) (*os.File, error)
674		-	removeAll func(string) error
675		-	gitSHA func() string
	732	+	clock clock
	733	+	launcher launcher
	734	+	mkdir func(string) error
	735	+	create func(string) (*os.File, error)
	736	+	removeAll func(string) error
	737	+	gitSHA func() string
	738	+	selectScenarios func(manifest) []scenario // test-only bounded fixture seam
676	739		}
677	740
678	741		func main() {

@@ -707,14 +770,18 @@ func run(args []string, h *harness) error {
707	770		results := make([]runResult, 0, len(m.Scenarios)*opt.repetitions)
708	771		var all []int64
709	772		var localSpans []span
710		-	for _, s := range m.Scenarios {
	773	+	scenarios := m.Scenarios
	774	+	if h.selectScenarios != nil {
	775	+	scenarios = h.selectScenarios(m)
	776	+	}
	777	+	for _, s := range scenarios {
711	778		if s.InapplicableReason != "" {
712	779		continue
713	780		}
714	781		for r := 1; r <= opt.repetitions; r++ {
715	782		rr, err := h.runOne(opt, m, s, r, raw)
716	783		if err != nil {
717		-	return err
	784	+	return fmt.Errorf("scenario %s run %d: %w", s.ID, r, err)
718	785		}
719	786		results = append(results, rr)
720	787		all = append(all, rr.EndToEnd...)

@@ -1246,7 +1313,12 @@ func mergeProcessTraces(mappings []processMapping) ([]span, error) {
1246	1313		}
1247	1314		}
1248	1315		if len(starts) != 0 {
1249		-	return nil, errors.New("missing process-local span pair")
	1316	+	keys := make([]string, 0, len(starts))
	1317	+	for key := range starts {
	1318	+	keys = append(keys, key)
	1319	+	}
	1320	+	sort.Strings(keys)
	1321	+	return nil, fmt.Errorf("missing process-local span pair for %s: %s", m.ProcessID, strings.Join(keys, ", "))
1250	1322		}
1251	1323		}
1252	1324		return out, nil

 	var failures []error
 	for i := len(processes) - 1; i >= 0; i-- {
 		if err := processes[i].process.Close(); err != nil {
-			failures = append(failures, err)
+			failures = append(failures, fmt.Errorf("close %s role: %w", processes[i].mapping.Role, err))
 		}
 	}
 	return errors.Join(failures...)
 	waitErr       chan error
 	waitTimeout   func() <-chan time.Time
 	forceCleanup  func()
+	forceKill     func()
 	terminalReady bool
 }
 	}
 	cmd := exec.Command(l.bin, role.Args...)
 	runDir := filepath.Dir(m.TracePath)
-	cmd.Env = append(withoutEnv(os.Environ(), "VEV", "VEV_PERF_TRACE", "VEV_PERF_PROCESS_ID", "VEV_REMOTE_TRANSPORT"), "VEV_PERF_TRACE="+m.TracePath, "VEV_PERF_PROCESS_ID="+m.ProcessID,
+	cmd.Env = append(withoutEnv(os.Environ(), "VEV", "VEV_PERF_TRACE", "VEV_PERF_PROCESS_ID", "VEV_PERF_SCENARIO", "VEV_PERF_RUN", "VEV_REMOTE_TRANSPORT"), traceEnvironment(m)...)
-		"XDG_RUNTIME_DIR="+filepath.Join(runDir, "runtime"),
+	cmd.Env = append(cmd.Env, "XDG_RUNTIME_DIR="+filepath.Join(runDir, "runtime"),
 		"XDG_STATE_HOME="+filepath.Join(runDir, "state"), "TERM=xterm-256color")
 	if m.Role == "client" {
 		l.mu.Lock()
 		p.pty, p.output, p.chunks, p.done = master, output, make(chan []byte, 32), make(chan struct{})
 		go p.copyTerminal()
 	} else {
+		// Every launched role owns a dedicated process group, so the bounded
+		// cleanup path never signals the harness/test runner's group.
+		cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
 		devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
 		if err != nil {
 			return nil, err
 set -eu
 case "$*" in
   *"_stdio"*)
-    exec env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _stdio %[6]q
+    exec env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q VEV_PERF_SCENARIO=%[12]q VEV_PERF_RUN=%[13]d XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _stdio %[6]q
     ;;
   *"_udp-bootstrap"*)
     rm -f %[7]q %[8]q %[9]q
-    env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _udp-proxy %[6]q >%[7]q 2>%[8]q &
+    env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q VEV_PERF_SCENARIO=%[12]q VEV_PERF_RUN=%[13]d XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _udp-proxy %[6]q >%[7]q 2>%[8]q &
     peer=$!
     printf '%%s\n' "$peer" > %[11]q
     i=0
     ;;
   *) echo 'vev harness ssh seam rejected non-vev command' >&2; exit 64 ;;
 esac
-`, m.TracePath, m.ProcessID, filepath.Join(runDir, "runtime"), filepath.Join(runDir, "state"), l.bin, role.Args[1], ready, stderr, target, netemPort, route.pidPath)
+`, m.TracePath, m.ProcessID, filepath.Join(runDir, "runtime"), filepath.Join(runDir, "state"), l.bin, role.Args[1], ready, stderr, target, netemPort, route.pidPath, m.Scenario, m.Run)
 	if err := os.WriteFile(shim, []byte(body), 0o700); err != nil {
 		if route.netem != nil {
 			_ = route.netem.Close()
 	}
 }
+// traceEnvironment is the complete process-local identity supplied by the
+// persisted harness processMapping. App turns Scenario/Run into production
+// mark correlation; sequence/request/epoch remain process-local operation IDs.
+func traceEnvironment(m processMapping) []string {
+	return []string{
+		"VEV_PERF_TRACE=" + m.TracePath,
+		"VEV_PERF_PROCESS_ID=" + m.ProcessID,
+		"VEV_PERF_SCENARIO=" + m.Scenario,
+		"VEV_PERF_RUN=" + strconv.Itoa(m.Run),
+	}
+}
 func withoutEnv(env []string, names ...string) []string {
 	remove := make(map[string]bool, len(names))
 	for _, name := range names {
 		select {
 		case chunk, ok := <-p.chunks:
 			if !ok {
-				return errors.New("client terminal closed before readiness output")
+				return p.terminalClosedError("readiness output")
 			}
 			if len(chunk) == 0 {
 				continue
 	}
 }
+func (p *cliProcess) terminalClosedError(boundary string) error {
+	if p.waitErr != nil {
+		select {
+		case err := <-p.waitErr:
+			// Close owns reaping and may need the result after a failed boundary.
+			p.waitErr <- err
+			if err != nil {
+				return fmt.Errorf("client terminal closed before %s: %w", boundary, err)
+			}
+		case <-time.After(10 * time.Millisecond):
+		}
+	}
+	return fmt.Errorf("client terminal closed before %s", boundary)
+}
 func (p *cliProcess) waitForTerminalMarker(marker, input []byte) error {
 	var received []byte
 	echo := newPTYLocalEcho(input)
 		select {
 		case chunk, ok := <-p.chunks:
 			if !ok {
-				return errors.New("client terminal closed before application output")
+				return p.terminalClosedError("application output")
 			}
 			received = append(received, echo.applicationOutput(chunk)...)
 		case err := <-p.waitErr:
 		if p.done != nil {
 			close(p.done)
 		}
-		// Closing the client PTY causes its stdin pump to detach, then close the
+		// A public terminal client owns a shell session. Ask that shell to exit
-		// transport and wait for its adapter receive end mark. Do this before a
+		// before closing the PTY so its transport can finish its receive span;
-		// signal so process-local spans are complete in the trace.
+		// this avoids turning ordinary harness teardown into an aborted trace.
+		exited := false
+		if p.pty != nil && p.waitErr != nil {
+			_, _ = p.pty.Write([]byte("exit\n"))
+			select {
+			case <-p.waitErr:
+				exited = true
+			case <-p.closeDeadline():
+			}
+		}
 		if p.pty != nil {
 			_ = p.pty.Close()
 		}
-		if p.waitErr != nil {
+		if p.waitErr != nil && !exited {
 			select {
 			case <-p.waitErr:
-				// The client exited through its normal detach path.
 			case <-p.closeDeadline():
+				// Teardown is deliberately bounded: a stuck public CLI or child
+				// must not hold a local harness run indefinitely.
 				p.forceProcessGroupCleanup()
-				<-p.waitErr
+				select {
+				case <-p.waitErr:
+				case <-p.closeDeadline():
+					p.forceProcessGroupKill()
+					select {
+					case <-p.waitErr:
+					case <-p.closeDeadline():
+						result = errors.New("timed out reaping public CLI process")
+					}
+				}
 			}
 		}
 		if p.output != nil {
-			result = p.output.Close()
+			if err := p.output.Close(); result == nil {
+				result = err
+			}
 		}
 	})
 	return result
 	}
 }
+func (p *cliProcess) forceProcessGroupKill() {
+	if p.forceKill != nil {
+		p.forceKill()
+		return
+	}
+	if p.cmd != nil && p.cmd.Process != nil {
+		_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGKILL)
+	}
+}
 // openPTY is the small Linux PTY boundary needed to drive the public terminal
 // client. No daemon or adapter implementation is imported by this command.
 func openPTY() (*os.File, *os.File, error) {
 }
 type harness struct {
-	clock     clock
+	clock           clock
-	launcher  launcher
+	launcher        launcher
-	mkdir     func(string) error
+	mkdir           func(string) error
-	create    func(string) (*os.File, error)
+	create          func(string) (*os.File, error)
-	removeAll func(string) error
+	removeAll       func(string) error
-	gitSHA    func() string
+	gitSHA          func() string
+	selectScenarios func(manifest) []scenario // test-only bounded fixture seam
 }
 func main() {
 	results := make([]runResult, 0, len(m.Scenarios)*opt.repetitions)
 	var all []int64
 	var localSpans []span
-	for _, s := range m.Scenarios {
+	scenarios := m.Scenarios
+	if h.selectScenarios != nil {
+		scenarios = h.selectScenarios(m)
+	}
+	for _, s := range scenarios {
 		if s.InapplicableReason != "" {
 			continue
 		}
 		for r := 1; r <= opt.repetitions; r++ {
 			rr, err := h.runOne(opt, m, s, r, raw)
 			if err != nil {
-				return err
+				return fmt.Errorf("scenario %s run %d: %w", s.ID, r, err)
 			}
 			results = append(results, rr)
 			all = append(all, rr.EndToEnd...)
 			}
 		}
 		if len(starts) != 0 {
-			return nil, errors.New("missing process-local span pair")
+			keys := make([]string, 0, len(starts))
+			for key := range starts {
+				keys = append(keys, key)
+			}
+			sort.Strings(keys)
+			return nil, fmt.Errorf("missing process-local span pair for %s: %s", m.ProcessID, strings.Join(keys, ", "))
 		}
 	}
 	return out, nil

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


@@ -6,6 +6,7 @@ import (
6	6		"errors"
7	7		"net"
8	8		"os"
	9	+	"os/exec"
9	10		"path/filepath"
10	11		"strconv"
11	12		"strings"

@@ -75,6 +76,117 @@ func (l *fakeLauncher) Launch(m processMapping, command roleCommand) (launchedPr
75	76		return p, nil
76	77		}
77	78
	79	+	func TestCLIProcessCloseIsBoundedWhenPublicRoleDoesNotExit(t *testing.T) {
	80	+	deadline := make(chan time.Time)
	81	+	close(deadline)
	82	+	var term, kill int
	83	+	p := &cliProcess{
	84	+	waitErr: make(chan error),
	85	+	waitTimeout: func() <-chan time.Time { return deadline },
	86	+	forceCleanup: func() { term++ },
	87	+	forceKill: func() { kill++ },
	88	+	}
	89	+	if err := p.Close(); err == nil {
	90	+	t.Fatal("stuck public role close succeeded")
	91	+	}
	92	+	if term != 1 \|\| kill != 1 {
	93	+	t.Fatalf("TERM/KILL calls = %d/%d, want 1/1", term, kill)
	94	+	}
	95	+	}
	96	+
	97	+	func TestHarnessCanonicalLocalSmokeRealRoles(t *testing.T) {
	98	+	if testing.Short() {
	99	+	t.Skip("public CLI smoke")
	100	+	}
	101	+	root, err := filepath.Abs(filepath.Join("..", ".."))
	102	+	if err != nil {
	103	+	t.Fatal(err)
	104	+	}
	105	+	bin := filepath.Join(t.TempDir(), "vev")
	106	+	build := exec.Command("/usr/local/go/bin/go", "build", "-o", bin, "./")
	107	+	build.Dir = root
	108	+	if output, err := build.CombinedOutput(); err != nil {
	109	+	t.Fatalf("build public CLI: %v\n%s", err, output)
	110	+	}
	111	+	m, err := readManifest(filepath.Join(root, "testdata", "perf", "manifest.json"))
	112	+	if err != nil {
	113	+	t.Fatal(err)
	114	+	}
	115	+	if err := validateManifest(m); err != nil {
	116	+	t.Fatal(err)
	117	+	}
	118	+	var local scenario
	119	+	for _, s := range m.Scenarios {
	120	+	if s.Topology == "1x4" && s.Workload == "idle" && s.Transport == "local" {
	121	+	local = s
	122	+	break
	123	+	}
	124	+	}
	125	+	if local.ID == "" {
	126	+	t.Fatal("canonical local scenario missing")
	127	+	}
	128	+	dir := t.TempDir()
	129	+	raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE\|os.O_EXCL\|os.O_WRONLY, 0o600)
	130	+	if err != nil {
	131	+	t.Fatal(err)
	132	+	}
	133	+	defer raw.Close()
	134	+	h := defaultHarness()
	135	+	h.clock = &fakeClock{} // injected duration keeps the public 30s contract bounded in test
	136	+	h.launcher = &cliLauncher{bin: bin}
	137	+	result, err := h.runOne(options{vevBin: bin, out: dir, duration: minimumDuration}, m, local, 1, raw)
	138	+	if err != nil {
	139	+	t.Fatal(err)
	140	+	}
	141	+	if result.Samples != 1 \|\| len(result.Processes) != 2 {
	142	+	t.Fatalf("canonical local smoke result=%+v", result)
	143	+	}
	144	+	}
	145	+
	146	+	func TestHarnessTestSelectionKeepsCanonicalManifestValidation(t *testing.T) {
	147	+	root, err := filepath.Abs(filepath.Join("..", ".."))
	148	+	if err != nil {
	149	+	t.Fatal(err)
	150	+	}
	151	+	manifestPath := filepath.Join(root, "testdata", "perf", "manifest.json")
	152	+	m, err := readManifest(manifestPath)
	153	+	if err != nil {
	154	+	t.Fatal(err)
	155	+	}
	156	+	if err := validateManifest(m); err != nil {
	157	+	t.Fatal(err)
	158	+	}
	159	+	selected := m.Scenarios[0]
	160	+	h := defaultHarness()
	161	+	h.clock, h.launcher = &fakeClock{}, &fakeLauncher{}
	162	+	h.selectScenarios = func(manifest) []scenario { return []scenario{selected} }
	163	+	out := filepath.Join(t.TempDir(), "out")
	164	+	if err := run([]string{"--vev-bin", "ignored", "--manifest", manifestPath, "--out", out, "--warmup", "0s", "--duration", "30s", "--repetitions", "10"}, h); err != nil {
	165	+	t.Fatal(err)
	166	+	}
	167	+	var runs []runResult
	168	+	b, err := os.ReadFile(filepath.Join(out, "runs.json"))
	169	+	if err != nil {
	170	+	t.Fatal(err)
	171	+	}
	172	+	if err := json.Unmarshal(b, &runs); err != nil {
	173	+	t.Fatal(err)
	174	+	}
	175	+	if len(runs) != 10 {
	176	+	t.Fatalf("selected canonical runs=%d, want 10", len(runs))
	177	+	}
	178	+	}
	179	+
	180	+	func TestTraceEnvironmentMatchesProcessMapping(t *testing.T) {
	181	+	mapping := processMapping{TracePath: "/tmp/trace.jsonl", ProcessID: "p-01", Scenario: "1x4-idle-local", Run: 3}
	182	+	got := strings.Join(traceEnvironment(mapping), "\n")
	183	+	for _, want := range []string{"VEV_PERF_TRACE=/tmp/trace.jsonl", "VEV_PERF_PROCESS_ID=p-01", "VEV_PERF_SCENARIO=1x4-idle-local", "VEV_PERF_RUN=3"} {
	184	+	if !strings.Contains(got, want) {
	185	+	t.Errorf("trace environment missing %q: %s", want, got)
	186	+	}
	187	+	}
	188	+	}
	189	+
78	190		func TestHarnessFlagsRejectShortMeasurements(t *testing.T) {
79	191		for _, args := range [][]string{
80	192		{"--vev-bin", "vev", "--manifest", "m", "--out", "o", "--duration", "29s", "--repetitions", "10"},

@@ -190,10 +302,18 @@ type fakeOutput struct {
190	302		func (o *fakeOutput) Sync() error { o.syncs++; return nil }
191	303		func (*fakeOutput) Close() error { return nil }
192	304
193		-	type orderedPTY struct{ close func() }
	305	+	type orderedPTY struct {
	306	+	write func([]byte)
	307	+	close func()
	308	+	}
194	309
195		-	func (*orderedPTY) Read([]byte) (int, error) { return 0, os.ErrClosed }
196		-	func (*orderedPTY) Write(b []byte) (int, error) { return len(b), nil }
	310	+	func (*orderedPTY) Read([]byte) (int, error) { return 0, os.ErrClosed }
	311	+	func (p *orderedPTY) Write(b []byte) (int, error) {
	312	+	if p.write != nil {
	313	+	p.write(b)
	314	+	}
	315	+	return len(b), nil
	316	+	}
197	317		func (p *orderedPTY) Close() error {
198	318		p.close()
199	319		return nil

@@ -213,13 +333,19 @@ func TestCLIProcessCloseGracefulDetachCompletesSpanBeforeForcedCleanup(t *testin
213	333		waitErr := make(chan error, 1)
214	334		timeout := make(chan time.Time)
215	335		p := &cliProcess{
216		-	pty: &orderedPTY{close: func() {
217		-	order = append(order, "pty_closed")
218		-	// The graceful client detach has closed its transport and emitted its
219		-	// adapter receive-end mark before cmd.Wait reports completion.
220		-	order = append(order, "adapter_receive_end")
221		-	waitErr <- nil
222		-	}},
	336	+	pty: &orderedPTY{
	337	+	write: func(input []byte) {
	338	+	if string(input) != "exit\n" {
	339	+	t.Fatalf("graceful exit input=%q", input)
	340	+	}
	341	+	order = append(order, "exit_written")
	342	+	// The graceful client exit has closed its transport and emitted
	343	+	// its adapter receive-end mark before cmd.Wait completes.
	344	+	order = append(order, "adapter_receive_end")
	345	+	waitErr <- nil
	346	+	},
	347	+	close: func() { order = append(order, "pty_closed") },
	348	+	},
223	349		output: &orderedOutput{close: func() { order = append(order, "output_closed") }},
224	350		waitErr: waitErr,
225	351		waitTimeout: func() <-chan time.Time { return timeout },

@@ -231,7 +357,7 @@ func TestCLIProcessCloseGracefulDetachCompletesSpanBeforeForcedCleanup(t *testin
231	357		if err := p.Close(); err != nil {
232	358		t.Fatal(err)
233	359		}
234		-	if !equalStrings(order, []string{"pty_closed", "adapter_receive_end", "output_closed"}) {
	360	+	if !equalStrings(order, []string{"exit_written", "adapter_receive_end", "pty_closed", "output_closed"}) {
235	361		t.Fatalf("cleanup order=%q", order)
236	362		}
237	363		}

@@ -569,7 +695,7 @@ func TestHarnessPreservesWorkloadAndCleanupErrors(t *testing.T) {
569	695		if !errors.Is(err, primary) \|\| !errors.Is(err, cleanup) {
570	696		t.Fatalf("workload and cleanup errors were not both preserved: %v", err)
571	697		}
572		-	if got, want := err.Error(), primary.Error()+"\n"+cleanup.Error(); got != want {
	698	+	if got, want := err.Error(), primary.Error()+"\nclose daemon role: "+cleanup.Error(); got != want {
573	699		t.Fatalf("error ordering=%q want %q", got, want)
574	700		}
575	701		for i, p := range launcher.process {


@@ -6,6 +6,7 @@ import (
6		"errors"	6		"errors"
7		"net"	7		"net"
8		"os"	8		"os"
			9	+	"os/exec"
9		"path/filepath"	10		"path/filepath"
10		"strconv"	11		"strconv"
11		"strings"	12		"strings"

@@ -75,6 +76,117 @@ func (l *fakeLauncher) Launch(m processMapping, command roleCommand) (launchedPr
75		return p, nil	76		return p, nil
76		}	77		}
77			78
			79	+	func TestCLIProcessCloseIsBoundedWhenPublicRoleDoesNotExit(t *testing.T) {
			80	+	deadline := make(chan time.Time)
			81	+	close(deadline)
			82	+	var term, kill int
			83	+	p := &cliProcess{
			84	+	waitErr: make(chan error),
			85	+	waitTimeout: func() <-chan time.Time { return deadline },
			86	+	forceCleanup: func() { term++ },
			87	+	forceKill: func() { kill++ },
			88	+	}
			89	+	if err := p.Close(); err == nil {
			90	+	t.Fatal("stuck public role close succeeded")
			91	+	}
			92	+	if term != 1 \|\| kill != 1 {
			93	+	t.Fatalf("TERM/KILL calls = %d/%d, want 1/1", term, kill)
			94	+	}
			95	+	}
			96	+
			97	+	func TestHarnessCanonicalLocalSmokeRealRoles(t *testing.T) {
			98	+	if testing.Short() {
			99	+	t.Skip("public CLI smoke")
			100	+	}
			101	+	root, err := filepath.Abs(filepath.Join("..", ".."))
			102	+	if err != nil {
			103	+	t.Fatal(err)
			104	+	}
			105	+	bin := filepath.Join(t.TempDir(), "vev")
			106	+	build := exec.Command("/usr/local/go/bin/go", "build", "-o", bin, "./")
			107	+	build.Dir = root
			108	+	if output, err := build.CombinedOutput(); err != nil {
			109	+	t.Fatalf("build public CLI: %v\n%s", err, output)
			110	+	}
			111	+	m, err := readManifest(filepath.Join(root, "testdata", "perf", "manifest.json"))
			112	+	if err != nil {
			113	+	t.Fatal(err)
			114	+	}
			115	+	if err := validateManifest(m); err != nil {
			116	+	t.Fatal(err)
			117	+	}
			118	+	var local scenario
			119	+	for _, s := range m.Scenarios {
			120	+	if s.Topology == "1x4" && s.Workload == "idle" && s.Transport == "local" {
			121	+	local = s
			122	+	break
			123	+	}
			124	+	}
			125	+	if local.ID == "" {
			126	+	t.Fatal("canonical local scenario missing")
			127	+	}
			128	+	dir := t.TempDir()
			129	+	raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE\|os.O_EXCL\|os.O_WRONLY, 0o600)
			130	+	if err != nil {
			131	+	t.Fatal(err)
			132	+	}
			133	+	defer raw.Close()
			134	+	h := defaultHarness()
			135	+	h.clock = &fakeClock{} // injected duration keeps the public 30s contract bounded in test
			136	+	h.launcher = &cliLauncher{bin: bin}
			137	+	result, err := h.runOne(options{vevBin: bin, out: dir, duration: minimumDuration}, m, local, 1, raw)
			138	+	if err != nil {
			139	+	t.Fatal(err)
			140	+	}
			141	+	if result.Samples != 1 \|\| len(result.Processes) != 2 {
			142	+	t.Fatalf("canonical local smoke result=%+v", result)
			143	+	}
			144	+	}
			145	+
			146	+	func TestHarnessTestSelectionKeepsCanonicalManifestValidation(t *testing.T) {
			147	+	root, err := filepath.Abs(filepath.Join("..", ".."))
			148	+	if err != nil {
			149	+	t.Fatal(err)
			150	+	}
			151	+	manifestPath := filepath.Join(root, "testdata", "perf", "manifest.json")
			152	+	m, err := readManifest(manifestPath)
			153	+	if err != nil {
			154	+	t.Fatal(err)
			155	+	}
			156	+	if err := validateManifest(m); err != nil {
			157	+	t.Fatal(err)
			158	+	}
			159	+	selected := m.Scenarios[0]
			160	+	h := defaultHarness()
			161	+	h.clock, h.launcher = &fakeClock{}, &fakeLauncher{}
			162	+	h.selectScenarios = func(manifest) []scenario { return []scenario{selected} }
			163	+	out := filepath.Join(t.TempDir(), "out")
			164	+	if err := run([]string{"--vev-bin", "ignored", "--manifest", manifestPath, "--out", out, "--warmup", "0s", "--duration", "30s", "--repetitions", "10"}, h); err != nil {
			165	+	t.Fatal(err)
			166	+	}
			167	+	var runs []runResult
			168	+	b, err := os.ReadFile(filepath.Join(out, "runs.json"))
			169	+	if err != nil {
			170	+	t.Fatal(err)
			171	+	}
			172	+	if err := json.Unmarshal(b, &runs); err != nil {
			173	+	t.Fatal(err)
			174	+	}
			175	+	if len(runs) != 10 {
			176	+	t.Fatalf("selected canonical runs=%d, want 10", len(runs))
			177	+	}
			178	+	}
			179	+
			180	+	func TestTraceEnvironmentMatchesProcessMapping(t *testing.T) {
			181	+	mapping := processMapping{TracePath: "/tmp/trace.jsonl", ProcessID: "p-01", Scenario: "1x4-idle-local", Run: 3}
			182	+	got := strings.Join(traceEnvironment(mapping), "\n")
			183	+	for _, want := range []string{"VEV_PERF_TRACE=/tmp/trace.jsonl", "VEV_PERF_PROCESS_ID=p-01", "VEV_PERF_SCENARIO=1x4-idle-local", "VEV_PERF_RUN=3"} {
			184	+	if !strings.Contains(got, want) {
			185	+	t.Errorf("trace environment missing %q: %s", want, got)
			186	+	}
			187	+	}
			188	+	}
			189	+
78		func TestHarnessFlagsRejectShortMeasurements(t *testing.T) {	190		func TestHarnessFlagsRejectShortMeasurements(t *testing.T) {
79		for _, args := range [][]string{	191		for _, args := range [][]string{
80		{"--vev-bin", "vev", "--manifest", "m", "--out", "o", "--duration", "29s", "--repetitions", "10"},	192		{"--vev-bin", "vev", "--manifest", "m", "--out", "o", "--duration", "29s", "--repetitions", "10"},

@@ -190,10 +302,18 @@ type fakeOutput struct {
190		func (o *fakeOutput) Sync() error { o.syncs++; return nil }	302		func (o *fakeOutput) Sync() error { o.syncs++; return nil }
191		func (*fakeOutput) Close() error { return nil }	303		func (*fakeOutput) Close() error { return nil }
192			304
193	-	type orderedPTY struct{ close func() }	305	+	type orderedPTY struct {
			306	+	write func([]byte)
			307	+	close func()
			308	+	}
194			309
195	-	func (*orderedPTY) Read([]byte) (int, error) { return 0, os.ErrClosed }	310	+	func (*orderedPTY) Read([]byte) (int, error) { return 0, os.ErrClosed }
196	-	func (*orderedPTY) Write(b []byte) (int, error) { return len(b), nil }	311	+	func (p *orderedPTY) Write(b []byte) (int, error) {
			312	+	if p.write != nil {
			313	+	p.write(b)
			314	+	}
			315	+	return len(b), nil
			316	+	}
197		func (p *orderedPTY) Close() error {	317		func (p *orderedPTY) Close() error {
198		p.close()	318		p.close()
199		return nil	319		return nil

@@ -213,13 +333,19 @@ func TestCLIProcessCloseGracefulDetachCompletesSpanBeforeForcedCleanup(t *testin
213		waitErr := make(chan error, 1)	333		waitErr := make(chan error, 1)
214		timeout := make(chan time.Time)	334		timeout := make(chan time.Time)
215		p := &cliProcess{	335		p := &cliProcess{
216	-	pty: &orderedPTY{close: func() {	336	+	pty: &orderedPTY{
217	-	order = append(order, "pty_closed")	337	+	write: func(input []byte) {
218	-	// The graceful client detach has closed its transport and emitted its	338	+	if string(input) != "exit\n" {
219	-	// adapter receive-end mark before cmd.Wait reports completion.	339	+	t.Fatalf("graceful exit input=%q", input)
220	-	order = append(order, "adapter_receive_end")	340	+	}
221	-	waitErr <- nil	341	+	order = append(order, "exit_written")
222	-	}},	342	+	// The graceful client exit has closed its transport and emitted
			343	+	// its adapter receive-end mark before cmd.Wait completes.
			344	+	order = append(order, "adapter_receive_end")
			345	+	waitErr <- nil
			346	+	},
			347	+	close: func() { order = append(order, "pty_closed") },
			348	+	},
223		output: &orderedOutput{close: func() { order = append(order, "output_closed") }},	349		output: &orderedOutput{close: func() { order = append(order, "output_closed") }},
224		waitErr: waitErr,	350		waitErr: waitErr,
225		waitTimeout: func() <-chan time.Time { return timeout },	351		waitTimeout: func() <-chan time.Time { return timeout },

@@ -231,7 +357,7 @@ func TestCLIProcessCloseGracefulDetachCompletesSpanBeforeForcedCleanup(t *testin
231		if err := p.Close(); err != nil {	357		if err := p.Close(); err != nil {
232		t.Fatal(err)	358		t.Fatal(err)
233		}	359		}
234	-	if !equalStrings(order, []string{"pty_closed", "adapter_receive_end", "output_closed"}) {	360	+	if !equalStrings(order, []string{"exit_written", "adapter_receive_end", "pty_closed", "output_closed"}) {
235		t.Fatalf("cleanup order=%q", order)	361		t.Fatalf("cleanup order=%q", order)
236		}	362		}
237		}	363		}

@@ -569,7 +695,7 @@ func TestHarnessPreservesWorkloadAndCleanupErrors(t *testing.T) {
569		if !errors.Is(err, primary) \|\| !errors.Is(err, cleanup) {	695		if !errors.Is(err, primary) \|\| !errors.Is(err, cleanup) {
570		t.Fatalf("workload and cleanup errors were not both preserved: %v", err)	696		t.Fatalf("workload and cleanup errors were not both preserved: %v", err)
571		}	697		}
572	-	if got, want := err.Error(), primary.Error()+"\n"+cleanup.Error(); got != want {	698	+	if got, want := err.Error(), primary.Error()+"\nclose daemon role: "+cleanup.Error(); got != want {
573		t.Fatalf("error ordering=%q want %q", got, want)	699		t.Fatalf("error ordering=%q want %q", got, want)
574		}	700		}
575		for i, p := range launcher.process {	701		for i, p := range launcher.process {

M internal/app/run.go +34 -9


@@ -247,7 +247,30 @@ func performanceTrace(clk ports.Clock) (ports.RuntimeObserver, io.Closer, error)
247	247		if path == "" {
248	248		return nil, nil, nil
249	249		}
250		-	return observability.NewJSONL(path, clk, processID)
	250	+	observer, closer, err := observability.NewJSONL(path, clk, processID)
	251	+	if err != nil {
	252	+	return nil, nil, err
	253	+	}
	254	+	// The harness supplies these manifest fields to every launched role. Keep a
	255	+	// valid standalone trace for operators that set only the original trace
	256	+	// variables, while ensuring harness traces match their process mapping.
	257	+	inputs := ports.RuntimeCorrelationInputs{Scenario: os.Getenv("VEV_PERF_SCENARIO"), Run: 1}
	258	+	if inputs.Scenario == "" {
	259	+	inputs.Scenario = "runtime"
	260	+	}
	261	+	if rawRun := os.Getenv("VEV_PERF_RUN"); rawRun != "" {
	262	+	inputs.Run, err = strconv.ParseUint(rawRun, 10, 64)
	263	+	if err != nil \|\| inputs.Run == 0 {
	264	+	_ = closer.Close()
	265	+	return nil, nil, fmt.Errorf("invalid VEV_PERF_RUN %q", rawRun)
	266	+	}
	267	+	}
	268	+	observer, err = ports.NewRuntimeCorrelationObserver(observer, inputs)
	269	+	if err != nil {
	270	+	_ = closer.Close()
	271	+	return nil, nil, err
	272	+	}
	273	+	return observer, closer, nil
251	274		}
252	275
253	276		func configureLogging(component logging.Component, rotateAtRuntime bool) (*slog.Logger, io.Closer, error) {

@@ -780,6 +803,13 @@ func runUDPProxy(ctx context.Context, session string, ready io.Writer) error {
780	803		defer func() { _ = logCloser.Close() }()
781	804		log.Debug("udp proxy starting", "session", session)
782	805
	806	+	observer, observerCloser, err := performanceTrace(clock.New())
	807	+	if err != nil {
	808	+	return fmt.Errorf("vev: performance trace: %w", err)
	809	+	}
	810	+	if observerCloser != nil {
	811	+	defer func() { _ = observerCloser.Close() }()
	812	+	}
783	813		portRange, err := parseUDPPortRange(os.Getenv(envUDPPortRange))
784	814		if err != nil {
785	815		return err

@@ -793,20 +823,15 @@ func runUDPProxy(ctx context.Context, session string, ready io.Writer) error {
793	823		if _, err := rand.Read(key); err != nil {
794	824		return err
795	825		}
796		-	daemonTr, err := ensureDaemon(ctx, ipc.SocketDir(), realDial, realSpawn, defaultBackoff)
	826	+	daemonTr, err := ensureDaemon(ctx, ipc.SocketDir(), func(dir string) (ports.Transport, error) {
	827	+	return ipc.Dial(dir, ipc.WithRuntimeObserver(observer))
	828	+	}, realSpawn, defaultBackoff)
797	829		if err != nil {
798	830		return err
799	831		}
800	832		defer func() { _ = daemonTr.Close() }()
801	833		udpOptions := udpProxyClientTransportOptions
802	834		udpOptions.Observe = dgram.DiagnosticLogObserver(log)
803		-	observer, observerCloser, err := performanceTrace(clock.New())
804		-	if err != nil {
805		-	return fmt.Errorf("vev: performance trace: %w", err)
806		-	}
807		-	if observerCloser != nil {
808		-	defer func() { _ = observerCloser.Close() }()
809		-	}
810	835		dg, err := dgram.NewTransportWithOptions(conn, nil, key, 2, 1, udpOptions, dgram.WithRuntimeObserver(observer))
811	836		if err != nil {
812	837		return err

 	if path == "" {
 		return nil, nil, nil
 	}
-	return observability.NewJSONL(path, clk, processID)
+	observer, closer, err := observability.NewJSONL(path, clk, processID)
+	if err != nil {
+		return nil, nil, err
+	}
+	// The harness supplies these manifest fields to every launched role. Keep a
+	// valid standalone trace for operators that set only the original trace
+	// variables, while ensuring harness traces match their process mapping.
+	inputs := ports.RuntimeCorrelationInputs{Scenario: os.Getenv("VEV_PERF_SCENARIO"), Run: 1}
+	if inputs.Scenario == "" {
+		inputs.Scenario = "runtime"
+	}
+	if rawRun := os.Getenv("VEV_PERF_RUN"); rawRun != "" {
+		inputs.Run, err = strconv.ParseUint(rawRun, 10, 64)
+		if err != nil || inputs.Run == 0 {
+			_ = closer.Close()
+			return nil, nil, fmt.Errorf("invalid VEV_PERF_RUN %q", rawRun)
+		}
+	}
+	observer, err = ports.NewRuntimeCorrelationObserver(observer, inputs)
+	if err != nil {
+		_ = closer.Close()
+		return nil, nil, err
+	}
+	return observer, closer, nil
 }
 func configureLogging(component logging.Component, rotateAtRuntime bool) (*slog.Logger, io.Closer, error) {
 	defer func() { _ = logCloser.Close() }()
 	log.Debug("udp proxy starting", "session", session)
+	observer, observerCloser, err := performanceTrace(clock.New())
+	if err != nil {
+		return fmt.Errorf("vev: performance trace: %w", err)
+	}
+	if observerCloser != nil {
+		defer func() { _ = observerCloser.Close() }()
+	}
 	portRange, err := parseUDPPortRange(os.Getenv(envUDPPortRange))
 	if err != nil {
 		return err
 	if _, err := rand.Read(key); err != nil {
 		return err
 	}
-	daemonTr, err := ensureDaemon(ctx, ipc.SocketDir(), realDial, realSpawn, defaultBackoff)
+	daemonTr, err := ensureDaemon(ctx, ipc.SocketDir(), func(dir string) (ports.Transport, error) {
+		return ipc.Dial(dir, ipc.WithRuntimeObserver(observer))
+	}, realSpawn, defaultBackoff)
 	if err != nil {
 		return err
 	}
 	defer func() { _ = daemonTr.Close() }()
 	udpOptions := udpProxyClientTransportOptions
 	udpOptions.Observe = dgram.DiagnosticLogObserver(log)
-	observer, observerCloser, err := performanceTrace(clock.New())
-	if err != nil {
-		return fmt.Errorf("vev: performance trace: %w", err)
-	}
-	if observerCloser != nil {
-		defer func() { _ = observerCloser.Close() }()
-	}
 	dg, err := dgram.NewTransportWithOptions(conn, nil, key, 2, 1, udpOptions, dgram.WithRuntimeObserver(observer))
 	if err != nil {
 		return err

M internal/app/transport_observability_test.go +50 -0

@@ -1,11 +1,61 @@
1	1		package app
2	2
3	3		import (
	4	+	"bufio"
	5	+	"encoding/json"
4	6		"os"
	7	+	"path/filepath"
5	8		"strings"
6	9		"testing"
	10	+	"time"
	11	+
	12	+	"github.com/bnema/vev/internal/ports"
7	13		)
8	14
	15	+	func TestPerformanceTraceUsesHarnessProcessMapping(t *testing.T) {
	16	+	path := filepath.Join(t.TempDir(), "process.jsonl")
	17	+	t.Setenv("VEV_PERF_TRACE", path)
	18	+	t.Setenv("VEV_PERF_PROCESS_ID", "idle-local-r001-daemon-01")
	19	+	t.Setenv("VEV_PERF_SCENARIO", "1x4-idle-local")
	20	+	t.Setenv("VEV_PERF_RUN", "1")
	21	+
	22	+	observer, closer, err := performanceTrace(&traceTestClock{})
	23	+	if err != nil {
	24	+	t.Fatal(err)
	25	+	}
	26	+	if observer == nil \|\| closer == nil {
	27	+	t.Fatal("performance trace was not configured")
	28	+	}
	29	+	observer.ObserveRuntime(ports.RuntimeMark{Schema: ports.RuntimeMarkSchema, Component: "daemon", Scenario: "runtime", Run: 1, Kind: ports.RuntimeEmitEnd, Valid: true})
	30	+	if err := closer.Close(); err != nil {
	31	+	t.Fatal(err)
	32	+	}
	33	+	f, err := os.Open(path)
	34	+	if err != nil {
	35	+	t.Fatal(err)
	36	+	}
	37	+	defer f.Close()
	38	+	var mark struct {
	39	+	ProcessID string `json:"process_id"`
	40	+	Scenario string `json:"scenario"`
	41	+	Run uint64 `json:"run"`
	42	+	Sequence uint64 `json:"sequence"`
	43	+	RequestID uint64 `json:"request_id"`
	44	+	Epoch uint64 `json:"epoch"`
	45	+	}
	46	+	if err := json.NewDecoder(bufio.NewReader(f)).Decode(&mark); err != nil {
	47	+	t.Fatal(err)
	48	+	}
	49	+	if mark.ProcessID != "idle-local-r001-daemon-01" \|\| mark.Scenario != "1x4-idle-local" \|\| mark.Run != 1 \|\| mark.Sequence == 0 \|\| mark.RequestID == 0 \|\| mark.Epoch == 0 {
	50	+	t.Fatalf("mark does not match harness mapping: %+v", mark)
	51	+	}
	52	+	}
	53	+
	54	+	type traceTestClock struct{}
	55	+
	56	+	func (*traceTestClock) Now() time.Time { return time.Unix(0, 1) }
	57	+	func (*traceTestClock) NewTimer(time.Duration) ports.Timer { return nil }
	58	+
9	59		func TestPerformanceTraceAppWiresOneObserverPerProcess(t *testing.T) {
10	60		source, err := os.ReadFile("run.go")
11	61		if err != nil {

@@ -1,11 +1,61 @@
1	package app	1		package app
2		2
3	import (	3		import (
		4	+	"bufio"
		5	+	"encoding/json"
4	"os"	6		"os"
		7	+	"path/filepath"
5	"strings"	8		"strings"
6	"testing"	9		"testing"
		10	+	"time"
		11	+
		12	+	"github.com/bnema/vev/internal/ports"
7	)	13		)
8		14
		15	+	func TestPerformanceTraceUsesHarnessProcessMapping(t *testing.T) {
		16	+	path := filepath.Join(t.TempDir(), "process.jsonl")
		17	+	t.Setenv("VEV_PERF_TRACE", path)
		18	+	t.Setenv("VEV_PERF_PROCESS_ID", "idle-local-r001-daemon-01")
		19	+	t.Setenv("VEV_PERF_SCENARIO", "1x4-idle-local")
		20	+	t.Setenv("VEV_PERF_RUN", "1")
		21	+
		22	+	observer, closer, err := performanceTrace(&traceTestClock{})
		23	+	if err != nil {
		24	+	t.Fatal(err)
		25	+	}
		26	+	if observer == nil \|\| closer == nil {
		27	+	t.Fatal("performance trace was not configured")
		28	+	}
		29	+	observer.ObserveRuntime(ports.RuntimeMark{Schema: ports.RuntimeMarkSchema, Component: "daemon", Scenario: "runtime", Run: 1, Kind: ports.RuntimeEmitEnd, Valid: true})
		30	+	if err := closer.Close(); err != nil {
		31	+	t.Fatal(err)
		32	+	}
		33	+	f, err := os.Open(path)
		34	+	if err != nil {
		35	+	t.Fatal(err)
		36	+	}
		37	+	defer f.Close()
		38	+	var mark struct {
		39	+	ProcessID string `json:"process_id"`
		40	+	Scenario string `json:"scenario"`
		41	+	Run uint64 `json:"run"`
		42	+	Sequence uint64 `json:"sequence"`
		43	+	RequestID uint64 `json:"request_id"`
		44	+	Epoch uint64 `json:"epoch"`
		45	+	}
		46	+	if err := json.NewDecoder(bufio.NewReader(f)).Decode(&mark); err != nil {
		47	+	t.Fatal(err)
		48	+	}
		49	+	if mark.ProcessID != "idle-local-r001-daemon-01" \|\| mark.Scenario != "1x4-idle-local" \|\| mark.Run != 1 \|\| mark.Sequence == 0 \|\| mark.RequestID == 0 \|\| mark.Epoch == 0 {
		50	+	t.Fatalf("mark does not match harness mapping: %+v", mark)
		51	+	}
		52	+	}
		53	+
		54	+	type traceTestClock struct{}
		55	+
		56	+	func (*traceTestClock) Now() time.Time { return time.Unix(0, 1) }
		57	+	func (*traceTestClock) NewTimer(time.Duration) ports.Timer { return nil }
		58	+
9	func TestPerformanceTraceAppWiresOneObserverPerProcess(t *testing.T) {	59		func TestPerformanceTraceAppWiresOneObserverPerProcess(t *testing.T) {
10	source, err := os.ReadFile("run.go")	60		source, err := os.ReadFile("run.go")
11	if err != nil {	61		if err != nil {

M internal/ports/runtime_observer.go +45 -1

@@ -1,6 +1,9 @@
1	1		package ports
2	2
3		-	import "sync/atomic"
	3	+	import (
	4	+	"errors"
	5	+	"sync/atomic"
	6	+	)
4	7
5	8		// RuntimeMarkSchema is the version of the process-local performance trace.
6	9		const RuntimeMarkSchema uint16 = 1

@@ -84,6 +87,47 @@ type RuntimeCorrelation struct {
84	87		Run, Sequence, RequestID, Epoch uint64
85	88		}
86	89
	90	+	// RuntimeCorrelationInputs are supplied once by process orchestration. They
	91	+	// are deliberately data-only: consumer APIs continue to receive only their
	92	+	// one RuntimeObserver argument and never a clock or tick.
	93	+	type RuntimeCorrelationInputs struct {
	94	+	Scenario string
	95	+	Run uint64
	96	+	}
	97	+
	98	+	// NewRuntimeCorrelationObserver binds every mark from one OS process to its
	99	+	// harness manifest scenario/run. Operation IDs remain producer-owned when
	100	+	// supplied; missing IDs are assigned from this process-local sequence.
	101	+	func NewRuntimeCorrelationObserver(observer RuntimeObserver, inputs RuntimeCorrelationInputs) (RuntimeObserver, error) {
	102	+	if observer == nil \|\| inputs.Scenario == "" \|\| inputs.Run == 0 {
	103	+	return nil, errors.New("invalid runtime correlation inputs")
	104	+	}
	105	+	return &runtimeCorrelationObserver{observer: observer, inputs: inputs}, nil
	106	+	}
	107	+
	108	+	type runtimeCorrelationObserver struct {
	109	+	observer RuntimeObserver
	110	+	inputs RuntimeCorrelationInputs
	111	+	sequence atomic.Uint64
	112	+	}
	113	+
	114	+	func (o *runtimeCorrelationObserver) ObserveRuntime(mark RuntimeMark) {
	115	+	if o == nil \|\| o.observer == nil {
	116	+	return
	117	+	}
	118	+	mark.Scenario, mark.Run = o.inputs.Scenario, o.inputs.Run
	119	+	if mark.Sequence == 0 {
	120	+	mark.Sequence = o.sequence.Add(1)
	121	+	}
	122	+	if mark.RequestID == 0 {
	123	+	mark.RequestID = mark.Sequence
	124	+	}
	125	+	if mark.Epoch == 0 {
	126	+	mark.Epoch = mark.RequestID
	127	+	}
	128	+	o.observer.ObserveRuntime(mark)
	129	+	}
	130	+
87	131		var runtimeMarkSequence atomic.Uint64
88	132
89	133		// NewRuntimeCorrelation makes a unique, process-local operation identity.

@@ -1,6 +1,9 @@
1		package ports	1		package ports
2			2
3	-	import "sync/atomic"	3	+	import (
			4	+	"errors"
			5	+	"sync/atomic"
			6	+	)
4			7
5		// RuntimeMarkSchema is the version of the process-local performance trace.	8		// RuntimeMarkSchema is the version of the process-local performance trace.
6		const RuntimeMarkSchema uint16 = 1	9		const RuntimeMarkSchema uint16 = 1

@@ -84,6 +87,47 @@ type RuntimeCorrelation struct {
84		Run, Sequence, RequestID, Epoch uint64	87		Run, Sequence, RequestID, Epoch uint64
85		}	88		}
86			89
			90	+	// RuntimeCorrelationInputs are supplied once by process orchestration. They
			91	+	// are deliberately data-only: consumer APIs continue to receive only their
			92	+	// one RuntimeObserver argument and never a clock or tick.
			93	+	type RuntimeCorrelationInputs struct {
			94	+	Scenario string
			95	+	Run uint64
			96	+	}
			97	+
			98	+	// NewRuntimeCorrelationObserver binds every mark from one OS process to its
			99	+	// harness manifest scenario/run. Operation IDs remain producer-owned when
			100	+	// supplied; missing IDs are assigned from this process-local sequence.
			101	+	func NewRuntimeCorrelationObserver(observer RuntimeObserver, inputs RuntimeCorrelationInputs) (RuntimeObserver, error) {
			102	+	if observer == nil \|\| inputs.Scenario == "" \|\| inputs.Run == 0 {
			103	+	return nil, errors.New("invalid runtime correlation inputs")
			104	+	}
			105	+	return &runtimeCorrelationObserver{observer: observer, inputs: inputs}, nil
			106	+	}
			107	+
			108	+	type runtimeCorrelationObserver struct {
			109	+	observer RuntimeObserver
			110	+	inputs RuntimeCorrelationInputs
			111	+	sequence atomic.Uint64
			112	+	}
			113	+
			114	+	func (o *runtimeCorrelationObserver) ObserveRuntime(mark RuntimeMark) {
			115	+	if o == nil \|\| o.observer == nil {
			116	+	return
			117	+	}
			118	+	mark.Scenario, mark.Run = o.inputs.Scenario, o.inputs.Run
			119	+	if mark.Sequence == 0 {
			120	+	mark.Sequence = o.sequence.Add(1)
			121	+	}
			122	+	if mark.RequestID == 0 {
			123	+	mark.RequestID = mark.Sequence
			124	+	}
			125	+	if mark.Epoch == 0 {
			126	+	mark.Epoch = mark.RequestID
			127	+	}
			128	+	o.observer.ObserveRuntime(mark)
			129	+	}
			130	+
87		var runtimeMarkSequence atomic.Uint64	131		var runtimeMarkSequence atomic.Uint64
88			132
89		// NewRuntimeCorrelation makes a unique, process-local operation identity.	133		// NewRuntimeCorrelation makes a unique, process-local operation identity.

M internal/ports/runtime_observer_test.go +36 -0


@@ -60,6 +60,42 @@ func TestRuntimeObserverRequiredSpanKinds(t *testing.T) {
60	60		}
61	61		}
62	62
	63	+	func TestRuntimeCorrelationObserverUsesHarnessProcessInputs(t *testing.T) {
	64	+	var got []RuntimeMark
	65	+	observer, err := NewRuntimeCorrelationObserver(runtimeObserverFunc(func(mark RuntimeMark) {
	66	+	got = append(got, mark)
	67	+	}), RuntimeCorrelationInputs{Scenario: "1x4-idle-local", Run: 7})
	68	+	if err != nil {
	69	+	t.Fatal(err)
	70	+	}
	71	+
	72	+	observer.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "daemon", Scenario: "runtime", Run: 1, Kind: RuntimeEmitEnd, Valid: true})
	73	+	observer.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "daemon", Scenario: "runtime", Run: 1, Sequence: 9, RequestID: 10, Epoch: 11, Kind: RuntimeEmitStart, Valid: true})
	74	+
	75	+	if len(got) != 2 {
	76	+	t.Fatalf("marks=%+v", got)
	77	+	}
	78	+	for _, mark := range got {
	79	+	if mark.Scenario != "1x4-idle-local" \|\| mark.Run != 7 {
	80	+	t.Fatalf("harness correlation was not applied: %+v", mark)
	81	+	}
	82	+	if mark.Sequence == 0 \|\| mark.RequestID == 0 \|\| mark.Epoch == 0 {
	83	+	t.Fatalf("missing deterministic operation identity: %+v", mark)
	84	+	}
	85	+	}
	86	+	if got[1].Sequence != 9 \|\| got[1].RequestID != 10 \|\| got[1].Epoch != 11 {
	87	+	t.Fatalf("explicit operation identity changed: %+v", got[1])
	88	+	}
	89	+	}
	90	+
	91	+	func TestRuntimeCorrelationObserverRejectsInvalidHarnessInputs(t *testing.T) {
	92	+	for _, in := range []RuntimeCorrelationInputs{{}, {Scenario: "scenario"}} {
	93	+	if _, err := NewRuntimeCorrelationObserver(runtimeObserverFunc(func(RuntimeMark) {}), in); err == nil {
	94	+	t.Fatalf("invalid inputs accepted: %+v", in)
	95	+	}
	96	+	}
	97	+	}
	98	+
63	99		func TestRuntimeMarkHasNoRenderingPayloadOrPolicyFields(t *testing.T) {
64	100		typ := reflect.TypeFor[RuntimeMark]()
65	101		for _, forbidden := range []string{"Cell", "Cells", "Payload", "Frame", "Renderer", "Policy", "Transport"} {


@@ -60,6 +60,42 @@ func TestRuntimeObserverRequiredSpanKinds(t *testing.T) {
60	}	60		}
61	}	61		}
62		62
		63	+	func TestRuntimeCorrelationObserverUsesHarnessProcessInputs(t *testing.T) {
		64	+	var got []RuntimeMark
		65	+	observer, err := NewRuntimeCorrelationObserver(runtimeObserverFunc(func(mark RuntimeMark) {
		66	+	got = append(got, mark)
		67	+	}), RuntimeCorrelationInputs{Scenario: "1x4-idle-local", Run: 7})
		68	+	if err != nil {
		69	+	t.Fatal(err)
		70	+	}
		71	+
		72	+	observer.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "daemon", Scenario: "runtime", Run: 1, Kind: RuntimeEmitEnd, Valid: true})
		73	+	observer.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "daemon", Scenario: "runtime", Run: 1, Sequence: 9, RequestID: 10, Epoch: 11, Kind: RuntimeEmitStart, Valid: true})
		74	+
		75	+	if len(got) != 2 {
		76	+	t.Fatalf("marks=%+v", got)
		77	+	}
		78	+	for _, mark := range got {
		79	+	if mark.Scenario != "1x4-idle-local" \|\| mark.Run != 7 {
		80	+	t.Fatalf("harness correlation was not applied: %+v", mark)
		81	+	}
		82	+	if mark.Sequence == 0 \|\| mark.RequestID == 0 \|\| mark.Epoch == 0 {
		83	+	t.Fatalf("missing deterministic operation identity: %+v", mark)
		84	+	}
		85	+	}
		86	+	if got[1].Sequence != 9 \|\| got[1].RequestID != 10 \|\| got[1].Epoch != 11 {
		87	+	t.Fatalf("explicit operation identity changed: %+v", got[1])
		88	+	}
		89	+	}
		90	+
		91	+	func TestRuntimeCorrelationObserverRejectsInvalidHarnessInputs(t *testing.T) {
		92	+	for _, in := range []RuntimeCorrelationInputs{{}, {Scenario: "scenario"}} {
		93	+	if _, err := NewRuntimeCorrelationObserver(runtimeObserverFunc(func(RuntimeMark) {}), in); err == nil {
		94	+	t.Fatalf("invalid inputs accepted: %+v", in)
		95	+	}
		96	+	}
		97	+	}
		98	+
63	func TestRuntimeMarkHasNoRenderingPayloadOrPolicyFields(t *testing.T) {	99		func TestRuntimeMarkHasNoRenderingPayloadOrPolicyFields(t *testing.T) {
64	typ := reflect.TypeFor[RuntimeMark]()	100		typ := reflect.TypeFor[RuntimeMark]()
65	for _, forbidden := range []string{"Cell", "Cells", "Payload", "Frame", "Renderer", "Policy", "Transport"} {	101		for _, forbidden := range []string{"Cell", "Cells", "Payload", "Frame", "Renderer", "Policy", "Transport"} {