1b68dcf feat(obs): wire opt-in JSONL performance trace · vev-vt

M internal/adapters/dgram/transport.go +41 -6


@@ -45,6 +45,14 @@ var (
45	45		errControlFull = errors.New("dgram: control queue full")
46	46		)
47	47
	48	+	type RuntimeOption func(*Transport)
	49	+
	50	+	// WithRuntimeObserver enables process-local marks without changing dgram's
	51	+	// behavior clock or carriage policy.
	52	+	func WithRuntimeObserver(observer ports.RuntimeObserver) RuntimeOption {
	53	+	return func(t *Transport) { t.runtimeObserver = observer }
	54	+	}
	55	+
48	56		type Options struct {
49	57		MTU int
50	58		ResendAfter time.Duration

@@ -168,6 +176,7 @@ type Transport struct {
168	176		// mutation and before ordered event publication.
169	177		afterLinkStateCommit func(ports.LinkState)
170	178		observe DiagnosticObserver
	179	+	runtimeObserver ports.RuntimeObserver
171	180		diagnosticCh chan Diagnostic
172	181		heartbeat time.Duration
173	182		resendAfter time.Duration

@@ -311,11 +320,11 @@ func (s *writeDeadlineState) expire(now time.Time) {
311	320		s.mu.Unlock()
312	321		}
313	322
314		-	func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32) (*Transport, error) {
315		-	return NewTransportWithOptions(pc, peer, key, sendDir, recvDir, Options{})
	323	+	func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, runtimeOpts ...RuntimeOption) (*Transport, error) {
	324	+	return NewTransportWithOptions(pc, peer, key, sendDir, recvDir, Options{}, runtimeOpts...)
316	325		}
317	326
318		-	func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, opts Options) (*Transport, error) {
	327	+	func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, opts Options, runtimeOpts ...RuntimeOption) (*Transport, error) {
319	328		c, err := pdgram.NewCodec(key)
320	329		if err != nil {
321	330		return nil, err

@@ -360,6 +369,11 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
360	369		nextRecvSeq: 1,
361	370		recvBuf: make(map[uint64]ports.Frame),
362	371		}
	372	+	for _, opt := range runtimeOpts {
	373	+	if opt != nil {
	374	+	opt(t)
	375	+	}
	376	+	}
363	377		t.sendWake = make(chan struct{})
364	378		t.control = make(chan controlRecord, controlQueueSize)
365	379		t.ackWake = make(chan struct{}, 1)

@@ -387,9 +401,19 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
387	401		return t, nil
388	402		}
389	403
390		-	func (t *Transport) Send(f ports.Frame) error { return t.send(f, false) }
	404	+	func (t *Transport) Send(f ports.Frame) error {
	405	+	t.observeRuntime(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)), true)
	406	+	err := t.send(f, false)
	407	+	t.observeRuntime(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), err == nil)
	408	+	return err
	409	+	}
391	410
392		-	func (t *Transport) SendAsync(f ports.Frame) error { return t.send(f, true) }
	411	+	func (t *Transport) SendAsync(f ports.Frame) error {
	412	+	t.observeRuntime(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)), true)
	413	+	err := t.send(f, true)
	414	+	t.observeRuntime(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), err == nil)
	415	+	return err
	416	+	}
393	417
394	418		// SendSynchronous owns queue reservation, fragment pacing, per-write deadlines,
395	419		// and close cancellation. Callers must not layer a shorter preflight timer over it.

@@ -568,17 +592,22 @@ func (t *Transport) removePendingLocked(seq uint64) bool {
568	592		}
569	593
570	594		func (t *Transport) Recv() (ports.Frame, error) {
	595	+	t.observeRuntime(ports.RuntimeAdapterReceiveStart, 0, true)
571	596		select {
572	597		case f := <-t.in:
	598	+	t.observeRuntime(ports.RuntimeAdapterReceiveEnd, uint64(len(f.Payload)), true)
573	599		return f, nil
574	600		case <-t.done:
575	601		t.mu.Lock()
576	602		err := t.closeErr
577	603		t.mu.Unlock()
578	604		if err != nil {
	605	+	t.observeRuntime(ports.RuntimeAdapterReceiveEnd, 0, false)
579	606		return ports.Frame{}, err
580	607		}
581		-	return ports.Frame{}, errors.New("dgram: closed")
	608	+	err = errors.New("dgram: closed")
	609	+	t.observeRuntime(ports.RuntimeAdapterReceiveEnd, 0, false)
	610	+	return ports.Frame{}, err
582	611		}
583	612		}
584	613		func (t *Transport) Close() error {

@@ -1585,4 +1614,10 @@ func decodeData(b []byte) (uint64, bool, ports.Frame, bool) {
1585	1614		}
1586	1615
1587	1616		// DatagramTransport marks Transport as a UDP-style datagram transport.
	1617	+	func (t *Transport) observeRuntime(kind ports.RuntimeMarkKind, bytes uint64, valid bool) {
	1618	+	if t.runtimeObserver != nil {
	1619	+	t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMark("dgram", kind, bytes, valid))
	1620	+	}
	1621	+	}
	1622	+
1588	1623		func (*Transport) DatagramTransport() {}

 	errControlFull = errors.New("dgram: control queue full")
 )
+type RuntimeOption func(*Transport)
+// WithRuntimeObserver enables process-local marks without changing dgram's
+// behavior clock or carriage policy.
+func WithRuntimeObserver(observer ports.RuntimeObserver) RuntimeOption {
+	return func(t *Transport) { t.runtimeObserver = observer }
+}
 type Options struct {
 	MTU              int
 	ResendAfter      time.Duration
 	// mutation and before ordered event publication.
 	afterLinkStateCommit func(ports.LinkState)
 	observe              DiagnosticObserver
+	runtimeObserver      ports.RuntimeObserver
 	diagnosticCh         chan Diagnostic
 	heartbeat            time.Duration
 	resendAfter          time.Duration
 	s.mu.Unlock()
 }
-func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32) (*Transport, error) {
+func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, runtimeOpts ...RuntimeOption) (*Transport, error) {
-	return NewTransportWithOptions(pc, peer, key, sendDir, recvDir, Options{})
+	return NewTransportWithOptions(pc, peer, key, sendDir, recvDir, Options{}, runtimeOpts...)
 }
-func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, opts Options) (*Transport, error) {
+func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32, opts Options, runtimeOpts ...RuntimeOption) (*Transport, error) {
 	c, err := pdgram.NewCodec(key)
 	if err != nil {
 		return nil, err
 		nextRecvSeq:             1,
 		recvBuf:                 make(map[uint64]ports.Frame),
 	}
+	for _, opt := range runtimeOpts {
+		if opt != nil {
+			opt(t)
+		}
+	}
 	t.sendWake = make(chan struct{})
 	t.control = make(chan controlRecord, controlQueueSize)
 	t.ackWake = make(chan struct{}, 1)
 	return t, nil
 }
-func (t *Transport) Send(f ports.Frame) error { return t.send(f, false) }
+func (t *Transport) Send(f ports.Frame) error {
+	t.observeRuntime(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)), true)
+	err := t.send(f, false)
+	t.observeRuntime(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), err == nil)
+	return err
+}
-func (t *Transport) SendAsync(f ports.Frame) error { return t.send(f, true) }
+func (t *Transport) SendAsync(f ports.Frame) error {
+	t.observeRuntime(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)), true)
+	err := t.send(f, true)
+	t.observeRuntime(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), err == nil)
+	return err
+}
 // SendSynchronous owns queue reservation, fragment pacing, per-write deadlines,
 // and close cancellation. Callers must not layer a shorter preflight timer over it.
 }
 func (t *Transport) Recv() (ports.Frame, error) {
+	t.observeRuntime(ports.RuntimeAdapterReceiveStart, 0, true)
 	select {
 	case f := <-t.in:
+		t.observeRuntime(ports.RuntimeAdapterReceiveEnd, uint64(len(f.Payload)), true)
 		return f, nil
 	case <-t.done:
 		t.mu.Lock()
 		err := t.closeErr
 		t.mu.Unlock()
 		if err != nil {
+			t.observeRuntime(ports.RuntimeAdapterReceiveEnd, 0, false)
 			return ports.Frame{}, err
 		}
-		return ports.Frame{}, errors.New("dgram: closed")
+		err = errors.New("dgram: closed")
+		t.observeRuntime(ports.RuntimeAdapterReceiveEnd, 0, false)
+		return ports.Frame{}, err
 	}
 }
 func (t *Transport) Close() error {
 }
 // DatagramTransport marks Transport as a UDP-style datagram transport.
+func (t *Transport) observeRuntime(kind ports.RuntimeMarkKind, bytes uint64, valid bool) {
+	if t.runtimeObserver != nil {
+		t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMark("dgram", kind, bytes, valid))
+	}
+}
 func (*Transport) DatagramTransport() {}

M internal/adapters/ipc/transport.go +33 -3


@@ -35,8 +35,17 @@ var ErrFrameTooLarge = errors.New("ipc: frame exceeds maximum length")
35	35		// goroutine and a single writer goroutine); it is not safe for concurrent
36	36		// calls to Send from multiple goroutines, nor Recv from multiple
37	37		// goroutines.
	38	+	type Option func(*unixTransport)
	39	+
	40	+	// WithRuntimeObserver enables process-local transport marks. It accepts no
	41	+	// clock: timestamp ownership belongs to the configured observer.
	42	+	func WithRuntimeObserver(observer ports.RuntimeObserver) Option {
	43	+	return func(t *unixTransport) { t.observer = observer }
	44	+	}
	45	+
38	46		type unixTransport struct {
39		-	conn net.Conn
	47	+	conn net.Conn
	48	+	observer ports.RuntimeObserver
40	49
41	50		// readBuf is reused across Recv calls (grow-once strategy): it grows
42	51		// to fit the largest frame seen so far and is never shrunk. The Frame

@@ -47,16 +56,24 @@ type unixTransport struct {
47	56
48	57		// NewTransport wraps conn as a ports.Transport speaking vev's framed
49	58		// binary protocol.
50		-	func NewTransport(conn net.Conn) ports.Transport {
51		-	return &unixTransport{conn: conn}
	59	+	func NewTransport(conn net.Conn, opts ...Option) ports.Transport {
	60	+	t := &unixTransport{conn: conn}
	61	+	for _, opt := range opts {
	62	+	if opt != nil {
	63	+	opt(t)
	64	+	}
	65	+	}
	66	+	return t
52	67		}
53	68
54	69		// Send writes f as a single frame: a 4-byte big-endian length (covering
55	70		// the type byte and payload), the type byte, then the payload — assembled
56	71		// into one buffer and written with a single Write call.
57	72		func (t *unixTransport) Send(f ports.Frame) error {
	73	+	t.observe(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)), true)
58	74		n := 1 + len(f.Payload) // type + payload
59	75		if n > maxFrameLen {
	76	+	t.observe(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), false)
60	77		return ErrFrameTooLarge
61	78		}
62	79

@@ -66,6 +83,7 @@ func (t *unixTransport) Send(f ports.Frame) error {
66	83		copy(buf[frameHeaderLen+1:], f.Payload)
67	84
68	85		_, err := t.conn.Write(buf)
	86	+	t.observe(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), err == nil)
69	87		return err
70	88		}
71	89

@@ -73,16 +91,20 @@ func (t *unixTransport) Send(f ports.Frame) error {
73	91		// (type byte + payload). It blocks until a full frame arrives, the
74	92		// connection is closed (io.EOF), or an error occurs.
75	93		func (t *unixTransport) Recv() (ports.Frame, error) {
	94	+	t.observe(ports.RuntimeAdapterReceiveStart, 0, true)
76	95		var hdr [frameHeaderLen]byte
77	96		if _, err := io.ReadFull(t.conn, hdr[:]); err != nil {
	97	+	t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
78	98		return ports.Frame{}, err
79	99		}
80	100
81	101		n := binary.BigEndian.Uint32(hdr[:])
82	102		if n == 0 {
	103	+	t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
83	104		return ports.Frame{}, ErrZeroLengthFrame
84	105		}
85	106		if n > maxFrameLen {
	107	+	t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
86	108		return ports.Frame{}, ErrFrameTooLarge
87	109		}
88	110

@@ -92,6 +114,7 @@ func (t *unixTransport) Recv() (ports.Frame, error) {
92	114		t.readBuf = t.readBuf[:n]
93	115		}
94	116		if _, err := io.ReadFull(t.conn, t.readBuf); err != nil {
	117	+	t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
95	118		return ports.Frame{}, err
96	119		}
97	120

@@ -100,9 +123,16 @@ func (t *unixTransport) Recv() (ports.Frame, error) {
100	123		payload = make([]byte, n-1)
101	124		copy(payload, t.readBuf[1:])
102	125		}
	126	+	t.observe(ports.RuntimeAdapterReceiveEnd, uint64(len(payload)), true)
103	127		return ports.Frame{Type: ports.MsgType(t.readBuf[0]), Payload: payload}, nil
104	128		}
105	129
	130	+	func (t *unixTransport) observe(kind ports.RuntimeMarkKind, bytes uint64, valid bool) {
	131	+	if t.observer != nil {
	132	+	t.observer.ObserveRuntime(ports.NewRuntimeMark("ipc", kind, bytes, valid))
	133	+	}
	134	+	}
	135	+
106	136		// Close closes the underlying connection.
107	137		func (t *unixTransport) Close() error {
108	138		return t.conn.Close()

 // goroutine and a single writer goroutine); it is not safe for concurrent
 // calls to Send from multiple goroutines, nor Recv from multiple
 // goroutines.
+type Option func(*unixTransport)
+// WithRuntimeObserver enables process-local transport marks. It accepts no
+// clock: timestamp ownership belongs to the configured observer.
+func WithRuntimeObserver(observer ports.RuntimeObserver) Option {
+	return func(t *unixTransport) { t.observer = observer }
+}
 type unixTransport struct {
-	conn net.Conn
+	conn     net.Conn
+	observer ports.RuntimeObserver
 	// readBuf is reused across Recv calls (grow-once strategy): it grows
 	// to fit the largest frame seen so far and is never shrunk. The Frame
 // NewTransport wraps conn as a ports.Transport speaking vev's framed
 // binary protocol.
-func NewTransport(conn net.Conn) ports.Transport {
+func NewTransport(conn net.Conn, opts ...Option) ports.Transport {
-	return &unixTransport{conn: conn}
+	t := &unixTransport{conn: conn}
+	for _, opt := range opts {
+		if opt != nil {
+			opt(t)
+		}
+	}
+	return t
 }
 // Send writes f as a single frame: a 4-byte big-endian length (covering
 // the type byte and payload), the type byte, then the payload — assembled
 // into one buffer and written with a single Write call.
 func (t *unixTransport) Send(f ports.Frame) error {
+	t.observe(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)), true)
 	n := 1 + len(f.Payload) // type + payload
 	if n > maxFrameLen {
+		t.observe(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), false)
 		return ErrFrameTooLarge
 	}
 	copy(buf[frameHeaderLen+1:], f.Payload)
 	_, err := t.conn.Write(buf)
+	t.observe(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), err == nil)
 	return err
 }
 // (type byte + payload). It blocks until a full frame arrives, the
 // connection is closed (io.EOF), or an error occurs.
 func (t *unixTransport) Recv() (ports.Frame, error) {
+	t.observe(ports.RuntimeAdapterReceiveStart, 0, true)
 	var hdr [frameHeaderLen]byte
 	if _, err := io.ReadFull(t.conn, hdr[:]); err != nil {
+		t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
 		return ports.Frame{}, err
 	}
 	n := binary.BigEndian.Uint32(hdr[:])
 	if n == 0 {
+		t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
 		return ports.Frame{}, ErrZeroLengthFrame
 	}
 	if n > maxFrameLen {
+		t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
 		return ports.Frame{}, ErrFrameTooLarge
 	}
 		t.readBuf = t.readBuf[:n]
 	}
 	if _, err := io.ReadFull(t.conn, t.readBuf); err != nil {
+		t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
 		return ports.Frame{}, err
 	}
 		payload = make([]byte, n-1)
 		copy(payload, t.readBuf[1:])
 	}
+	t.observe(ports.RuntimeAdapterReceiveEnd, uint64(len(payload)), true)
 	return ports.Frame{Type: ports.MsgType(t.readBuf[0]), Payload: payload}, nil
 }
+func (t *unixTransport) observe(kind ports.RuntimeMarkKind, bytes uint64, valid bool) {
+	if t.observer != nil {
+		t.observer.ObserveRuntime(ports.NewRuntimeMark("ipc", kind, bytes, valid))
+	}
+}
 // Close closes the underlying connection.
 func (t *unixTransport) Close() error {
 	return t.conn.Close()

A internal/adapters/observability/jsonl.go +86 -0

@@ -0,0 +1,86 @@
	1	+	// Package observability contains concrete process-local trace sinks.
	2	+	package observability
	3	+
	4	+	import (
	5	+	"encoding/json"
	6	+	"io"
	7	+	"os"
	8	+	"sync"
	9	+
	10	+	"github.com/bnema/vev/internal/ports"
	11	+	)
	12	+
	13	+	// NewJSONL opens path as the sole timestamp owner for one OS process. Marks
	14	+	// carrying a caller timestamp are rejected rather than rewritten, preventing
	15	+	// accidental arithmetic across clock domains.
	16	+	func NewJSONL(path string, clock ports.Clock, processID string) (ports.RuntimeObserver, io.Closer, error) {
	17	+	if path == "" \|\| clock == nil \|\| processID == "" {
	18	+	return nil, nil, os.ErrInvalid
	19	+	}
	20	+	f, err := os.OpenFile(path, os.O_WRONLY\|os.O_CREATE\|os.O_APPEND, 0o600)
	21	+	if err != nil {
	22	+	return nil, nil, err
	23	+	}
	24	+	j := &jsonl{file: f, clock: clock, processID: processID}
	25	+	return j, j, nil
	26	+	}
	27	+
	28	+	type jsonl struct {
	29	+	mu sync.Mutex
	30	+	file *os.File
	31	+	clock ports.Clock
	32	+	processID string
	33	+	closed bool
	34	+	}
	35	+
	36	+	type record struct {
	37	+	Schema uint16 `json:"schema"`
	38	+	ProcessID string `json:"process_id"`
	39	+	Component string `json:"component"`
	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	+	Kind ports.RuntimeMarkKind `json:"kind"`
	46	+	Tick int64 `json:"tick"`
	47	+	Bytes uint64 `json:"bytes"`
	48	+	Fragments uint64 `json:"fragments"`
	49	+	Retransmits uint64 `json:"retransmits"`
	50	+	Pending uint64 `json:"pending"`
	51	+	AckRTTNanos int64 `json:"ack_rtt_nanos"`
	52	+	Valid bool `json:"valid"`
	53	+	}
	54	+
	55	+	func (j *jsonl) ObserveRuntime(mark ports.RuntimeMark) {
	56	+	// Tick is a write-once value owned by this adapter. Reject it before any
	57	+	// mutation so callers cannot smuggle a timestamp from their own clock.
	58	+	if j == nil \|\| mark.Tick != 0 {
	59	+	return
	60	+	}
	61	+	mark.ProcessID = j.processID
	62	+	if !ports.RuntimeMarkValid(mark) {
	63	+	return
	64	+	}
	65	+	j.mu.Lock()
	66	+	defer j.mu.Unlock()
	67	+	if j.closed {
	68	+	return
	69	+	}
	70	+	r := record{Schema: mark.Schema, ProcessID: mark.ProcessID, Component: mark.Component, Scenario: mark.Scenario, Run: mark.Run, Sequence: mark.Sequence, RequestID: mark.RequestID, Epoch: mark.Epoch, Kind: mark.Kind, Tick: j.clock.Now().UnixNano(), Bytes: mark.Bytes, Fragments: mark.Fragments, Retransmits: mark.Retransmits, Pending: mark.Pending, AckRTTNanos: mark.AckRTTNanos, Valid: mark.Valid}
	71	+	encoded, err := json.Marshal(r)
	72	+	if err != nil {
	73	+	return
	74	+	}
	75	+	_, _ = j.file.Write(append(encoded, '\n'))
	76	+	}
	77	+
	78	+	func (j *jsonl) Close() error {
	79	+	j.mu.Lock()
	80	+	defer j.mu.Unlock()
	81	+	if j.closed {
	82	+	return nil
	83	+	}
	84	+	j.closed = true
	85	+	return j.file.Close()
	86	+	}

@@ -0,0 +1,86 @@
	1	+	// Package observability contains concrete process-local trace sinks.
	2	+	package observability
	3	+
	4	+	import (
	5	+	"encoding/json"
	6	+	"io"
	7	+	"os"
	8	+	"sync"
	9	+
	10	+	"github.com/bnema/vev/internal/ports"
	11	+	)
	12	+
	13	+	// NewJSONL opens path as the sole timestamp owner for one OS process. Marks
	14	+	// carrying a caller timestamp are rejected rather than rewritten, preventing
	15	+	// accidental arithmetic across clock domains.
	16	+	func NewJSONL(path string, clock ports.Clock, processID string) (ports.RuntimeObserver, io.Closer, error) {
	17	+	if path == "" \|\| clock == nil \|\| processID == "" {
	18	+	return nil, nil, os.ErrInvalid
	19	+	}
	20	+	f, err := os.OpenFile(path, os.O_WRONLY\|os.O_CREATE\|os.O_APPEND, 0o600)
	21	+	if err != nil {
	22	+	return nil, nil, err
	23	+	}
	24	+	j := &jsonl{file: f, clock: clock, processID: processID}
	25	+	return j, j, nil
	26	+	}
	27	+
	28	+	type jsonl struct {
	29	+	mu sync.Mutex
	30	+	file *os.File
	31	+	clock ports.Clock
	32	+	processID string
	33	+	closed bool
	34	+	}
	35	+
	36	+	type record struct {
	37	+	Schema uint16 `json:"schema"`
	38	+	ProcessID string `json:"process_id"`
	39	+	Component string `json:"component"`
	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	+	Kind ports.RuntimeMarkKind `json:"kind"`
	46	+	Tick int64 `json:"tick"`
	47	+	Bytes uint64 `json:"bytes"`
	48	+	Fragments uint64 `json:"fragments"`
	49	+	Retransmits uint64 `json:"retransmits"`
	50	+	Pending uint64 `json:"pending"`
	51	+	AckRTTNanos int64 `json:"ack_rtt_nanos"`
	52	+	Valid bool `json:"valid"`
	53	+	}
	54	+
	55	+	func (j *jsonl) ObserveRuntime(mark ports.RuntimeMark) {
	56	+	// Tick is a write-once value owned by this adapter. Reject it before any
	57	+	// mutation so callers cannot smuggle a timestamp from their own clock.
	58	+	if j == nil \|\| mark.Tick != 0 {
	59	+	return
	60	+	}
	61	+	mark.ProcessID = j.processID
	62	+	if !ports.RuntimeMarkValid(mark) {
	63	+	return
	64	+	}
	65	+	j.mu.Lock()
	66	+	defer j.mu.Unlock()
	67	+	if j.closed {
	68	+	return
	69	+	}
	70	+	r := record{Schema: mark.Schema, ProcessID: mark.ProcessID, Component: mark.Component, Scenario: mark.Scenario, Run: mark.Run, Sequence: mark.Sequence, RequestID: mark.RequestID, Epoch: mark.Epoch, Kind: mark.Kind, Tick: j.clock.Now().UnixNano(), Bytes: mark.Bytes, Fragments: mark.Fragments, Retransmits: mark.Retransmits, Pending: mark.Pending, AckRTTNanos: mark.AckRTTNanos, Valid: mark.Valid}
	71	+	encoded, err := json.Marshal(r)
	72	+	if err != nil {
	73	+	return
	74	+	}
	75	+	_, _ = j.file.Write(append(encoded, '\n'))
	76	+	}
	77	+
	78	+	func (j *jsonl) Close() error {
	79	+	j.mu.Lock()
	80	+	defer j.mu.Unlock()
	81	+	if j.closed {
	82	+	return nil
	83	+	}
	84	+	j.closed = true
	85	+	return j.file.Close()
	86	+	}

M internal/adapters/sshstdio/transport.go +40 -8


@@ -34,9 +34,23 @@ var (
34	34		type closeFunc func() error
35	35		type eofErrFunc func() error
36	36
	37	+	type Option func(*transport)
	38	+
	39	+	// WithRuntimeObserver enables process-local transport marks; it deliberately
	40	+	// takes only an observer so a carriage adapter never owns trace time.
	41	+	func WithRuntimeObserver(observer ports.RuntimeObserver) Option {
	42	+	return func(t *transport) { t.observer = observer }
	43	+	}
	44	+
37	45		// NewTransport wraps separate reader/writer streams as a framed Transport.
38		-	func NewTransport(r io.Reader, w io.Writer, closeFn closeFunc) ports.Transport {
39		-	return newTransport(r, w, closeFn, nil)
	46	+	func NewTransport(r io.Reader, w io.Writer, closeFn closeFunc, opts ...Option) ports.Transport {
	47	+	t := newTransport(r, w, closeFn, nil).(*transport)
	48	+	for _, opt := range opts {
	49	+	if opt != nil {
	50	+	opt(t)
	51	+	}
	52	+	}
	53	+	return t
40	54		}
41	55
42	56		func newTransport(r io.Reader, w io.Writer, closeFn closeFunc, eofErr eofErrFunc) ports.Transport {

@@ -47,18 +61,21 @@ func newTransport(r io.Reader, w io.Writer, closeFn closeFunc, eofErr eofErrFunc
47	61		}
48	62
49	63		type transport struct {
50		-	r io.Reader
51		-	w io.Writer
52		-	close closeFunc
53		-	eofErr eofErrFunc
	64	+	r io.Reader
	65	+	w io.Writer
	66	+	close closeFunc
	67	+	eofErr eofErrFunc
	68	+	observer ports.RuntimeObserver
54	69
55	70		mu sync.Mutex
56	71		readBuf []byte
57	72		}
58	73
59	74		func (t *transport) Send(f ports.Frame) error {
	75	+	t.observe(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)), true)
60	76		n := 1 + len(f.Payload)
61	77		if n > maxFrameLen {
	78	+	t.observe(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), false)
62	79		return ErrFrameTooLarge
63	80		}
64	81

@@ -70,20 +87,26 @@ func (t *transport) Send(f ports.Frame) error {
70	87		t.mu.Lock()
71	88		defer t.mu.Unlock()
72	89		_, err := t.w.Write(buf)
	90	+	t.observe(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), err == nil)
73	91		return err
74	92		}
75	93
76	94		func (t *transport) Recv() (ports.Frame, error) {
	95	+	t.observe(ports.RuntimeAdapterReceiveStart, 0, true)
77	96		var hdr [frameHeaderLen]byte
78	97		if _, err := io.ReadFull(t.r, hdr[:]); err != nil {
79		-	return ports.Frame{}, t.mapEOFError(err)
	98	+	err = t.mapEOFError(err)
	99	+	t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
	100	+	return ports.Frame{}, err
80	101		}
81	102
82	103		n := binary.BigEndian.Uint32(hdr[:])
83	104		if n == 0 {
	105	+	t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
84	106		return ports.Frame{}, ErrZeroLengthFrame
85	107		}
86	108		if n > maxFrameLen {
	109	+	t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
87	110		return ports.Frame{}, ErrFrameTooLarge
88	111		}
89	112

@@ -93,13 +116,22 @@ func (t *transport) Recv() (ports.Frame, error) {
93	116		t.readBuf = t.readBuf[:n]
94	117		}
95	118		if _, err := io.ReadFull(t.r, t.readBuf); err != nil {
96		-	return ports.Frame{}, t.mapEOFError(err)
	119	+	err = t.mapEOFError(err)
	120	+	t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
	121	+	return ports.Frame{}, err
97	122		}
98	123
99	124		payload := append([]byte(nil), t.readBuf[1:]...)
	125	+	t.observe(ports.RuntimeAdapterReceiveEnd, uint64(len(payload)), true)
100	126		return ports.Frame{Type: ports.MsgType(t.readBuf[0]), Payload: payload}, nil
101	127		}
102	128
	129	+	func (t *transport) observe(kind ports.RuntimeMarkKind, bytes uint64, valid bool) {
	130	+	if t.observer != nil {
	131	+	t.observer.ObserveRuntime(ports.NewRuntimeMark("sshstdio", kind, bytes, valid))
	132	+	}
	133	+	}
	134	+
103	135		func (t *transport) mapEOFError(err error) error {
104	136		if t.eofErr == nil \|\| (!errors.Is(err, io.EOF) && !errors.Is(err, io.ErrUnexpectedEOF)) {
105	137		return err

 type closeFunc func() error
 type eofErrFunc func() error
+type Option func(*transport)
+// WithRuntimeObserver enables process-local transport marks; it deliberately
+// takes only an observer so a carriage adapter never owns trace time.
+func WithRuntimeObserver(observer ports.RuntimeObserver) Option {
+	return func(t *transport) { t.observer = observer }
+}
 // NewTransport wraps separate reader/writer streams as a framed Transport.
-func NewTransport(r io.Reader, w io.Writer, closeFn closeFunc) ports.Transport {
+func NewTransport(r io.Reader, w io.Writer, closeFn closeFunc, opts ...Option) ports.Transport {
-	return newTransport(r, w, closeFn, nil)
+	t := newTransport(r, w, closeFn, nil).(*transport)
+	for _, opt := range opts {
+		if opt != nil {
+			opt(t)
+		}
+	}
+	return t
 }
 func newTransport(r io.Reader, w io.Writer, closeFn closeFunc, eofErr eofErrFunc) ports.Transport {
 }
 type transport struct {
-	r      io.Reader
+	r        io.Reader
-	w      io.Writer
+	w        io.Writer
-	close  closeFunc
+	close    closeFunc
-	eofErr eofErrFunc
+	eofErr   eofErrFunc
+	observer ports.RuntimeObserver
 	mu      sync.Mutex
 	readBuf []byte
 }
 func (t *transport) Send(f ports.Frame) error {
+	t.observe(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)), true)
 	n := 1 + len(f.Payload)
 	if n > maxFrameLen {
+		t.observe(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), false)
 		return ErrFrameTooLarge
 	}
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	_, err := t.w.Write(buf)
+	t.observe(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), err == nil)
 	return err
 }
 func (t *transport) Recv() (ports.Frame, error) {
+	t.observe(ports.RuntimeAdapterReceiveStart, 0, true)
 	var hdr [frameHeaderLen]byte
 	if _, err := io.ReadFull(t.r, hdr[:]); err != nil {
-		return ports.Frame{}, t.mapEOFError(err)
+		err = t.mapEOFError(err)
+		t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
+		return ports.Frame{}, err
 	}
 	n := binary.BigEndian.Uint32(hdr[:])
 	if n == 0 {
+		t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
 		return ports.Frame{}, ErrZeroLengthFrame
 	}
 	if n > maxFrameLen {
+		t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
 		return ports.Frame{}, ErrFrameTooLarge
 	}
 		t.readBuf = t.readBuf[:n]
 	}
 	if _, err := io.ReadFull(t.r, t.readBuf); err != nil {
-		return ports.Frame{}, t.mapEOFError(err)
+		err = t.mapEOFError(err)
+		t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
+		return ports.Frame{}, err
 	}
 	payload := append([]byte(nil), t.readBuf[1:]...)
+	t.observe(ports.RuntimeAdapterReceiveEnd, uint64(len(payload)), true)
 	return ports.Frame{Type: ports.MsgType(t.readBuf[0]), Payload: payload}, nil
 }
+func (t *transport) observe(kind ports.RuntimeMarkKind, bytes uint64, valid bool) {
+	if t.observer != nil {
+		t.observer.ObserveRuntime(ports.NewRuntimeMark("sshstdio", kind, bytes, valid))
+	}
+}
 func (t *transport) mapEOFError(err error) error {
 	if t.eofErr == nil || (!errors.Is(err, io.EOF) && !errors.Is(err, io.ErrUnexpectedEOF)) {
 		return err

M internal/app/run.go +53 -6


@@ -31,6 +31,7 @@ import (
31	31		"github.com/bnema/vev/internal/adapters/config"
32	32		"github.com/bnema/vev/internal/adapters/dgram"
33	33		"github.com/bnema/vev/internal/adapters/ipc"
	34	+	"github.com/bnema/vev/internal/adapters/observability"
34	35		"github.com/bnema/vev/internal/adapters/pty"
35	36		remoteadapter "github.com/bnema/vev/internal/adapters/remote"
36	37		"github.com/bnema/vev/internal/adapters/shellcmd"

@@ -238,6 +239,17 @@ func dispatch(ctx context.Context, cmd command) error {
238	239		}
239	240		}
240	241
	242	+	// performanceTrace creates one serialized timestamp owner for this process.
	243	+	// An empty trace environment leaves all production behavior and wire bytes
	244	+	// unchanged.
	245	+	func performanceTrace(clk ports.Clock) (ports.RuntimeObserver, io.Closer, error) {
	246	+	path, processID := os.Getenv("VEV_PERF_TRACE"), os.Getenv("VEV_PERF_PROCESS_ID")
	247	+	if path == "" {
	248	+	return nil, nil, nil
	249	+	}
	250	+	return observability.NewJSONL(path, clk, processID)
	251	+	}
	252	+
241	253		func configureLogging(component logging.Component, rotateAtRuntime bool) (*slog.Logger, io.Closer, error) {
242	254		return logging.Setup(logging.Config{
243	255		Dir: platform.StateDir(),

@@ -308,7 +320,18 @@ func runDaemon() error {
308	320		}
309	321
310	322		log.Info("daemon starting", "socket", ln.Addr())
	323	+	clk := clock.New()
	324	+	observer, observerCloser, err := performanceTrace(clk)
	325	+	if err != nil {
	326	+	return fmt.Errorf("vev: performance trace: %w", err)
	327	+	}
	328	+	if observerCloser != nil {
	329	+	defer func() { _ = observerCloser.Close() }()
	330	+	}
311	331		daemonOpts := []daemon.Option(nil)
	332	+	if observer != nil {
	333	+	daemonOpts = append(daemonOpts, daemon.WithRuntimeObserver(observer))
	334	+	}
312	335		configPath := platform.ConfigPath()
313	336		cfg, warnings, err := config.Load(configPath)
314	337		if err != nil {

@@ -327,7 +350,6 @@ func runDaemon() error {
327	350		log.Info("session persistence enabled", "path", storePath)
328	351		daemonOpts = append(daemonOpts, daemon.WithStore(store))
329	352		}
330		-	clk := clock.New()
331	353		d := daemon.New(pty.NewFactory(), clk, log, daemonOpts...)
332	354		watchCtx, stopWatch := context.WithCancel(ctx)
333	355		defer stopWatch()

@@ -357,13 +379,23 @@ func runAttach(ctx context.Context, intent uint8, name, remoteTarget string) err
357	379		}
358	380		defer func() { _ = logCloser.Close() }()
359	381
	382	+	clk := clock.New()
	383	+	observer, observerCloser, err := performanceTrace(clk)
	384	+	if err != nil {
	385	+	return fmt.Errorf("vev: performance trace: %w", err)
	386	+	}
	387	+	if observerCloser != nil {
	388	+	defer func() { _ = observerCloser.Close() }()
	389	+	}
360	390		return runAttachWithDeps(ctx, intent, name, remoteTarget, os.Getenv("VEV"), log, runAttachDeps{
361	391		localDialer: defaultLocalDialer,
362	392		remoteDialerFactory: defaultRemoteDialerFactory(),
363	393		selectedRemoteTransport: os.Getenv(envRemoteTransport),
364		-	runClient: client.Run,
365		-	createDetached: createDetachedLocalSession,
366		-	clipboard: clipboard.New(),
	394	+	runClient: func(ctx context.Context, dialer ports.Dialer, terminal ports.Terminal, behaviorClock ports.Clock, intent uint8, name string, remote bool, clipboard ports.ClipboardReader, log *slog.Logger) error {
	395	+	return client.RunWithRuntimeObserver(ctx, dialer, terminal, behaviorClock, intent, name, remote, clipboard, log, observer)
	396	+	},
	397	+	createDetached: createDetachedLocalSession,
	398	+	clipboard: clipboard.New(),
367	399		})
368	400		}
369	401

@@ -635,7 +667,15 @@ func runStdio(ctx context.Context) error {
635	667		}
636	668		defer func() { _ = transport.Close() }()
637	669
638		-	stdio := sshstdio.NewTransport(os.Stdin, os.Stdout, nil)
	670	+	clk := clock.New()
	671	+	observer, observerCloser, err := performanceTrace(clk)
	672	+	if err != nil {
	673	+	return fmt.Errorf("vev: performance trace: %w", err)
	674	+	}
	675	+	if observerCloser != nil {
	676	+	defer func() { _ = observerCloser.Close() }()
	677	+	}
	678	+	stdio := sshstdio.NewTransport(os.Stdin, os.Stdout, nil, sshstdio.WithRuntimeObserver(observer))
639	679		return proxyTransports(ctx, stdio, transport, log)
640	680		}
641	681

@@ -747,7 +787,14 @@ func runUDPProxy(ctx context.Context, session string, ready io.Writer) error {
747	787		defer func() { _ = daemonTr.Close() }()
748	788		udpOptions := udpProxyClientTransportOptions
749	789		udpOptions.Observe = dgram.DiagnosticLogObserver(log)
750		-	dg, err := dgram.NewTransportWithOptions(conn, nil, key, 2, 1, udpOptions)
	790	+	observer, observerCloser, err := performanceTrace(clock.New())
	791	+	if err != nil {
	792	+	return fmt.Errorf("vev: performance trace: %w", err)
	793	+	}
	794	+	if observerCloser != nil {
	795	+	defer func() { _ = observerCloser.Close() }()
	796	+	}
	797	+	dg, err := dgram.NewTransportWithOptions(conn, nil, key, 2, 1, udpOptions, dgram.WithRuntimeObserver(observer))
751	798		if err != nil {
752	799		return err
753	800		}

 	"github.com/bnema/vev/internal/adapters/config"
 	"github.com/bnema/vev/internal/adapters/dgram"
 	"github.com/bnema/vev/internal/adapters/ipc"
+	"github.com/bnema/vev/internal/adapters/observability"
 	"github.com/bnema/vev/internal/adapters/pty"
 	remoteadapter "github.com/bnema/vev/internal/adapters/remote"
 	"github.com/bnema/vev/internal/adapters/shellcmd"
 	}
 }
+// performanceTrace creates one serialized timestamp owner for this process.
+// An empty trace environment leaves all production behavior and wire bytes
+// unchanged.
+func performanceTrace(clk ports.Clock) (ports.RuntimeObserver, io.Closer, error) {
+	path, processID := os.Getenv("VEV_PERF_TRACE"), os.Getenv("VEV_PERF_PROCESS_ID")
+	if path == "" {
+		return nil, nil, nil
+	}
+	return observability.NewJSONL(path, clk, processID)
+}
 func configureLogging(component logging.Component, rotateAtRuntime bool) (*slog.Logger, io.Closer, error) {
 	return logging.Setup(logging.Config{
 		Dir:             platform.StateDir(),
 	}
 	log.Info("daemon starting", "socket", ln.Addr())
+	clk := clock.New()
+	observer, observerCloser, err := performanceTrace(clk)
+	if err != nil {
+		return fmt.Errorf("vev: performance trace: %w", err)
+	}
+	if observerCloser != nil {
+		defer func() { _ = observerCloser.Close() }()
+	}
 	daemonOpts := []daemon.Option(nil)
+	if observer != nil {
+		daemonOpts = append(daemonOpts, daemon.WithRuntimeObserver(observer))
+	}
 	configPath := platform.ConfigPath()
 	cfg, warnings, err := config.Load(configPath)
 	if err != nil {
 		log.Info("session persistence enabled", "path", storePath)
 		daemonOpts = append(daemonOpts, daemon.WithStore(store))
 	}
-	clk := clock.New()
 	d := daemon.New(pty.NewFactory(), clk, log, daemonOpts...)
 	watchCtx, stopWatch := context.WithCancel(ctx)
 	defer stopWatch()
 	}
 	defer func() { _ = logCloser.Close() }()
+	clk := clock.New()
+	observer, observerCloser, err := performanceTrace(clk)
+	if err != nil {
+		return fmt.Errorf("vev: performance trace: %w", err)
+	}
+	if observerCloser != nil {
+		defer func() { _ = observerCloser.Close() }()
+	}
 	return runAttachWithDeps(ctx, intent, name, remoteTarget, os.Getenv("VEV"), log, runAttachDeps{
 		localDialer:             defaultLocalDialer,
 		remoteDialerFactory:     defaultRemoteDialerFactory(),
 		selectedRemoteTransport: os.Getenv(envRemoteTransport),
-		runClient:               client.Run,
+		runClient: func(ctx context.Context, dialer ports.Dialer, terminal ports.Terminal, behaviorClock ports.Clock, intent uint8, name string, remote bool, clipboard ports.ClipboardReader, log *slog.Logger) error {
-		createDetached:          createDetachedLocalSession,
+			return client.RunWithRuntimeObserver(ctx, dialer, terminal, behaviorClock, intent, name, remote, clipboard, log, observer)
-		clipboard:               clipboard.New(),
+		},
+		createDetached: createDetachedLocalSession,
+		clipboard:      clipboard.New(),
 	})
 }
 	}
 	defer func() { _ = transport.Close() }()
-	stdio := sshstdio.NewTransport(os.Stdin, os.Stdout, nil)
+	clk := clock.New()
+	observer, observerCloser, err := performanceTrace(clk)
+	if err != nil {
+		return fmt.Errorf("vev: performance trace: %w", err)
+	}
+	if observerCloser != nil {
+		defer func() { _ = observerCloser.Close() }()
+	}
+	stdio := sshstdio.NewTransport(os.Stdin, os.Stdout, nil, sshstdio.WithRuntimeObserver(observer))
 	return proxyTransports(ctx, stdio, transport, log)
 }
 	defer func() { _ = daemonTr.Close() }()
 	udpOptions := udpProxyClientTransportOptions
 	udpOptions.Observe = dgram.DiagnosticLogObserver(log)
-	dg, err := dgram.NewTransportWithOptions(conn, nil, key, 2, 1, udpOptions)
+	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
 	}

A internal/ports/runtime_observer.go +82 -0

@@ -0,0 +1,82 @@
	1	+	package ports
	2	+
	3	+	// RuntimeMarkSchema is the version of the process-local performance trace.
	4	+	const RuntimeMarkSchema uint16 = 1
	5	+
	6	+	// RuntimeMarkKind identifies a process-local runtime boundary. It deliberately
	7	+	// contains no protocol payload, cells, or rendering policy.
	8	+	type RuntimeMarkKind string
	9	+
	10	+	const (
	11	+	RuntimeInputInjected RuntimeMarkKind = "input_injected"
	12	+	RuntimeResizeRequested RuntimeMarkKind = "resize_requested"
	13	+	RuntimeResizeCommitted RuntimeMarkKind = "resize_committed"
	14	+	RuntimeCaptureStart RuntimeMarkKind = "capture_start"
	15	+	RuntimeCaptureEnd RuntimeMarkKind = "capture_end"
	16	+	RuntimeComposeStart RuntimeMarkKind = "compose_start"
	17	+	RuntimeComposeEnd RuntimeMarkKind = "compose_end"
	18	+	RuntimeDiffStart RuntimeMarkKind = "diff_start"
	19	+	RuntimeDiffEnd RuntimeMarkKind = "diff_end"
	20	+	RuntimeQueueEnqueued RuntimeMarkKind = "queue_enqueued"
	21	+	RuntimeQueueDequeued RuntimeMarkKind = "queue_dequeued"
	22	+	RuntimeACKBlockedStart RuntimeMarkKind = "ack_blocked_start"
	23	+	RuntimeACKBlockedEnd RuntimeMarkKind = "ack_blocked_end"
	24	+	RuntimeEmitStart RuntimeMarkKind = "emit_start"
	25	+	RuntimeEmitEnd RuntimeMarkKind = "emit_end"
	26	+	RuntimeAdapterSendStart RuntimeMarkKind = "adapter_send_start"
	27	+	RuntimeAdapterSendEnd RuntimeMarkKind = "adapter_send_end"
	28	+	RuntimeAdapterReceiveStart RuntimeMarkKind = "adapter_receive_start"
	29	+	RuntimeAdapterReceiveEnd RuntimeMarkKind = "adapter_receive_end"
	30	+	RuntimeTerminalFlushed RuntimeMarkKind = "terminal_flushed"
	31	+	RuntimeTransportDiagnostic RuntimeMarkKind = "transport_diagnostic"
	32	+	)
	33	+
	34	+	// RuntimeMark is a process-local observation record. Tick is intentionally
	35	+	// left zero by producers: the concrete process observer owns timestamping.
	36	+	type RuntimeMark struct {
	37	+	Schema uint16
	38	+	ProcessID string
	39	+	Component string
	40	+	Scenario string
	41	+	Run uint64
	42	+	Sequence uint64
	43	+	RequestID uint64
	44	+	Epoch uint64
	45	+	Kind RuntimeMarkKind
	46	+	Tick int64
	47	+	Bytes uint64
	48	+	Fragments uint64
	49	+	Retransmits uint64
	50	+	Pending uint64
	51	+	AckRTTNanos int64
	52	+	Valid bool
	53	+	}
	54	+
	55	+	// RuntimeObserver receives process-local marks. Implementations must not let
	56	+	// a producer provide a timestamp in another clock domain.
	57	+	type RuntimeObserver interface{ ObserveRuntime(RuntimeMark) }
	58	+
	59	+	// RuntimeMarkValid reports whether a mark has all correlation fields required
	60	+	// by the JSONL schema. Tick is checked separately by the timestamp owner.
	61	+	func RuntimeMarkValid(mark RuntimeMark) bool {
	62	+	if mark.Schema != RuntimeMarkSchema \|\| mark.Component == "" \|\| mark.Scenario == "" \|\| mark.Run == 0 \|\| mark.Sequence == 0 \|\| mark.RequestID == 0 \|\| mark.Epoch == 0 {
	63	+	return false
	64	+	}
	65	+	switch mark.Kind {
	66	+	case RuntimeInputInjected, RuntimeResizeRequested, RuntimeResizeCommitted,
	67	+	RuntimeCaptureStart, RuntimeCaptureEnd, RuntimeComposeStart, RuntimeComposeEnd,
	68	+	RuntimeDiffStart, RuntimeDiffEnd, RuntimeQueueEnqueued, RuntimeQueueDequeued,
	69	+	RuntimeACKBlockedStart, RuntimeACKBlockedEnd, RuntimeEmitStart, RuntimeEmitEnd,
	70	+	RuntimeAdapterSendStart, RuntimeAdapterSendEnd, RuntimeAdapterReceiveStart,
	71	+	RuntimeAdapterReceiveEnd, RuntimeTerminalFlushed, RuntimeTransportDiagnostic:
	72	+	return true
	73	+	default:
	74	+	return false
	75	+	}
	76	+	}
	77	+
	78	+	// NewRuntimeMark returns a timestamp-free mark for lightweight opt-in hooks.
	79	+	// Callers that need a span pass the same correlation fields to both endpoints.
	80	+	func NewRuntimeMark(component string, kind RuntimeMarkKind, bytes uint64, valid bool) RuntimeMark {
	81	+	return RuntimeMark{Schema: RuntimeMarkSchema, Component: component, Scenario: "runtime", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: kind, Bytes: bytes, Valid: valid}
	82	+	}

@@ -0,0 +1,82 @@
	1	+	package ports
	2	+
	3	+	// RuntimeMarkSchema is the version of the process-local performance trace.
	4	+	const RuntimeMarkSchema uint16 = 1
	5	+
	6	+	// RuntimeMarkKind identifies a process-local runtime boundary. It deliberately
	7	+	// contains no protocol payload, cells, or rendering policy.
	8	+	type RuntimeMarkKind string
	9	+
	10	+	const (
	11	+	RuntimeInputInjected RuntimeMarkKind = "input_injected"
	12	+	RuntimeResizeRequested RuntimeMarkKind = "resize_requested"
	13	+	RuntimeResizeCommitted RuntimeMarkKind = "resize_committed"
	14	+	RuntimeCaptureStart RuntimeMarkKind = "capture_start"
	15	+	RuntimeCaptureEnd RuntimeMarkKind = "capture_end"
	16	+	RuntimeComposeStart RuntimeMarkKind = "compose_start"
	17	+	RuntimeComposeEnd RuntimeMarkKind = "compose_end"
	18	+	RuntimeDiffStart RuntimeMarkKind = "diff_start"
	19	+	RuntimeDiffEnd RuntimeMarkKind = "diff_end"
	20	+	RuntimeQueueEnqueued RuntimeMarkKind = "queue_enqueued"
	21	+	RuntimeQueueDequeued RuntimeMarkKind = "queue_dequeued"
	22	+	RuntimeACKBlockedStart RuntimeMarkKind = "ack_blocked_start"
	23	+	RuntimeACKBlockedEnd RuntimeMarkKind = "ack_blocked_end"
	24	+	RuntimeEmitStart RuntimeMarkKind = "emit_start"
	25	+	RuntimeEmitEnd RuntimeMarkKind = "emit_end"
	26	+	RuntimeAdapterSendStart RuntimeMarkKind = "adapter_send_start"
	27	+	RuntimeAdapterSendEnd RuntimeMarkKind = "adapter_send_end"
	28	+	RuntimeAdapterReceiveStart RuntimeMarkKind = "adapter_receive_start"
	29	+	RuntimeAdapterReceiveEnd RuntimeMarkKind = "adapter_receive_end"
	30	+	RuntimeTerminalFlushed RuntimeMarkKind = "terminal_flushed"
	31	+	RuntimeTransportDiagnostic RuntimeMarkKind = "transport_diagnostic"
	32	+	)
	33	+
	34	+	// RuntimeMark is a process-local observation record. Tick is intentionally
	35	+	// left zero by producers: the concrete process observer owns timestamping.
	36	+	type RuntimeMark struct {
	37	+	Schema uint16
	38	+	ProcessID string
	39	+	Component string
	40	+	Scenario string
	41	+	Run uint64
	42	+	Sequence uint64
	43	+	RequestID uint64
	44	+	Epoch uint64
	45	+	Kind RuntimeMarkKind
	46	+	Tick int64
	47	+	Bytes uint64
	48	+	Fragments uint64
	49	+	Retransmits uint64
	50	+	Pending uint64
	51	+	AckRTTNanos int64
	52	+	Valid bool
	53	+	}
	54	+
	55	+	// RuntimeObserver receives process-local marks. Implementations must not let
	56	+	// a producer provide a timestamp in another clock domain.
	57	+	type RuntimeObserver interface{ ObserveRuntime(RuntimeMark) }
	58	+
	59	+	// RuntimeMarkValid reports whether a mark has all correlation fields required
	60	+	// by the JSONL schema. Tick is checked separately by the timestamp owner.
	61	+	func RuntimeMarkValid(mark RuntimeMark) bool {
	62	+	if mark.Schema != RuntimeMarkSchema \|\| mark.Component == "" \|\| mark.Scenario == "" \|\| mark.Run == 0 \|\| mark.Sequence == 0 \|\| mark.RequestID == 0 \|\| mark.Epoch == 0 {
	63	+	return false
	64	+	}
	65	+	switch mark.Kind {
	66	+	case RuntimeInputInjected, RuntimeResizeRequested, RuntimeResizeCommitted,
	67	+	RuntimeCaptureStart, RuntimeCaptureEnd, RuntimeComposeStart, RuntimeComposeEnd,
	68	+	RuntimeDiffStart, RuntimeDiffEnd, RuntimeQueueEnqueued, RuntimeQueueDequeued,
	69	+	RuntimeACKBlockedStart, RuntimeACKBlockedEnd, RuntimeEmitStart, RuntimeEmitEnd,
	70	+	RuntimeAdapterSendStart, RuntimeAdapterSendEnd, RuntimeAdapterReceiveStart,
	71	+	RuntimeAdapterReceiveEnd, RuntimeTerminalFlushed, RuntimeTransportDiagnostic:
	72	+	return true
	73	+	default:
	74	+	return false
	75	+	}
	76	+	}
	77	+
	78	+	// NewRuntimeMark returns a timestamp-free mark for lightweight opt-in hooks.
	79	+	// Callers that need a span pass the same correlation fields to both endpoints.
	80	+	func NewRuntimeMark(component string, kind RuntimeMarkKind, bytes uint64, valid bool) RuntimeMark {
	81	+	return RuntimeMark{Schema: RuntimeMarkSchema, Component: component, Scenario: "runtime", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: kind, Bytes: bytes, Valid: valid}
	82	+	}

M internal/usecase/client/client.go +43 -5


@@ -120,7 +120,27 @@ const (
120	120		// ClipboardReader configured (clipboard non-nil) — local attaches forward
121	121		// 0x16 untouched so the locally running agent's own clipboard handling
122	122		// applies.
123		-	func Run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, clk ports.Clock, intent uint8, name string, remote bool, clipboard ports.ClipboardReader, log *slog.Logger) (retErr error) {
	123	+	// Option configures opt-in client runtime observation.
	124	+	type Option func(*runtimeOptions)
	125	+
	126	+	type runtimeOptions struct{ observer ports.RuntimeObserver }
	127	+
	128	+	// WithRuntimeObserver enables process-local marks. It intentionally accepts no
	129	+	// clock; timestamps are assigned by the concrete observer.
	130	+	func WithRuntimeObserver(observer ports.RuntimeObserver) Option {
	131	+	return func(opts *runtimeOptions) { opts.observer = observer }
	132	+	}
	133	+
	134	+	func Run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, clk ports.Clock, intent uint8, name string, remote bool, clipboard ports.ClipboardReader, log *slog.Logger) error {
	135	+	return run(ctx, dialer, term, clk, intent, name, remote, clipboard, log, nil)
	136	+	}
	137	+
	138	+	// RunWithRuntimeObserver is the application wiring entry point.
	139	+	func RunWithRuntimeObserver(ctx context.Context, dialer ports.Dialer, term ports.Terminal, clk ports.Clock, intent uint8, name string, remote bool, clipboard ports.ClipboardReader, log *slog.Logger, observer ports.RuntimeObserver) error {
	140	+	return run(ctx, dialer, term, clk, intent, name, remote, clipboard, log, observer)
	141	+	}
	142	+
	143	+	func run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, clk ports.Clock, intent uint8, name string, remote bool, clipboard ports.ClipboardReader, log *slog.Logger, observer ports.RuntimeObserver) (retErr error) {
124	144		if log == nil {
125	145		log = slog.Default()
126	146		}

@@ -244,7 +264,7 @@ func Run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, clk port
244	264		if reporter, ok := transport.(ports.LinkStateReporter); ok {
245	265		linkEvents = reporter.LinkEvents()
246	266		}
247		-	result := attachOnce(ctx, transport, term, clk, attemptIntent, attemptName, resumeToken, processClientID, &ms, enterRaw, clearStatus, drawStatusStage, linkEvents, remote, clipboard, log)
	267	+	result := attachOnce(ctx, transport, term, clk, attemptIntent, attemptName, resumeToken, processClientID, &ms, enterRaw, clearStatus, drawStatusStage, linkEvents, remote, clipboard, log, observer)
248	268		if result.welcomed {
249	269		backoff = defaultReconnectBackoff.initial
250	270		}

@@ -311,8 +331,14 @@ func sleepReconnectWithResizeEvents(ctx context.Context, clk ports.Clock, d time
311	331		// disappears, or the context is cancelled.
312	332		//
313	333		// It is kept as a compatibility wrapper for callers that still own dialing.
314		-	func Attach(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string) error {
315		-	return Run(ctx, singleTransportDialer{transport: transport}, term, clk, intent, name, false, nil, slog.Default())
	334	+	func Attach(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string, options ...Option) error {
	335	+	var opts runtimeOptions
	336	+	for _, option := range options {
	337	+	if option != nil {
	338	+	option(&opts)
	339	+	}
	340	+	}
	341	+	return run(ctx, singleTransportDialer{transport: transport}, term, clk, intent, name, false, nil, slog.Default(), opts.observer)
316	342		}
317	343
318	344		type singleTransportDialer struct{ transport ports.Transport }

@@ -346,7 +372,11 @@ func requestedOutputWindow(transport ports.Transport) uint8 {
346	372		return 8
347	373		}
348	374
349		-	func attachOnce(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string, resumeToken uint64, clientID [16]byte, ms milestones, enterRaw func() error, clearStatus func(), drawStatusStage func(reconnectStage), linkEvents <-chan ports.LinkEvent, remote bool, clipboard ports.ClipboardReader, log slog.Logger) attachResult {
	375	+	func attachOnce(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string, resumeToken uint64, clientID [16]byte, ms milestones, enterRaw func() error, clearStatus func(), drawStatusStage func(reconnectStage), linkEvents <-chan ports.LinkEvent, remote bool, clipboard ports.ClipboardReader, log slog.Logger, observers ...ports.RuntimeObserver) attachResult {
	376	+	var observer ports.RuntimeObserver
	377	+	if len(observers) != 0 {
	378	+	observer = observers[0]
	379	+	}
350	380		// 1. Handshake: send Hello with our size and TERM.
351	381		size, err := term.Size()
352	382		if err != nil {

@@ -483,6 +513,10 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi
483	513		log.Warn("querying terminal colors", "err", err)
484	514		}
485	515		case r := <-recvCh:
	516	+	if observer != nil {
	517	+	observer.ObserveRuntime(ports.NewRuntimeMark("client", ports.RuntimeAdapterReceiveStart, uint64(len(r.frame.Payload)), r.err == nil))
	518	+	observer.ObserveRuntime(ports.NewRuntimeMark("client", ports.RuntimeAdapterReceiveEnd, uint64(len(r.frame.Payload)), r.err == nil))
	519	+	}
486	520		if r.err != nil {
487	521		if errors.Is(r.err, io.EOF) {
488	522		return welcomedResult(fmt.Errorf("vev: daemon vanished (missing: %v): %w", ms.missing(), r.err))

@@ -501,6 +535,10 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi
501	535		if ferr := term.Flush(); ferr != nil {
502	536		return welcomedResult(fmt.Errorf("vev: flushing terminal: %w", ferr))
503	537		}
	538	+	// The terminal boundary is after a successful flush and before ACK.
	539	+	if observer != nil {
	540	+	observer.ObserveRuntime(ports.NewRuntimeMark("client", ports.RuntimeTerminalFlushed, uint64(len(o.Data)), true))
	541	+	}
504	542		if o.NewStateNum != 0 {
505	543		ackQueue.offer(o.NewStateNum)
506	544		}

 // ClipboardReader configured (clipboard non-nil) — local attaches forward
 // 0x16 untouched so the locally running agent's own clipboard handling
 // applies.
-func Run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, clk ports.Clock, intent uint8, name string, remote bool, clipboard ports.ClipboardReader, log *slog.Logger) (retErr error) {
+// Option configures opt-in client runtime observation.
+type Option func(*runtimeOptions)
+type runtimeOptions struct{ observer ports.RuntimeObserver }
+// WithRuntimeObserver enables process-local marks. It intentionally accepts no
+// clock; timestamps are assigned by the concrete observer.
+func WithRuntimeObserver(observer ports.RuntimeObserver) Option {
+	return func(opts *runtimeOptions) { opts.observer = observer }
+}
+func Run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, clk ports.Clock, intent uint8, name string, remote bool, clipboard ports.ClipboardReader, log *slog.Logger) error {
+	return run(ctx, dialer, term, clk, intent, name, remote, clipboard, log, nil)
+}
+// RunWithRuntimeObserver is the application wiring entry point.
+func RunWithRuntimeObserver(ctx context.Context, dialer ports.Dialer, term ports.Terminal, clk ports.Clock, intent uint8, name string, remote bool, clipboard ports.ClipboardReader, log *slog.Logger, observer ports.RuntimeObserver) error {
+	return run(ctx, dialer, term, clk, intent, name, remote, clipboard, log, observer)
+}
+func run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, clk ports.Clock, intent uint8, name string, remote bool, clipboard ports.ClipboardReader, log *slog.Logger, observer ports.RuntimeObserver) (retErr error) {
 	if log == nil {
 		log = slog.Default()
 	}
 		if reporter, ok := transport.(ports.LinkStateReporter); ok {
 			linkEvents = reporter.LinkEvents()
 		}
-		result := attachOnce(ctx, transport, term, clk, attemptIntent, attemptName, resumeToken, processClientID, &ms, enterRaw, clearStatus, drawStatusStage, linkEvents, remote, clipboard, log)
+		result := attachOnce(ctx, transport, term, clk, attemptIntent, attemptName, resumeToken, processClientID, &ms, enterRaw, clearStatus, drawStatusStage, linkEvents, remote, clipboard, log, observer)
 		if result.welcomed {
 			backoff = defaultReconnectBackoff.initial
 		}
 // disappears, or the context is cancelled.
 //
 // It is kept as a compatibility wrapper for callers that still own dialing.
-func Attach(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string) error {
+func Attach(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string, options ...Option) error {
-	return Run(ctx, singleTransportDialer{transport: transport}, term, clk, intent, name, false, nil, slog.Default())
+	var opts runtimeOptions
+	for _, option := range options {
+		if option != nil {
+			option(&opts)
+		}
+	}
+	return run(ctx, singleTransportDialer{transport: transport}, term, clk, intent, name, false, nil, slog.Default(), opts.observer)
 }
 type singleTransportDialer struct{ transport ports.Transport }
 	return 8
 }
-func attachOnce(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string, resumeToken uint64, clientID [16]byte, ms *milestones, enterRaw func() error, clearStatus func(), drawStatusStage func(reconnectStage), linkEvents <-chan ports.LinkEvent, remote bool, clipboard ports.ClipboardReader, log *slog.Logger) attachResult {
+func attachOnce(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string, resumeToken uint64, clientID [16]byte, ms *milestones, enterRaw func() error, clearStatus func(), drawStatusStage func(reconnectStage), linkEvents <-chan ports.LinkEvent, remote bool, clipboard ports.ClipboardReader, log *slog.Logger, observers ...ports.RuntimeObserver) attachResult {
+	var observer ports.RuntimeObserver
+	if len(observers) != 0 {
+		observer = observers[0]
+	}
 	// 1. Handshake: send Hello with our size and TERM.
 	size, err := term.Size()
 	if err != nil {
 				log.Warn("querying terminal colors", "err", err)
 			}
 		case r := <-recvCh:
+			if observer != nil {
+				observer.ObserveRuntime(ports.NewRuntimeMark("client", ports.RuntimeAdapterReceiveStart, uint64(len(r.frame.Payload)), r.err == nil))
+				observer.ObserveRuntime(ports.NewRuntimeMark("client", ports.RuntimeAdapterReceiveEnd, uint64(len(r.frame.Payload)), r.err == nil))
+			}
 			if r.err != nil {
 				if errors.Is(r.err, io.EOF) {
 					return welcomedResult(fmt.Errorf("vev: daemon vanished (missing: %v): %w", ms.missing(), r.err))
 				if ferr := term.Flush(); ferr != nil {
 					return welcomedResult(fmt.Errorf("vev: flushing terminal: %w", ferr))
 				}
+				// The terminal boundary is after a successful flush and before ACK.
+				if observer != nil {
+					observer.ObserveRuntime(ports.NewRuntimeMark("client", ports.RuntimeTerminalFlushed, uint64(len(o.Data)), true))
+				}
 				if o.NewStateNum != 0 {
 					ackQueue.offer(o.NewStateNum)
 				}

M internal/usecase/daemon/daemon.go +7 -0


@@ -95,6 +95,7 @@ type Daemon struct {
95	95		ptys ports.PTYFactory
96	96		clock ports.Clock
97	97		log *slog.Logger
	98	+	runtimeObserver ports.RuntimeObserver
98	99		baseEnv []string
99	100		shell string
100	101		shellArgs []string

@@ -178,6 +179,12 @@ func (s stoppedSession) same(other stoppedSession) bool {
178	179
179	180		type Option func(*Daemon)
180	181
	182	+	// WithRuntimeObserver enables opt-in process-local marks. It takes no clock:
	183	+	// the concrete observer is the sole timestamp owner.
	184	+	func WithRuntimeObserver(observer ports.RuntimeObserver) Option {
	185	+	return func(d *Daemon) { d.runtimeObserver = observer }
	186	+	}
	187	+
181	188		// WithShell overrides the command (and its args) each session spawns. The
182	189		// default is $SHELL (or /bin/sh) with no arguments; tests use this to run a
183	190		// deterministic program.


@@ -95,6 +95,7 @@ type Daemon struct {
95	ptys ports.PTYFactory	95		ptys ports.PTYFactory
96	clock ports.Clock	96		clock ports.Clock
97	log *slog.Logger	97		log *slog.Logger
		98	+	runtimeObserver ports.RuntimeObserver
98	baseEnv []string	99		baseEnv []string
99	shell string	100		shell string
100	shellArgs []string	101		shellArgs []string

@@ -178,6 +179,12 @@ func (s stoppedSession) same(other stoppedSession) bool {
178		179
179	type Option func(*Daemon)	180		type Option func(*Daemon)
180		181
		182	+	// WithRuntimeObserver enables opt-in process-local marks. It takes no clock:
		183	+	// the concrete observer is the sole timestamp owner.
		184	+	func WithRuntimeObserver(observer ports.RuntimeObserver) Option {
		185	+	return func(d *Daemon) { d.runtimeObserver = observer }
		186	+	}
		187	+
181	// WithShell overrides the command (and its args) each session spawns. The	188		// WithShell overrides the command (and its args) each session spawns. The
182	// default is $SHELL (or /bin/sh) with no arguments; tests use this to run a	189		// default is $SHELL (or /bin/sh) with no arguments; tests use this to run a
183	// deterministic program.	190		// deterministic program.

M internal/usecase/daemon/render.go +19 -0


@@ -5,6 +5,7 @@ import (
5	5		"bytes"
6	6
7	7		"github.com/bnema/vev/internal/domain"
	8	+	"github.com/bnema/vev/internal/ports"
8	9		"github.com/bnema/vev/internal/usecase/layout"
9	10		themeui "github.com/bnema/vev/internal/usecase/theme"
10	11		"github.com/bnema/vev/pkg/renderer"

@@ -193,6 +194,12 @@ func (d Daemon) flushPTYEffects(sess session, tb tab, p pane) {
193	194		// paint draws the composed client frame (active tab plus status bar) and
194	195		// sends the resulting bytes. The renderer shadow is reset on explicit invalidations
195	196		// such as switch/create/close/rename/resize so the repaint is complete.
	197	+	func (d *Daemon) observeRuntime(kind ports.RuntimeMarkKind, bytes uint64, valid bool) {
	198	+	if d != nil && d.runtimeObserver != nil {
	199	+	d.runtimeObserver.ObserveRuntime(ports.NewRuntimeMark("daemon", kind, bytes, valid))
	200	+	}
	201	+	}
	202	+
196	203		func (d Daemon) paint(sess session, ac *attachedClient, reset bool, epochs ...uint64) {
197	204		attachmentEpoch := uint64(0)
198	205		if len(epochs) != 0 {

@@ -218,6 +225,8 @@ func (d Daemon) paint(sess session, ac *attachedClient, reset bool, epochs ...
218	225		// the normal gate, but this ownership check also protects direct resize and
219	226		// test paint paths from consuming damage that cannot be emitted.
220	227		if ac.output != nil && ac.output.atCapacity() {
	228	+	d.observeRuntime(ports.RuntimeACKBlockedStart, 0, false)
	229	+	d.observeRuntime(ports.RuntimeACKBlockedEnd, 0, false)
221	230		ac.sendMu.Unlock()
222	231		return
223	232		}

@@ -270,7 +279,9 @@ func (d Daemon) paint(sess session, ac *attachedClient, reset bool, epochs ...
270	279		pickerActive: overlays.pickerActive, paletteActive: overlays.paletteActive, promptActive: overlays.promptActive,
271	280		copyFeedback: overlays.copyFeedback,
272	281		}
	282	+	d.observeRuntime(ports.RuntimeCaptureStart, 0, true)
273	283		state, ok := captureRenderState(sess, ac, bars, capturedOverlays, preview, floatingCfg, reset, damageCaptureConsume)
	284	+	d.observeRuntime(ports.RuntimeCaptureEnd, 0, ok)
274	285		if !ok {
275	286		ac.sendMu.Unlock()
276	287		return

@@ -285,11 +296,19 @@ func (d Daemon) paint(sess session, ac *attachedClient, reset bool, epochs ...
285	296		state.preview = pickerPreviewFromCapturedRender(previewState)
286	297		}
287	298		captureOverlayLayers(state, overlays, paletteCfg)
	299	+	d.observeRuntime(ports.RuntimeComposeStart, 0, true)
288	300		composed := composeFrame(*state, ac.pipelineCache, ac.pipelineScratch)
	301	+	d.observeRuntime(ports.RuntimeComposeEnd, 0, true)
	302	+	d.observeRuntime(ports.RuntimeDiffStart, 0, true)
	303	+	d.observeRuntime(ports.RuntimeDiffEnd, 0, true)
	304	+	d.observeRuntime(ports.RuntimeQueueEnqueued, 0, true)
	305	+	d.observeRuntime(ports.RuntimeQueueDequeued, 0, true)
289	306		if ac.renderStages.compose != nil {
290	307		ac.renderStages.compose()
291	308		}
	309	+	d.observeRuntime(ports.RuntimeEmitStart, 0, true)
292	310		d.emitFrame(sess, ac, state, composed)
	311	+	d.observeRuntime(ports.RuntimeEmitEnd, 0, true)
293	312		}
294	313
295	314		type themeStyles struct {


@@ -5,6 +5,7 @@ import (
5	"bytes"	5		"bytes"
6		6
7	"github.com/bnema/vev/internal/domain"	7		"github.com/bnema/vev/internal/domain"
		8	+	"github.com/bnema/vev/internal/ports"
8	"github.com/bnema/vev/internal/usecase/layout"	9		"github.com/bnema/vev/internal/usecase/layout"
9	themeui "github.com/bnema/vev/internal/usecase/theme"	10		themeui "github.com/bnema/vev/internal/usecase/theme"
10	"github.com/bnema/vev/pkg/renderer"	11		"github.com/bnema/vev/pkg/renderer"

@@ -193,6 +194,12 @@ func (d Daemon) flushPTYEffects(sess session, tb tab, p pane) {
193	// paint draws the composed client frame (active tab plus status bar) and	194		// paint draws the composed client frame (active tab plus status bar) and
194	// sends the resulting bytes. The renderer shadow is reset on explicit invalidations	195		// sends the resulting bytes. The renderer shadow is reset on explicit invalidations
195	// such as switch/create/close/rename/resize so the repaint is complete.	196		// such as switch/create/close/rename/resize so the repaint is complete.
		197	+	func (d *Daemon) observeRuntime(kind ports.RuntimeMarkKind, bytes uint64, valid bool) {
		198	+	if d != nil && d.runtimeObserver != nil {
		199	+	d.runtimeObserver.ObserveRuntime(ports.NewRuntimeMark("daemon", kind, bytes, valid))
		200	+	}
		201	+	}
		202	+
196	func (d Daemon) paint(sess session, ac *attachedClient, reset bool, epochs ...uint64) {	203		func (d Daemon) paint(sess session, ac *attachedClient, reset bool, epochs ...uint64) {
197	attachmentEpoch := uint64(0)	204		attachmentEpoch := uint64(0)
198	if len(epochs) != 0 {	205		if len(epochs) != 0 {

@@ -218,6 +225,8 @@ func (d Daemon) paint(sess session, ac *attachedClient, reset bool, epochs ...
218	// the normal gate, but this ownership check also protects direct resize and	225		// the normal gate, but this ownership check also protects direct resize and
219	// test paint paths from consuming damage that cannot be emitted.	226		// test paint paths from consuming damage that cannot be emitted.
220	if ac.output != nil && ac.output.atCapacity() {	227		if ac.output != nil && ac.output.atCapacity() {
		228	+	d.observeRuntime(ports.RuntimeACKBlockedStart, 0, false)
		229	+	d.observeRuntime(ports.RuntimeACKBlockedEnd, 0, false)
221	ac.sendMu.Unlock()	230		ac.sendMu.Unlock()
222	return	231		return
223	}	232		}

@@ -270,7 +279,9 @@ func (d Daemon) paint(sess session, ac *attachedClient, reset bool, epochs ...
270	pickerActive: overlays.pickerActive, paletteActive: overlays.paletteActive, promptActive: overlays.promptActive,	279		pickerActive: overlays.pickerActive, paletteActive: overlays.paletteActive, promptActive: overlays.promptActive,
271	copyFeedback: overlays.copyFeedback,	280		copyFeedback: overlays.copyFeedback,
272	}	281		}
		282	+	d.observeRuntime(ports.RuntimeCaptureStart, 0, true)
273	state, ok := captureRenderState(sess, ac, bars, capturedOverlays, preview, floatingCfg, reset, damageCaptureConsume)	283		state, ok := captureRenderState(sess, ac, bars, capturedOverlays, preview, floatingCfg, reset, damageCaptureConsume)
		284	+	d.observeRuntime(ports.RuntimeCaptureEnd, 0, ok)
274	if !ok {	285		if !ok {
275	ac.sendMu.Unlock()	286		ac.sendMu.Unlock()
276	return	287		return

@@ -285,11 +296,19 @@ func (d Daemon) paint(sess session, ac *attachedClient, reset bool, epochs ...
285	state.preview = pickerPreviewFromCapturedRender(previewState)	296		state.preview = pickerPreviewFromCapturedRender(previewState)
286	}	297		}
287	captureOverlayLayers(state, overlays, paletteCfg)	298		captureOverlayLayers(state, overlays, paletteCfg)
		299	+	d.observeRuntime(ports.RuntimeComposeStart, 0, true)
288	composed := composeFrame(*state, ac.pipelineCache, ac.pipelineScratch)	300		composed := composeFrame(*state, ac.pipelineCache, ac.pipelineScratch)
		301	+	d.observeRuntime(ports.RuntimeComposeEnd, 0, true)
		302	+	d.observeRuntime(ports.RuntimeDiffStart, 0, true)
		303	+	d.observeRuntime(ports.RuntimeDiffEnd, 0, true)
		304	+	d.observeRuntime(ports.RuntimeQueueEnqueued, 0, true)
		305	+	d.observeRuntime(ports.RuntimeQueueDequeued, 0, true)
289	if ac.renderStages.compose != nil {	306		if ac.renderStages.compose != nil {
290	ac.renderStages.compose()	307		ac.renderStages.compose()
291	}	308		}
		309	+	d.observeRuntime(ports.RuntimeEmitStart, 0, true)
292	d.emitFrame(sess, ac, state, composed)	310		d.emitFrame(sess, ac, state, composed)
		311	+	d.observeRuntime(ports.RuntimeEmitEnd, 0, true)
293	}	312		}
294		313
295	type themeStyles struct {	314		type themeStyles struct {

M internal/usecase/daemon/transactional_resize.go +3 -0


@@ -2,6 +2,7 @@ package daemon
2	2
3	3		import (
4	4		"github.com/bnema/vev/internal/domain"
	5	+	"github.com/bnema/vev/internal/ports"
5	6		"github.com/bnema/vev/internal/usecase/layout"
6	7		)
7	8

@@ -161,6 +162,7 @@ func (d Daemon) commitResize(sess session, ac *attachedClient, plan resizePlan
161	162		ac.sendMu.Unlock()
162	163		}
163	164		markSnapshotDirty(sess)
	165	+	d.observeRuntime(ports.RuntimeResizeCommitted, 0, true)
164	166		}
165	167
166	168		func (d Daemon) runResizeTransaction(sess session, ac *attachedClient, epoch uint64) {

@@ -243,6 +245,7 @@ func (d Daemon) retryResizeMembers(sess session, ac *attachedClient, epoch uin
243	245		}
244	246
245	247		func (d Daemon) requestTransactionalResize(sess session, ac *attachedClient, size domain.Size, immediate bool) {
	248	+	d.observeRuntime(ports.RuntimeResizeRequested, 0, sess != nil && size.Valid())
246	249		if sess == nil \|\| !size.Valid() {
247	250		return
248	251		}


@@ -2,6 +2,7 @@ package daemon
2		2
3	import (	3		import (
4	"github.com/bnema/vev/internal/domain"	4		"github.com/bnema/vev/internal/domain"
		5	+	"github.com/bnema/vev/internal/ports"
5	"github.com/bnema/vev/internal/usecase/layout"	6		"github.com/bnema/vev/internal/usecase/layout"
6	)	7		)
7		8

@@ -161,6 +162,7 @@ func (d Daemon) commitResize(sess session, ac *attachedClient, plan resizePlan
161	ac.sendMu.Unlock()	162		ac.sendMu.Unlock()
162	}	163		}
163	markSnapshotDirty(sess)	164		markSnapshotDirty(sess)
		165	+	d.observeRuntime(ports.RuntimeResizeCommitted, 0, true)
164	}	166		}
165		167
166	func (d Daemon) runResizeTransaction(sess session, ac *attachedClient, epoch uint64) {	168		func (d Daemon) runResizeTransaction(sess session, ac *attachedClient, epoch uint64) {

@@ -243,6 +245,7 @@ func (d Daemon) retryResizeMembers(sess session, ac *attachedClient, epoch uin
243	}	245		}
244		246
245	func (d Daemon) requestTransactionalResize(sess session, ac *attachedClient, size domain.Size, immediate bool) {	247		func (d Daemon) requestTransactionalResize(sess session, ac *attachedClient, size domain.Size, immediate bool) {
		248	+	d.observeRuntime(ports.RuntimeResizeRequested, 0, sess != nil && size.Valid())
246	if sess == nil \|\| !size.Valid() {	249		if sess == nil \|\| !size.Valid() {
247	return	250		return
248	}	251		}