1f1baaa fix(obs): mark actual runtime boundaries · vev

M internal/adapters/dgram/ssh_dialer.go +2 -1


@@ -102,6 +102,7 @@ type RemoteDialer struct {
102	102		BootstrapTimeout time.Duration
103	103		ProbeTimeout time.Duration
104	104		Log *slog.Logger
	105	+	RuntimeObserver ports.RuntimeObserver
105	106		}
106	107
107	108		func NewRemoteDialer(target, session string) RemoteDialer {

@@ -180,7 +181,7 @@ func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) {
180	181		RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
181	182		return listenUDPPacket(ctx)
182	183		},
183		-	})
	184	+	}, WithRuntimeObserver(d.RuntimeObserver))
184	185		if err != nil {
185	186		_ = pc.Close()
186	187		return nil, udpUnavailable("create UDP transport", err, &stderr)

 	BootstrapTimeout time.Duration
 	ProbeTimeout     time.Duration
 	Log              *slog.Logger
+	RuntimeObserver  ports.RuntimeObserver
 }
 func NewRemoteDialer(target, session string) RemoteDialer {
 		RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
 			return listenUDPPacket(ctx)
 		},
-	})
+	}, WithRuntimeObserver(d.RuntimeObserver))
 	if err != nil {
 		_ = pc.Close()
 		return nil, udpUnavailable("create UDP transport", err, &stderr)

M internal/adapters/dgram/transport.go +20 -11


@@ -402,16 +402,16 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
402	402		}
403	403
404	404		func (t *Transport) Send(f ports.Frame) error {
405		-	t.observeRuntime(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)), true)
	405	+	end := t.beginRuntimeOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
406	406		err := t.send(f, false)
407		-	t.observeRuntime(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), err == nil)
	407	+	end(err == nil)
408	408		return err
409	409		}
410	410
411	411		func (t *Transport) SendAsync(f ports.Frame) error {
412		-	t.observeRuntime(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)), true)
	412	+	end := t.beginRuntimeOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
413	413		err := t.send(f, true)
414		-	t.observeRuntime(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), err == nil)
	414	+	end(err == nil)
415	415		return err
416	416		}
417	417

@@ -592,21 +592,21 @@ func (t *Transport) removePendingLocked(seq uint64) bool {
592	592		}
593	593
594	594		func (t *Transport) Recv() (ports.Frame, error) {
595		-	t.observeRuntime(ports.RuntimeAdapterReceiveStart, 0, true)
	595	+	end := t.beginRuntimeOperation(ports.RuntimeAdapterReceiveStart, 0)
596	596		select {
597	597		case f := <-t.in:
598		-	t.observeRuntime(ports.RuntimeAdapterReceiveEnd, uint64(len(f.Payload)), true)
	598	+	end(true)
599	599		return f, nil
600	600		case <-t.done:
601	601		t.mu.Lock()
602	602		err := t.closeErr
603	603		t.mu.Unlock()
604	604		if err != nil {
605		-	t.observeRuntime(ports.RuntimeAdapterReceiveEnd, 0, false)
	605	+	end(false)
606	606		return ports.Frame{}, err
607	607		}
608	608		err = errors.New("dgram: closed")
609		-	t.observeRuntime(ports.RuntimeAdapterReceiveEnd, 0, false)
	609	+	end(false)
610	610		return ports.Frame{}, err
611	611		}
612	612		}

@@ -1614,9 +1614,18 @@ func decodeData(b []byte) (uint64, bool, ports.Frame, bool) {
1614	1614		}
1615	1615
1616	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))
	1617	+	func (t *Transport) beginRuntimeOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
	1618	+	if t.runtimeObserver == nil {
	1619	+	return func(bool) {}
	1620	+	}
	1621	+	correlation := ports.NewRuntimeCorrelation()
	1622	+	t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, start, bytes, true))
	1623	+	end := ports.RuntimeAdapterSendEnd
	1624	+	if start == ports.RuntimeAdapterReceiveStart {
	1625	+	end = ports.RuntimeAdapterReceiveEnd
	1626	+	}
	1627	+	return func(valid bool) {
	1628	+	t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, end, bytes, valid))
1620	1629		}
1621	1630		}
1622	1631

 }
 func (t *Transport) Send(f ports.Frame) error {
-	t.observeRuntime(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)), true)
+	end := t.beginRuntimeOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
 	err := t.send(f, false)
-	t.observeRuntime(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), err == nil)
+	end(err == nil)
 	return err
 }
 func (t *Transport) SendAsync(f ports.Frame) error {
-	t.observeRuntime(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)), true)
+	end := t.beginRuntimeOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
 	err := t.send(f, true)
-	t.observeRuntime(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), err == nil)
+	end(err == nil)
 	return err
 }
 }
 func (t *Transport) Recv() (ports.Frame, error) {
-	t.observeRuntime(ports.RuntimeAdapterReceiveStart, 0, true)
+	end := t.beginRuntimeOperation(ports.RuntimeAdapterReceiveStart, 0)
 	select {
 	case f := <-t.in:
-		t.observeRuntime(ports.RuntimeAdapterReceiveEnd, uint64(len(f.Payload)), true)
+		end(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)
+			end(false)
 			return ports.Frame{}, err
 		}
 		err = errors.New("dgram: closed")
-		t.observeRuntime(ports.RuntimeAdapterReceiveEnd, 0, false)
+		end(false)
 		return ports.Frame{}, err
 	}
 }
 }
 // DatagramTransport marks Transport as a UDP-style datagram transport.
-func (t *Transport) observeRuntime(kind ports.RuntimeMarkKind, bytes uint64, valid bool) {
+func (t *Transport) beginRuntimeOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
-	if t.runtimeObserver != nil {
+	if t.runtimeObserver == nil {
-		t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMark("dgram", kind, bytes, valid))
+		return func(bool) {}
+	}
+	correlation := ports.NewRuntimeCorrelation()
+	t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, start, bytes, true))
+	end := ports.RuntimeAdapterSendEnd
+	if start == ports.RuntimeAdapterReceiveStart {
+		end = ports.RuntimeAdapterReceiveEnd
+	}
+	return func(valid bool) {
+		t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, end, bytes, valid))
 	}
 }

M internal/adapters/ipc/dial.go +2 -2


@@ -12,11 +12,11 @@ import (
12	12		// a Transport speaking vev's framed protocol. A dial failure (no daemon
13	13		// listening, or a stale/absent socket) is reported as a wrapped error so
14	14		// callers can decide whether to spawn a daemon and retry.
15		-	func Dial(dir string) (ports.Transport, error) {
	15	+	func Dial(dir string, opts ...Option) (ports.Transport, error) {
16	16		sockPath := filepath.Join(dir, socketFileName)
17	17		conn, err := net.Dial("unix", sockPath)
18	18		if err != nil {
19	19		return nil, fmt.Errorf("ipc: dial %s: %w", sockPath, err)
20	20		}
21		-	return NewTransport(conn), nil
	21	+	return NewTransport(conn, opts...), nil
22	22		}


@@ -12,11 +12,11 @@ import (
12		// a Transport speaking vev's framed protocol. A dial failure (no daemon	12		// a Transport speaking vev's framed protocol. A dial failure (no daemon
13		// listening, or a stale/absent socket) is reported as a wrapped error so	13		// listening, or a stale/absent socket) is reported as a wrapped error so
14		// callers can decide whether to spawn a daemon and retry.	14		// callers can decide whether to spawn a daemon and retry.
15	-	func Dial(dir string) (ports.Transport, error) {	15	+	func Dial(dir string, opts ...Option) (ports.Transport, error) {
16		sockPath := filepath.Join(dir, socketFileName)	16		sockPath := filepath.Join(dir, socketFileName)
17		conn, err := net.Dial("unix", sockPath)	17		conn, err := net.Dial("unix", sockPath)
18		if err != nil {	18		if err != nil {
19		return nil, fmt.Errorf("ipc: dial %s: %w", sockPath, err)	19		return nil, fmt.Errorf("ipc: dial %s: %w", sockPath, err)
20		}	20		}
21	-	return NewTransport(conn), nil	21	+	return NewTransport(conn, opts...), nil
22		}	22		}

M internal/adapters/ipc/listener.go +4 -3


@@ -25,6 +25,7 @@ var ErrDaemonRunning = errors.New("ipc: a daemon is already listening on this so
25	25		type unixListener struct {
26	26		ln *net.UnixListener
27	27		addr string
	28	+	opts []Option
28	29		}
29	30
30	31		// Listen creates the socket directory (0700) if needed and starts

@@ -35,7 +36,7 @@ type unixListener struct {
35	36		// the previous owner died without cleaning up, so the stale socket file is
36	37		// unlinked and bind is retried once. A successful dial means a live daemon
37	38		// owns the socket, and Listen returns ErrDaemonRunning.
38		-	func Listen(dir string) (ports.Listener, error) {
	39	+	func Listen(dir string, opts ...Option) (ports.Listener, error) {
39	40		if err := safedir.EnsurePrivate(dir); err != nil {
40	41		return nil, fmt.Errorf("ipc: securing socket directory: %w", err)
41	42		}

@@ -69,7 +70,7 @@ func Listen(dir string) (ports.Listener, error) {
69	70		return nil, fmt.Errorf("ipc: chmod %s: %w", sockPath, err)
70	71		}
71	72
72		-	return &unixListener{ln: ln, addr: sockPath}, nil
	73	+	return &unixListener{ln: ln, addr: sockPath, opts: opts}, nil
73	74		}
74	75
75	76		// bindUnix binds and listens on sockPath.

@@ -104,7 +105,7 @@ func (l *unixListener) Accept() (ports.Transport, error) {
104	105		if err != nil {
105	106		return nil, err
106	107		}
107		-	return NewTransport(conn), nil
	108	+	return NewTransport(conn, l.opts...), nil
108	109		}
109	110
110	111		// Close stops accepting connections. The underlying net.UnixListener

 type unixListener struct {
 	ln   *net.UnixListener
 	addr string
+	opts []Option
 }
 // Listen creates the socket directory (0700) if needed and starts
 // the previous owner died without cleaning up, so the stale socket file is
 // unlinked and bind is retried once. A successful dial means a live daemon
 // owns the socket, and Listen returns ErrDaemonRunning.
-func Listen(dir string) (ports.Listener, error) {
+func Listen(dir string, opts ...Option) (ports.Listener, error) {
 	if err := safedir.EnsurePrivate(dir); err != nil {
 		return nil, fmt.Errorf("ipc: securing socket directory: %w", err)
 	}
 		return nil, fmt.Errorf("ipc: chmod %s: %w", sockPath, err)
 	}
-	return &unixListener{ln: ln, addr: sockPath}, nil
+	return &unixListener{ln: ln, addr: sockPath, opts: opts}, nil
 }
 // bindUnix binds and listens on sockPath.
 	if err != nil {
 		return nil, err
 	}
-	return NewTransport(conn), nil
+	return NewTransport(conn, l.opts...), nil
 }
 // Close stops accepting connections. The underlying net.UnixListener

M internal/adapters/ipc/transport.go +21 -12


@@ -70,10 +70,10 @@ func NewTransport(conn net.Conn, opts ...Option) ports.Transport {
70	70		// the type byte and payload), the type byte, then the payload — assembled
71	71		// into one buffer and written with a single Write call.
72	72		func (t *unixTransport) Send(f ports.Frame) error {
73		-	t.observe(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)), true)
	73	+	end := t.beginOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
74	74		n := 1 + len(f.Payload) // type + payload
75	75		if n > maxFrameLen {
76		-	t.observe(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), false)
	76	+	end(false)
77	77		return ErrFrameTooLarge
78	78		}
79	79

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

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

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

@@ -123,13 +123,22 @@ func (t *unixTransport) Recv() (ports.Frame, error) {
123	123		payload = make([]byte, n-1)
124	124		copy(payload, t.readBuf[1:])
125	125		}
126		-	t.observe(ports.RuntimeAdapterReceiveEnd, uint64(len(payload)), true)
	126	+	end(true)
127	127		return ports.Frame{Type: ports.MsgType(t.readBuf[0]), Payload: payload}, nil
128	128		}
129	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))
	130	+	func (t *unixTransport) beginOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
	131	+	if t.observer == nil {
	132	+	return func(bool) {}
	133	+	}
	134	+	correlation := ports.NewRuntimeCorrelation()
	135	+	t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("ipc", correlation, start, bytes, true))
	136	+	end := ports.RuntimeAdapterSendEnd
	137	+	if start == ports.RuntimeAdapterReceiveStart {
	138	+	end = ports.RuntimeAdapterReceiveEnd
	139	+	}
	140	+	return func(valid bool) {
	141	+	t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("ipc", correlation, end, bytes, valid))
133	142		}
134	143		}
135	144

 // 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)
+	end := t.beginOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
 	n := 1 + len(f.Payload) // type + payload
 	if n > maxFrameLen {
-		t.observe(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), false)
+		end(false)
 		return ErrFrameTooLarge
 	}
 	copy(buf[frameHeaderLen+1:], f.Payload)
 	_, err := t.conn.Write(buf)
-	t.observe(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), err == nil)
+	end(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)
+	end := t.beginOperation(ports.RuntimeAdapterReceiveStart, 0)
 	var hdr [frameHeaderLen]byte
 	if _, err := io.ReadFull(t.conn, hdr[:]); err != nil {
-		t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
+		end(false)
 		return ports.Frame{}, err
 	}
 	n := binary.BigEndian.Uint32(hdr[:])
 	if n == 0 {
-		t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
+		end(false)
 		return ports.Frame{}, ErrZeroLengthFrame
 	}
 	if n > maxFrameLen {
-		t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
+		end(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)
+		end(false)
 		return ports.Frame{}, err
 	}
 		payload = make([]byte, n-1)
 		copy(payload, t.readBuf[1:])
 	}
-	t.observe(ports.RuntimeAdapterReceiveEnd, uint64(len(payload)), true)
+	end(true)
 	return ports.Frame{Type: ports.MsgType(t.readBuf[0]), Payload: payload}, nil
 }
-func (t *unixTransport) observe(kind ports.RuntimeMarkKind, bytes uint64, valid bool) {
+func (t *unixTransport) beginOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
-	if t.observer != nil {
+	if t.observer == nil {
-		t.observer.ObserveRuntime(ports.NewRuntimeMark("ipc", kind, bytes, valid))
+		return func(bool) {}
+	}
+	correlation := ports.NewRuntimeCorrelation()
+	t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("ipc", correlation, start, bytes, true))
+	end := ports.RuntimeAdapterSendEnd
+	if start == ports.RuntimeAdapterReceiveStart {
+		end = ports.RuntimeAdapterReceiveEnd
+	}
+	return func(valid bool) {
+		t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("ipc", correlation, end, bytes, valid))
 	}
 }

M internal/adapters/observability/jsonl.go +16 -2


@@ -31,6 +31,7 @@ type jsonl struct {
31	31		clock ports.Clock
32	32		processID string
33	33		closed bool
	34	+	err error
34	35		}
35	36
36	37		type record struct {

@@ -70,9 +71,18 @@ func (j *jsonl) ObserveRuntime(mark ports.RuntimeMark) {
70	71		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	72		encoded, err := json.Marshal(r)
72	73		if err != nil {
	74	+	j.recordError(err)
73	75		return
74	76		}
75		-	_, _ = j.file.Write(append(encoded, '\n'))
	77	+	if _, err := j.file.Write(append(encoded, '\n')); err != nil {
	78	+	j.recordError(err)
	79	+	}
	80	+	}
	81	+
	82	+	func (j *jsonl) recordError(err error) {
	83	+	if err != nil && j.err == nil {
	84	+	j.err = err
	85	+	}
76	86		}
77	87
78	88		func (j *jsonl) Close() error {

@@ -82,5 +92,9 @@ func (j *jsonl) Close() error {
82	92		return nil
83	93		}
84	94		j.closed = true
85		-	return j.file.Close()
	95	+	closeErr := j.file.Close()
	96	+	if j.err != nil {
	97	+	return j.err
	98	+	}
	99	+	return closeErr
86	100		}

 	clock     ports.Clock
 	processID string
 	closed    bool
+	err       error
 }
 type record struct {
 	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}
 	encoded, err := json.Marshal(r)
 	if err != nil {
+		j.recordError(err)
 		return
 	}
-	_, _ = j.file.Write(append(encoded, '\n'))
+	if _, err := j.file.Write(append(encoded, '\n')); err != nil {
+		j.recordError(err)
+	}
+}
+func (j *jsonl) recordError(err error) {
+	if err != nil && j.err == nil {
+		j.err = err
+	}
 }
 func (j *jsonl) Close() error {
 		return nil
 	}
 	j.closed = true
-	return j.file.Close()
+	closeErr := j.file.Close()
+	if j.err != nil {
+		return j.err
+	}
+	return closeErr
 }

M internal/adapters/observability/jsonl_test.go +13 -0


@@ -73,6 +73,19 @@ func TestTransportObservabilityJSONLRejectsInvalidRequiredFields(t *testing.T) {
73	73		}
74	74		}
75	75
	76	+	func TestTransportObservabilityJSONLWriteErrorSurfacesOnClose(t *testing.T) {
	77	+	// /dev/full accepts Open but rejects every write, exercising the concrete
	78	+	// observer lifecycle without changing the one-way RuntimeObserver port.
	79	+	observer, closer, err := NewJSONL("/dev/full", fixedRuntimeClock{now: time.Unix(0, 1)}, "p")
	80	+	if err != nil {
	81	+	t.Skipf("/dev/full unavailable: %v", err)
	82	+	}
	83	+	observer.ObserveRuntime(validRuntimeMark(ports.RuntimeEmitStart))
	84	+	if err := closer.Close(); err == nil {
	85	+	t.Fatal("Close() error = nil, want surfaced JSONL write error")
	86	+	}
	87	+	}
	88	+
76	89		func TestTransportObservabilitySameProcessSpanClockDomains(t *testing.T) {
77	90		for _, tc := range []struct {
78	91		name string


@@ -73,6 +73,19 @@ func TestTransportObservabilityJSONLRejectsInvalidRequiredFields(t *testing.T) {
73	}	73		}
74	}	74		}
75		75
		76	+	func TestTransportObservabilityJSONLWriteErrorSurfacesOnClose(t *testing.T) {
		77	+	// /dev/full accepts Open but rejects every write, exercising the concrete
		78	+	// observer lifecycle without changing the one-way RuntimeObserver port.
		79	+	observer, closer, err := NewJSONL("/dev/full", fixedRuntimeClock{now: time.Unix(0, 1)}, "p")
		80	+	if err != nil {
		81	+	t.Skipf("/dev/full unavailable: %v", err)
		82	+	}
		83	+	observer.ObserveRuntime(validRuntimeMark(ports.RuntimeEmitStart))
		84	+	if err := closer.Close(); err == nil {
		85	+	t.Fatal("Close() error = nil, want surfaced JSONL write error")
		86	+	}
		87	+	}
		88	+
76	func TestTransportObservabilitySameProcessSpanClockDomains(t *testing.T) {	89		func TestTransportObservabilitySameProcessSpanClockDomains(t *testing.T) {
77	for _, tc := range []struct {	90		for _, tc := range []struct {
78	name string	91		name string

M internal/adapters/remote/remote_factory.go +17 -8


@@ -11,35 +11,44 @@ import (
11	11		)
12	12
13	13		// DialerFactory selects the requested remote transport adapter.
14		-	type DialerFactory struct{}
	14	+	type DialerFactory struct{ observer ports.RuntimeObserver }
15	15
16	16		var _ ports.RemoteDialerFactory = DialerFactory{}
17	17
18	18		func NewDialerFactory() DialerFactory { return DialerFactory{} }
19	19
20		-	func (DialerFactory) DialerForRemote(target string, session string, mode ports.RemoteTransportMode, log *slog.Logger) (ports.Dialer, error) {
	20	+	// NewDialerFactoryWithRuntimeObserver wires the one process-local sink into
	21	+	// whichever concrete remote carriage adapter is selected.
	22	+	func NewDialerFactoryWithRuntimeObserver(observer ports.RuntimeObserver) DialerFactory {
	23	+	return DialerFactory{observer: observer}
	24	+	}
	25	+
	26	+	func (f DialerFactory) DialerForRemote(target string, session string, mode ports.RemoteTransportMode, log *slog.Logger) (ports.Dialer, error) {
21	27		switch mode {
22	28		case ports.RemoteTransportUDP:
23	29		if log != nil {
24	30		log.Info("remote transport selected", "mode", mode, "target", target, "session", session)
25	31		}
26		-	return dgram.NewRemoteDialerWithLogger(target, session, log), nil
	32	+	dialer := dgram.NewRemoteDialerWithLogger(target, session, log)
	33	+	dialer.RuntimeObserver = f.observer
	34	+	return dialer, nil
27	35		case ports.RemoteTransportStdio:
28	36		if log != nil {
29	37		log.Info("remote transport selected", "mode", mode, "target", target, "session", session)
30	38		}
31		-	return stdioDialer{target: target, session: session, log: log}, nil
	39	+	return stdioDialer{target: target, session: session, log: log, observer: f.observer}, nil
32	40		default:
33	41		return nil, fmt.Errorf("vev: unsupported remote transport %q", mode)
34	42		}
35	43		}
36	44
37	45		type stdioDialer struct {
38		-	target string
39		-	session string
40		-	log *slog.Logger
	46	+	target string
	47	+	session string
	48	+	log *slog.Logger
	49	+	observer ports.RuntimeObserver
41	50		}
42	51
43	52		func (d stdioDialer) Dial(ctx context.Context) (ports.Transport, error) {
44		-	return sshstdio.DialContext(ctx, d.target, d.session, d.log)
	53	+	return sshstdio.DialContextWithRuntimeObserver(ctx, d.target, d.session, d.log, sshstdio.WithRuntimeObserver(d.observer))
45	54		}

 )
 // DialerFactory selects the requested remote transport adapter.
-type DialerFactory struct{}
+type DialerFactory struct{ observer ports.RuntimeObserver }
 var _ ports.RemoteDialerFactory = DialerFactory{}
 func NewDialerFactory() DialerFactory { return DialerFactory{} }
-func (DialerFactory) DialerForRemote(target string, session string, mode ports.RemoteTransportMode, log *slog.Logger) (ports.Dialer, error) {
+// NewDialerFactoryWithRuntimeObserver wires the one process-local sink into
+// whichever concrete remote carriage adapter is selected.
+func NewDialerFactoryWithRuntimeObserver(observer ports.RuntimeObserver) DialerFactory {
+	return DialerFactory{observer: observer}
+}
+func (f DialerFactory) DialerForRemote(target string, session string, mode ports.RemoteTransportMode, log *slog.Logger) (ports.Dialer, error) {
 	switch mode {
 	case ports.RemoteTransportUDP:
 		if log != nil {
 			log.Info("remote transport selected", "mode", mode, "target", target, "session", session)
 		}
-		return dgram.NewRemoteDialerWithLogger(target, session, log), nil
+		dialer := dgram.NewRemoteDialerWithLogger(target, session, log)
+		dialer.RuntimeObserver = f.observer
+		return dialer, nil
 	case ports.RemoteTransportStdio:
 		if log != nil {
 			log.Info("remote transport selected", "mode", mode, "target", target, "session", session)
 		}
-		return stdioDialer{target: target, session: session, log: log}, nil
+		return stdioDialer{target: target, session: session, log: log, observer: f.observer}, nil
 	default:
 		return nil, fmt.Errorf("vev: unsupported remote transport %q", mode)
 	}
 }
 type stdioDialer struct {
-	target  string
+	target   string
-	session string
+	session  string
-	log     *slog.Logger
+	log      *slog.Logger
+	observer ports.RuntimeObserver
 }
 func (d stdioDialer) Dial(ctx context.Context) (ports.Transport, error) {
-	return sshstdio.DialContext(ctx, d.target, d.session, d.log)
+	return sshstdio.DialContextWithRuntimeObserver(ctx, d.target, d.session, d.log, sshstdio.WithRuntimeObserver(d.observer))
 }

M internal/adapters/sshstdio/transport.go +41 -17


@@ -72,10 +72,10 @@ type transport struct {
72	72		}
73	73
74	74		func (t *transport) Send(f ports.Frame) error {
75		-	t.observe(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)), true)
	75	+	end := t.beginOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
76	76		n := 1 + len(f.Payload)
77	77		if n > maxFrameLen {
78		-	t.observe(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), false)
	78	+	end(false)
79	79		return ErrFrameTooLarge
80	80		}
81	81

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

@@ -117,18 +117,27 @@ func (t *transport) Recv() (ports.Frame, error) {
117	117		}
118	118		if _, err := io.ReadFull(t.r, t.readBuf); err != nil {
119	119		err = t.mapEOFError(err)
120		-	t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
	120	+	end(false)
121	121		return ports.Frame{}, err
122	122		}
123	123
124	124		payload := append([]byte(nil), t.readBuf[1:]...)
125		-	t.observe(ports.RuntimeAdapterReceiveEnd, uint64(len(payload)), true)
	125	+	end(true)
126	126		return ports.Frame{Type: ports.MsgType(t.readBuf[0]), Payload: payload}, nil
127	127		}
128	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))
	129	+	func (t *transport) beginOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
	130	+	if t.observer == nil {
	131	+	return func(bool) {}
	132	+	}
	133	+	correlation := ports.NewRuntimeCorrelation()
	134	+	t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("sshstdio", correlation, start, bytes, true))
	135	+	end := ports.RuntimeAdapterSendEnd
	136	+	if start == ports.RuntimeAdapterReceiveStart {
	137	+	end = ports.RuntimeAdapterReceiveEnd
	138	+	}
	139	+	return func(valid bool) {
	140	+	t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("sshstdio", correlation, end, bytes, valid))
132	141		}
133	142		}
134	143

@@ -254,18 +263,27 @@ func Dial(target, session string) (ports.Transport, error) {
254	263		return DialContext(context.Background(), target, session)
255	264		}
256	265
	266	+	// DialContextWithRuntimeObserver is the option-bearing production constructor.
	267	+	func DialContextWithRuntimeObserver(ctx context.Context, target, session string, logger *slog.Logger, opts ...Option) (ports.Transport, error) {
	268	+	return dialContext(ctx, target, session, logger, opts...)
	269	+	}
	270	+
257	271		// DialContext is like Dial, but the context is propagated to ssh startup so a
258	272		// canceled attach attempt interrupts the local ssh process. Callers may pass a
259	273		// logger to record ssh start failures and non-clean exits without logging the
260	274		// generated command line.
261	275		func DialContext(ctx context.Context, target, session string, logger ...*slog.Logger) (ports.Transport, error) {
262		-	if err := ctx.Err(); err != nil {
263		-	return nil, err
264		-	}
265	276		var log *slog.Logger
266	277		if len(logger) > 0 {
267	278		log = logger[0]
268	279		}
	280	+	return dialContext(ctx, target, session, log)
	281	+	}
	282	+
	283	+	func dialContext(ctx context.Context, target, session string, log *slog.Logger, opts ...Option) (ports.Transport, error) {
	284	+	if err := ctx.Err(); err != nil {
	285	+	return nil, err
	286	+	}
269	287		spec := BuildCommand(target, session)
270	288		cmd := exec.CommandContext(ctx, spec.Path, spec.Args...)
271	289		stdin, err := cmd.StdinPipe()

@@ -286,5 +304,11 @@ func DialContext(ctx context.Context, target, session string, logger ...*slog.Lo
286	304		}
287	305
288	306		waiter := newProcessWaiter(cmd, stdin, &stderr, sshCloseTimeout, log, target, session)
289		-	return newTransport(stdout, stdin, waiter.close, waiter.eofErr), nil
	307	+	transport := newTransport(stdout, stdin, waiter.close, waiter.eofErr).(*transport)
	308	+	for _, opt := range opts {
	309	+	if opt != nil {
	310	+	opt(transport)
	311	+	}
	312	+	}
	313	+	return transport, nil
290	314		}

 }
 func (t *transport) Send(f ports.Frame) error {
-	t.observe(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)), true)
+	end := t.beginOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
 	n := 1 + len(f.Payload)
 	if n > maxFrameLen {
-		t.observe(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), false)
+		end(false)
 		return ErrFrameTooLarge
 	}
 	copy(buf[frameHeaderLen+1:], f.Payload)
 	t.mu.Lock()
-	defer t.mu.Unlock()
 	_, err := t.w.Write(buf)
-	t.observe(ports.RuntimeAdapterSendEnd, uint64(len(f.Payload)), err == nil)
+	t.mu.Unlock()
+	end(err == nil)
 	return err
 }
 func (t *transport) Recv() (ports.Frame, error) {
-	t.observe(ports.RuntimeAdapterReceiveStart, 0, true)
+	end := t.beginOperation(ports.RuntimeAdapterReceiveStart, 0)
 	var hdr [frameHeaderLen]byte
 	if _, err := io.ReadFull(t.r, hdr[:]); err != nil {
 		err = t.mapEOFError(err)
-		t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
+		end(false)
 		return ports.Frame{}, err
 	}
 	n := binary.BigEndian.Uint32(hdr[:])
 	if n == 0 {
-		t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
+		end(false)
 		return ports.Frame{}, ErrZeroLengthFrame
 	}
 	if n > maxFrameLen {
-		t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
+		end(false)
 		return ports.Frame{}, ErrFrameTooLarge
 	}
 	}
 	if _, err := io.ReadFull(t.r, t.readBuf); err != nil {
 		err = t.mapEOFError(err)
-		t.observe(ports.RuntimeAdapterReceiveEnd, 0, false)
+		end(false)
 		return ports.Frame{}, err
 	}
 	payload := append([]byte(nil), t.readBuf[1:]...)
-	t.observe(ports.RuntimeAdapterReceiveEnd, uint64(len(payload)), true)
+	end(true)
 	return ports.Frame{Type: ports.MsgType(t.readBuf[0]), Payload: payload}, nil
 }
-func (t *transport) observe(kind ports.RuntimeMarkKind, bytes uint64, valid bool) {
+func (t *transport) beginOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
-	if t.observer != nil {
+	if t.observer == nil {
-		t.observer.ObserveRuntime(ports.NewRuntimeMark("sshstdio", kind, bytes, valid))
+		return func(bool) {}
+	}
+	correlation := ports.NewRuntimeCorrelation()
+	t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("sshstdio", correlation, start, bytes, true))
+	end := ports.RuntimeAdapterSendEnd
+	if start == ports.RuntimeAdapterReceiveStart {
+		end = ports.RuntimeAdapterReceiveEnd
+	}
+	return func(valid bool) {
+		t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("sshstdio", correlation, end, bytes, valid))
 	}
 }
 	return DialContext(context.Background(), target, session)
 }
+// DialContextWithRuntimeObserver is the option-bearing production constructor.
+func DialContextWithRuntimeObserver(ctx context.Context, target, session string, logger *slog.Logger, opts ...Option) (ports.Transport, error) {
+	return dialContext(ctx, target, session, logger, opts...)
+}
 // DialContext is like Dial, but the context is propagated to ssh startup so a
 // canceled attach attempt interrupts the local ssh process. Callers may pass a
 // logger to record ssh start failures and non-clean exits without logging the
 // generated command line.
 func DialContext(ctx context.Context, target, session string, logger ...*slog.Logger) (ports.Transport, error) {
-	if err := ctx.Err(); err != nil {
-		return nil, err
-	}
 	var log *slog.Logger
 	if len(logger) > 0 {
 		log = logger[0]
 	}
+	return dialContext(ctx, target, session, log)
+}
+func dialContext(ctx context.Context, target, session string, log *slog.Logger, opts ...Option) (ports.Transport, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
 	spec := BuildCommand(target, session)
 	cmd := exec.CommandContext(ctx, spec.Path, spec.Args...)
 	stdin, err := cmd.StdinPipe()
 	}
 	waiter := newProcessWaiter(cmd, stdin, &stderr, sshCloseTimeout, log, target, session)
-	return newTransport(stdout, stdin, waiter.close, waiter.eofErr), nil
+	transport := newTransport(stdout, stdin, waiter.close, waiter.eofErr).(*transport)
+	for _, opt := range opts {
+		if opt != nil {
+			opt(transport)
+		}
+	}
+	return transport, nil
 }

M internal/app/run.go +32 -19


@@ -297,7 +297,16 @@ func runDaemon() error {
297	297		}
298	298		defer func() { _ = logCloser.Close() }()
299	299
300		-	ln, err := ipc.Listen(ipc.SocketDir())
	300	+	clk := clock.New()
	301	+	observer, observerCloser, err := performanceTrace(clk)
	302	+	if err != nil {
	303	+	return fmt.Errorf("vev: performance trace: %w", err)
	304	+	}
	305	+	if observerCloser != nil {
	306	+	defer func() { _ = observerCloser.Close() }()
	307	+	}
	308	+	// Every accepted local connection shares this process's serialized sink.
	309	+	ln, err := ipc.Listen(ipc.SocketDir(), ipc.WithRuntimeObserver(observer))
301	310		if err != nil {
302	311		log.Error("daemon listen failed", "socket_dir", ipc.SocketDir(), "err", err)
303	312		return fmt.Errorf("vev: daemon listen: %w", err)

@@ -320,14 +329,6 @@ func runDaemon() error {
320	329		}
321	330
322	331		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		-	}
331	332		daemonOpts := []daemon.Option(nil)
332	333		if observer != nil {
333	334		daemonOpts = append(daemonOpts, daemon.WithRuntimeObserver(observer))

@@ -388,8 +389,10 @@ func runAttach(ctx context.Context, intent uint8, name, remoteTarget string) err
388	389		defer func() { _ = observerCloser.Close() }()
389	390		}
390	391		return runAttachWithDeps(ctx, intent, name, remoteTarget, os.Getenv("VEV"), log, runAttachDeps{
391		-	localDialer: defaultLocalDialer,
392		-	remoteDialerFactory: defaultRemoteDialerFactory(),
	392	+	localDialer: func() ports.Dialer {
	393	+	return localDaemonDialer{dir: ipc.SocketDir(), observer: observer}
	394	+	},
	395	+	remoteDialerFactory: remoteadapter.NewDialerFactoryWithRuntimeObserver(observer),
393	396		selectedRemoteTransport: os.Getenv(envRemoteTransport),
394	397		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	398		return client.RunWithRuntimeObserver(ctx, dialer, terminal, behaviorClock, intent, name, remote, clipboard, log, observer)

@@ -531,10 +534,19 @@ func isDaemonVersionDrift(err error) bool {
531	534		return protocolErr.Code == ports.ErrVersionMismatch \|\| protocolErr.Text == legacyMalformedHelloSignal
532	535		}
533	536
534		-	type localDaemonDialer struct{ dir string }
	537	+	type localDaemonDialer struct {
	538	+	dir string
	539	+	observer ports.RuntimeObserver
	540	+	}
535	541
536	542		func (d localDaemonDialer) Dial(ctx context.Context) (ports.Transport, error) {
537		-	return ensureDaemon(ctx, d.dir, realDial, realSpawn, defaultBackoff)
	543	+	dial := realDial
	544	+	if d.observer != nil {
	545	+	dial = func(dir string) (ports.Transport, error) {
	546	+	return ipc.Dial(dir, ipc.WithRuntimeObserver(d.observer))
	547	+	}
	548	+	}
	549	+	return ensureDaemon(ctx, d.dir, dial, realSpawn, defaultBackoff)
538	550		}
539	551
540	552		func detachedLocalHello(name, cwd string) ports.Hello {

@@ -661,12 +673,6 @@ func runStdio(ctx context.Context) error {
661	673		defer func() { _ = logCloser.Close() }()
662	674		log.Debug("stdio proxy starting")
663	675
664		-	transport, err := ensureDaemon(ctx, ipc.SocketDir(), realDial, realSpawn, defaultBackoff)
665		-	if err != nil {
666		-	return err
667		-	}
668		-	defer func() { _ = transport.Close() }()
669		-
670	676		clk := clock.New()
671	677		observer, observerCloser, err := performanceTrace(clk)
672	678		if err != nil {

@@ -675,6 +681,13 @@ func runStdio(ctx context.Context) error {
675	681		if observerCloser != nil {
676	682		defer func() { _ = observerCloser.Close() }()
677	683		}
	684	+	transport, err := ensureDaemon(ctx, ipc.SocketDir(), func(dir string) (ports.Transport, error) {
	685	+	return ipc.Dial(dir, ipc.WithRuntimeObserver(observer))
	686	+	}, realSpawn, defaultBackoff)
	687	+	if err != nil {
	688	+	return err
	689	+	}
	690	+	defer func() { _ = transport.Close() }()
678	691		stdio := sshstdio.NewTransport(os.Stdin, os.Stdout, nil, sshstdio.WithRuntimeObserver(observer))
679	692		return proxyTransports(ctx, stdio, transport, log)
680	693		}

 	}
 	defer func() { _ = logCloser.Close() }()
-	ln, err := ipc.Listen(ipc.SocketDir())
+	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() }()
+	}
+	// Every accepted local connection shares this process's serialized sink.
+	ln, err := ipc.Listen(ipc.SocketDir(), ipc.WithRuntimeObserver(observer))
 	if err != nil {
 		log.Error("daemon listen failed", "socket_dir", ipc.SocketDir(), "err", err)
 		return fmt.Errorf("vev: daemon listen: %w", err)
 	}
 	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))
 		defer func() { _ = observerCloser.Close() }()
 	}
 	return runAttachWithDeps(ctx, intent, name, remoteTarget, os.Getenv("VEV"), log, runAttachDeps{
-		localDialer:             defaultLocalDialer,
+		localDialer: func() ports.Dialer {
-		remoteDialerFactory:     defaultRemoteDialerFactory(),
+			return localDaemonDialer{dir: ipc.SocketDir(), observer: observer}
+		},
+		remoteDialerFactory:     remoteadapter.NewDialerFactoryWithRuntimeObserver(observer),
 		selectedRemoteTransport: os.Getenv(envRemoteTransport),
 		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 {
 			return client.RunWithRuntimeObserver(ctx, dialer, terminal, behaviorClock, intent, name, remote, clipboard, log, observer)
 	return protocolErr.Code == ports.ErrVersionMismatch || protocolErr.Text == legacyMalformedHelloSignal
 }
-type localDaemonDialer struct{ dir string }
+type localDaemonDialer struct {
+	dir      string
+	observer ports.RuntimeObserver
+}
 func (d localDaemonDialer) Dial(ctx context.Context) (ports.Transport, error) {
-	return ensureDaemon(ctx, d.dir, realDial, realSpawn, defaultBackoff)
+	dial := realDial
+	if d.observer != nil {
+		dial = func(dir string) (ports.Transport, error) {
+			return ipc.Dial(dir, ipc.WithRuntimeObserver(d.observer))
+		}
+	}
+	return ensureDaemon(ctx, d.dir, dial, realSpawn, defaultBackoff)
 }
 func detachedLocalHello(name, cwd string) ports.Hello {
 	defer func() { _ = logCloser.Close() }()
 	log.Debug("stdio proxy starting")
-	transport, err := ensureDaemon(ctx, ipc.SocketDir(), realDial, realSpawn, defaultBackoff)
-	if err != nil {
-		return err
-	}
-	defer func() { _ = transport.Close() }()
 	clk := clock.New()
 	observer, observerCloser, err := performanceTrace(clk)
 	if err != nil {
 	if observerCloser != nil {
 		defer func() { _ = observerCloser.Close() }()
 	}
+	transport, 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() { _ = transport.Close() }()
 	stdio := sshstdio.NewTransport(os.Stdin, os.Stdout, nil, sshstdio.WithRuntimeObserver(observer))
 	return proxyTransports(ctx, stdio, transport, log)
 }

M internal/ports/runtime_observer.go +28 -3

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

@@ -75,8 +77,31 @@ func RuntimeMarkValid(mark RuntimeMark) bool {
75	77		}
76	78		}
77	79
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	+	// RuntimeCorrelation identifies one attempted operation. It is deliberately
	81	+	// copied to both endpoints of a span; process ID remains owned by the sink.
	82	+	type RuntimeCorrelation struct {
	83	+	Scenario string
	84	+	Run, Sequence, RequestID, Epoch uint64
	85	+	}
	86	+
	87	+	var runtimeMarkSequence atomic.Uint64
	88	+
	89	+	// NewRuntimeCorrelation makes a unique, process-local operation identity.
	90	+	// Components with a domain request/epoch should use those values directly via
	91	+	// NewRuntimeMarkWithCorrelation instead of inventing a second identity.
	92	+	func NewRuntimeCorrelation() RuntimeCorrelation {
	93	+	sequence := runtimeMarkSequence.Add(1)
	94	+	return RuntimeCorrelation{Scenario: "runtime", Run: 1, Sequence: sequence, RequestID: sequence, Epoch: sequence}
	95	+	}
	96	+
	97	+	// NewRuntimeMarkWithCorrelation returns a timestamp-free mark carrying the
	98	+	// supplied operation identity. Start/end callers must reuse correlation.
	99	+	func NewRuntimeMarkWithCorrelation(component string, correlation RuntimeCorrelation, kind RuntimeMarkKind, bytes uint64, valid bool) RuntimeMark {
	100	+	return RuntimeMark{Schema: RuntimeMarkSchema, Component: component, Scenario: correlation.Scenario, Run: correlation.Run, Sequence: correlation.Sequence, RequestID: correlation.RequestID, Epoch: correlation.Epoch, Kind: kind, Bytes: bytes, Valid: valid}
	101	+	}
	102	+
	103	+	// NewRuntimeMark is retained for one-shot marks. Unlike the old helper it
	104	+	// never returns constant correlation IDs.
80	105		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}
	106	+	return NewRuntimeMarkWithCorrelation(component, NewRuntimeCorrelation(), kind, bytes, valid)
82	107		}

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

@@ -75,8 +77,31 @@ func RuntimeMarkValid(mark RuntimeMark) bool {
75		}	77		}
76		}	78		}
77			79
78	-	// NewRuntimeMark returns a timestamp-free mark for lightweight opt-in hooks.	80	+	// RuntimeCorrelation identifies one attempted operation. It is deliberately
79	-	// Callers that need a span pass the same correlation fields to both endpoints.	81	+	// copied to both endpoints of a span; process ID remains owned by the sink.
			82	+	type RuntimeCorrelation struct {
			83	+	Scenario string
			84	+	Run, Sequence, RequestID, Epoch uint64
			85	+	}
			86	+
			87	+	var runtimeMarkSequence atomic.Uint64
			88	+
			89	+	// NewRuntimeCorrelation makes a unique, process-local operation identity.
			90	+	// Components with a domain request/epoch should use those values directly via
			91	+	// NewRuntimeMarkWithCorrelation instead of inventing a second identity.
			92	+	func NewRuntimeCorrelation() RuntimeCorrelation {
			93	+	sequence := runtimeMarkSequence.Add(1)
			94	+	return RuntimeCorrelation{Scenario: "runtime", Run: 1, Sequence: sequence, RequestID: sequence, Epoch: sequence}
			95	+	}
			96	+
			97	+	// NewRuntimeMarkWithCorrelation returns a timestamp-free mark carrying the
			98	+	// supplied operation identity. Start/end callers must reuse correlation.
			99	+	func NewRuntimeMarkWithCorrelation(component string, correlation RuntimeCorrelation, kind RuntimeMarkKind, bytes uint64, valid bool) RuntimeMark {
			100	+	return RuntimeMark{Schema: RuntimeMarkSchema, Component: component, Scenario: correlation.Scenario, Run: correlation.Run, Sequence: correlation.Sequence, RequestID: correlation.RequestID, Epoch: correlation.Epoch, Kind: kind, Bytes: bytes, Valid: valid}
			101	+	}
			102	+
			103	+	// NewRuntimeMark is retained for one-shot marks. Unlike the old helper it
			104	+	// never returns constant correlation IDs.
80		func NewRuntimeMark(component string, kind RuntimeMarkKind, bytes uint64, valid bool) RuntimeMark {	105		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}	106	+	return NewRuntimeMarkWithCorrelation(component, NewRuntimeCorrelation(), kind, bytes, valid)
82		}	107		}

M internal/usecase/client/client.go +0 -4


@@ -513,10 +513,6 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi
513	513		log.Warn("querying terminal colors", "err", err)
514	514		}
515	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		-	}
520	516		if r.err != nil {
521	517		if errors.Is(r.err, io.EOF) {
522	518		return welcomedResult(fmt.Errorf("vev: daemon vanished (missing: %v): %w", ms.missing(), r.err))


@@ -513,10 +513,6 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi
513		log.Warn("querying terminal colors", "err", err)	513	log.Warn("querying terminal colors", "err", err)
514		}	514	}
515		case r := <-recvCh:	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	-	}
520		if r.err != nil {	516	if r.err != nil {
521		if errors.Is(r.err, io.EOF) {	517	if errors.Is(r.err, io.EOF) {
522		return welcomedResult(fmt.Errorf("vev: daemon vanished (missing: %v): %w", ms.missing(), r.err))	518	return welcomedResult(fmt.Errorf("vev: daemon vanished (missing: %v): %w", ms.missing(), r.err))

M internal/usecase/client/transport_observability_test.go +3 -1


@@ -38,7 +38,9 @@ func TestTerminalFlushBoundaryTransportObservability(t *testing.T) {
38	38		require.NoError(t, err)
39	39		require.Equal(t, "unchanged-by-observer", out.String(), "observer must not alter terminal bytes")
40	40		require.Equal(t, int32(1), restores.Load())
41		-	observer.requireOrdered(t, ports.RuntimeAdapterReceiveStart, ports.RuntimeAdapterReceiveEnd, ports.RuntimeTerminalFlushed)
	41	+	// Carriage spans belong only to concrete adapters. The client owns the
	42	+	// post-successful-flush terminal boundary, before its ACK is queued.
	43	+	observer.requireOrdered(t, ports.RuntimeTerminalFlushed)
42	44		}
43	45
44	46		type clientRuntimeObserver struct {

 	require.NoError(t, err)
 	require.Equal(t, "unchanged-by-observer", out.String(), "observer must not alter terminal bytes")
 	require.Equal(t, int32(1), restores.Load())
-	observer.requireOrdered(t, ports.RuntimeAdapterReceiveStart, ports.RuntimeAdapterReceiveEnd, ports.RuntimeTerminalFlushed)
+	// Carriage spans belong only to concrete adapters. The client owns the
+	// post-successful-flush terminal boundary, before its ACK is queued.
+	observer.requireOrdered(t, ports.RuntimeTerminalFlushed)
 }
 type clientRuntimeObserver struct {

M internal/usecase/daemon/client.go +2 -1


@@ -395,7 +395,8 @@ func (d Daemon) attachCoordinator(sess session, old, current *attachedClient,
395	395		rc := sess.renderCoordinator()
396	396		if rc == nil {
397	397		rc = newRenderCoordinator(renderCoordinatorOptions{
398		-	clock: d.clock,
	398	+	clock: d.clock,
	399	+	observer: d.runtimeObserver,
399	400		wake: func(w renderWake) {
400	401		// Never reread sess.client here: a wake is bound to both its
401	402		// attachment and its coordinator incarnation. The second check in

 	rc := sess.renderCoordinator()
 	if rc == nil {
 		rc = newRenderCoordinator(renderCoordinatorOptions{
-			clock: d.clock,
+			clock:    d.clock,
+			observer: d.runtimeObserver,
 			wake: func(w renderWake) {
 				// Never reread sess.client here: a wake is bound to both its
 				// attachment and its coordinator incarnation. The second check in

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


@@ -200,6 +200,19 @@ func (d *Daemon) observeRuntime(kind ports.RuntimeMarkKind, bytes uint64, valid
200	200		}
201	201		}
202	202
	203	+	// runtimeSpan emits endpoints for one real operation. The correlation is
	204	+	// allocated once before work begins and reused even when the operation fails.
	205	+	func (d *Daemon) runtimeSpan(start, end ports.RuntimeMarkKind, bytes uint64) func(uint64, bool) {
	206	+	if d == nil \|\| d.runtimeObserver == nil {
	207	+	return func(uint64, bool) {}
	208	+	}
	209	+	correlation := ports.NewRuntimeCorrelation()
	210	+	d.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", correlation, start, bytes, true))
	211	+	return func(endBytes uint64, valid bool) {
	212	+	d.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", correlation, end, endBytes, valid))
	213	+	}
	214	+	}
	215	+
203	216		func (d Daemon) paint(sess session, ac *attachedClient, reset bool, epochs ...uint64) {
204	217		attachmentEpoch := uint64(0)
205	218		if len(epochs) != 0 {

@@ -225,8 +238,8 @@ func (d Daemon) paint(sess session, ac *attachedClient, reset bool, epochs ...
225	238		// the normal gate, but this ownership check also protects direct resize and
226	239		// test paint paths from consuming damage that cannot be emitted.
227	240		if ac.output != nil && ac.output.atCapacity() {
228		-	d.observeRuntime(ports.RuntimeACKBlockedStart, 0, false)
229		-	d.observeRuntime(ports.RuntimeACKBlockedEnd, 0, false)
	241	+	// The coordinator owns the blocked interval; this guard only protects
	242	+	// direct test/resize paint calls from destructive capture.
230	243		ac.sendMu.Unlock()
231	244		return
232	245		}

@@ -279,9 +292,9 @@ func (d Daemon) paint(sess session, ac *attachedClient, reset bool, epochs ...
279	292		pickerActive: overlays.pickerActive, paletteActive: overlays.paletteActive, promptActive: overlays.promptActive,
280	293		copyFeedback: overlays.copyFeedback,
281	294		}
282		-	d.observeRuntime(ports.RuntimeCaptureStart, 0, true)
	295	+	endCapture := d.runtimeSpan(ports.RuntimeCaptureStart, ports.RuntimeCaptureEnd, 0)
283	296		state, ok := captureRenderState(sess, ac, bars, capturedOverlays, preview, floatingCfg, reset, damageCaptureConsume)
284		-	d.observeRuntime(ports.RuntimeCaptureEnd, 0, ok)
	297	+	endCapture(0, ok)
285	298		if !ok {
286	299		ac.sendMu.Unlock()
287	300		return

@@ -296,19 +309,13 @@ func (d Daemon) paint(sess session, ac *attachedClient, reset bool, epochs ...
296	309		state.preview = pickerPreviewFromCapturedRender(previewState)
297	310		}
298	311		captureOverlayLayers(state, overlays, paletteCfg)
299		-	d.observeRuntime(ports.RuntimeComposeStart, 0, true)
	312	+	endCompose := d.runtimeSpan(ports.RuntimeComposeStart, ports.RuntimeComposeEnd, 0)
300	313		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)
	314	+	endCompose(0, true)
306	315		if ac.renderStages.compose != nil {
307	316		ac.renderStages.compose()
308	317		}
309		-	d.observeRuntime(ports.RuntimeEmitStart, 0, true)
310	318		d.emitFrame(sess, ac, state, composed)
311		-	d.observeRuntime(ports.RuntimeEmitEnd, 0, true)
312	319		}
313	320
314	321		type themeStyles struct {

 	}
 }
+// runtimeSpan emits endpoints for one real operation. The correlation is
+// allocated once before work begins and reused even when the operation fails.
+func (d *Daemon) runtimeSpan(start, end ports.RuntimeMarkKind, bytes uint64) func(uint64, bool) {
+	if d == nil || d.runtimeObserver == nil {
+		return func(uint64, bool) {}
+	}
+	correlation := ports.NewRuntimeCorrelation()
+	d.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", correlation, start, bytes, true))
+	return func(endBytes uint64, valid bool) {
+		d.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", correlation, end, endBytes, valid))
+	}
+}
 func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool, epochs ...uint64) {
 	attachmentEpoch := uint64(0)
 	if len(epochs) != 0 {
 	// the normal gate, but this ownership check also protects direct resize and
 	// test paint paths from consuming damage that cannot be emitted.
 	if ac.output != nil && ac.output.atCapacity() {
-		d.observeRuntime(ports.RuntimeACKBlockedStart, 0, false)
+		// The coordinator owns the blocked interval; this guard only protects
-		d.observeRuntime(ports.RuntimeACKBlockedEnd, 0, false)
+		// direct test/resize paint calls from destructive capture.
 		ac.sendMu.Unlock()
 		return
 	}
 		pickerActive: overlays.pickerActive, paletteActive: overlays.paletteActive, promptActive: overlays.promptActive,
 		copyFeedback: overlays.copyFeedback,
 	}
-	d.observeRuntime(ports.RuntimeCaptureStart, 0, true)
+	endCapture := d.runtimeSpan(ports.RuntimeCaptureStart, ports.RuntimeCaptureEnd, 0)
 	state, ok := captureRenderState(sess, ac, bars, capturedOverlays, preview, floatingCfg, reset, damageCaptureConsume)
-	d.observeRuntime(ports.RuntimeCaptureEnd, 0, ok)
+	endCapture(0, ok)
 	if !ok {
 		ac.sendMu.Unlock()
 		return
 		state.preview = pickerPreviewFromCapturedRender(previewState)
 	}
 	captureOverlayLayers(state, overlays, paletteCfg)
-	d.observeRuntime(ports.RuntimeComposeStart, 0, true)
+	endCompose := d.runtimeSpan(ports.RuntimeComposeStart, ports.RuntimeComposeEnd, 0)
 	composed := composeFrame(*state, ac.pipelineCache, ac.pipelineScratch)
-	d.observeRuntime(ports.RuntimeComposeEnd, 0, true)
+	endCompose(0, true)
-	d.observeRuntime(ports.RuntimeDiffStart, 0, true)
-	d.observeRuntime(ports.RuntimeDiffEnd, 0, true)
-	d.observeRuntime(ports.RuntimeQueueEnqueued, 0, true)
-	d.observeRuntime(ports.RuntimeQueueDequeued, 0, true)
 	if ac.renderStages.compose != nil {
 		ac.renderStages.compose()
 	}
-	d.observeRuntime(ports.RuntimeEmitStart, 0, true)
 	d.emitFrame(sess, ac, state, composed)
-	d.observeRuntime(ports.RuntimeEmitEnd, 0, true)
 }
 type themeStyles struct {

M internal/usecase/daemon/render_coordinator.go +64 -1


@@ -78,7 +78,8 @@ type resizeRequestMetadata struct {
78	78
79	79		// renderCoordinatorOptions wires one coordinator instance.
80	80		type renderCoordinatorOptions struct {
81		-	clock ports.Clock
	81	+	clock ports.Clock
	82	+	observer ports.RuntimeObserver
82	83		// wake is the transitional composition target.
83	84		wake func(renderWake)
84	85		// ackReady reports whether the attachment may compose another output

@@ -124,6 +125,10 @@ type renderCoordinator struct {
124	125		pending bool
125	126		pendingReset bool
126	127		pendingUrgent bool
	128	+	// queueMarked spans the coordinator's actual pending interval, from first
	129	+	// enqueue until fire consumes the coalesced work.
	130	+	queueMarked bool
	131	+	queueCorrelation ports.RuntimeCorrelation
127	132		// pendingPreview is accounted separately from the target primary wake:
128	133		// viewers may consume a coalesced target snapshot while that target is
129	134		// still ACK-blocked.

@@ -131,6 +136,10 @@ type renderCoordinator struct {
131	136		// ackDeferred records that an expired deadline or watchdog was blocked by
132	137		// output-window capacity. ACK notifications may flush only this state.
133	138		ackDeferred bool
	139	+	// ackBlocked retains one correlation from the first rejected ACK-capacity
	140	+	// probe until notifyAck releases that exact blocked interval.
	141	+	ackBlocked bool
	142	+	ackCorrelation ports.RuntimeCorrelation
134	143		// deadlineDue closes the handoff between a timer worker receiving its tick
135	144		// and fire observing unavailable ACK capacity.
136	145		deadlineDue bool

@@ -279,9 +288,17 @@ func (c renderCoordinator) invalidateForAttachment(source attachedClient, inv
279	288		onInvalidate := c.opts.onInvalidate
280	289		c.metrics.invalidations.Add(1)
281	290		wasPending, wasUrgent, wasPreviewPending := c.pending, c.pendingUrgent, c.pendingPreview
	291	+	queueStart := false
	292	+	var queueCorrelation ports.RuntimeCorrelation
282	293		if !wasPending {
283	294		c.ackDeferred = false
284	295		c.deadlineDue = false
	296	+	if c.opts.observer != nil {
	297	+	c.queueMarked = true
	298	+	c.queueCorrelation = ports.NewRuntimeCorrelation()
	299	+	queueCorrelation = c.queueCorrelation
	300	+	queueStart = true
	301	+	}
285	302		}
286	303		c.pending = true
287	304		c.pendingReset = c.pendingReset \|\| inv.reset

@@ -304,7 +321,11 @@ func (c renderCoordinator) invalidateForAttachment(source attachedClient, inv
304	321		}
305	322		clock := c.opts.clock
306	323		c.armed = c.armed \|\| arm
	324	+	observer := c.opts.observer
307	325		c.mu.Unlock()
	326	+	if queueStart && observer != nil {
	327	+	observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueEnqueued, 0, true))
	328	+	}
308	329		stopTimer(old)
309	330		if !arm \|\| clock == nil {
310	331		if onInvalidate != nil {

@@ -368,7 +389,12 @@ func (c *renderCoordinator) notifyAck() {
368	389		return
369	390		}
370	391		gen := c.generation
	392	+	observer, correlation, blocked := c.opts.observer, c.ackCorrelation, c.ackBlocked
	393	+	c.ackBlocked = false
371	394		c.mu.Unlock()
	395	+	if blocked && observer != nil {
	396	+	observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", correlation, ports.RuntimeACKBlockedEnd, 0, true))
	397	+	}
372	398		c.fire(gen, false, false)
373	399		}
374	400

@@ -652,10 +678,19 @@ func (c renderCoordinator) noteDetach(ac attachedClient) {
652	678		retryTimer, c.retryTimer = c.retryTimer, nil
653	679		c.retryGen++
654	680		}
	681	+	observer, queueCorrelation, queueMarked := c.opts.observer, c.queueCorrelation, c.queueMarked
	682	+	ackCorrelation, ackBlocked := c.ackCorrelation, c.ackBlocked
	683	+	c.queueMarked, c.ackBlocked = false, false
655	684		c.mu.Unlock()
656	685		stopTimer(timer)
657	686		stopTimer(resizeTimer)
658	687		stopTimer(retryTimer)
	688	+	if queueMarked && observer != nil {
	689	+	observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, false))
	690	+	}
	691	+	if ackBlocked && observer != nil {
	692	+	observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, false))
	693	+	}
659	694		}
660	695
661	696		// noteReplace hands the coordinator from old to replacement; callbacks

@@ -700,10 +735,19 @@ func (c renderCoordinator) noteReplace(old, replacement attachedClient, readin
700	735		retryTimer, c.retryTimer = c.retryTimer, nil
701	736		c.retryGen++
702	737		}
	738	+	observer, queueCorrelation, queueMarked := c.opts.observer, c.queueCorrelation, c.queueMarked
	739	+	ackCorrelation, ackBlocked := c.ackCorrelation, c.ackBlocked
	740	+	c.queueMarked, c.ackBlocked = false, false
703	741		c.mu.Unlock()
704	742		stopTimer(timer)
705	743		stopTimer(resizeTimer)
706	744		stopTimer(retryTimer)
	745	+	if queueMarked && observer != nil {
	746	+	observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, false))
	747	+	}
	748	+	if ackBlocked && observer != nil {
	749	+	observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, false))
	750	+	}
707	751		}
708	752
709	753		// notePark invalidates pending wakes when the attachment parks for resume.

@@ -738,7 +782,16 @@ func (c *renderCoordinator) noteSessionTeardown() {
738	782		c.armed = false
739	783		timer := c.detachNormalTimerLocked()
740	784		timers := c.detachSyncTimersLocked()
	785	+	observer, queueCorrelation, queueMarked := c.opts.observer, c.queueCorrelation, c.queueMarked
	786	+	ackCorrelation, ackBlocked := c.ackCorrelation, c.ackBlocked
	787	+	c.queueMarked, c.ackBlocked = false, false
741	788		c.mu.Unlock()
	789	+	if queueMarked && observer != nil {
	790	+	observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, false))
	791	+	}
	792	+	if ackBlocked && observer != nil {
	793	+	observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, false))
	794	+	}
742	795		stopTimer(timer)
743	796		stopTimer(resizeTimer)
744	797		stopTimer(retryTimer)

@@ -1045,6 +1098,11 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) {
1045	1098		if !headlessPreviewOnly && (!ready \|\| (c.attachment != nil && !c.attachmentReady)) {
1046	1099		if !ready && deadline {
1047	1100		c.ackDeferred = true
	1101	+	if !c.ackBlocked && c.opts.observer != nil {
	1102	+	c.ackCorrelation = ports.NewRuntimeCorrelation()
	1103	+	c.ackBlocked = true
	1104	+	c.opts.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", c.ackCorrelation, ports.RuntimeACKBlockedStart, 0, true))
	1105	+	}
1048	1106		}
1049	1107		preview, previews := c.takePendingPreviewsLocked()
1050	1108		c.mu.Unlock()

@@ -1066,9 +1124,14 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) {
1066	1124		}
1067	1125		timer := c.detachNormalTimerLocked()
1068	1126		c.pending, c.pendingReset, c.pendingUrgent, c.ackDeferred, c.deadlineDue, c.coalesced, c.armed = false, false, false, false, false, 0, false
	1127	+	observer, queueCorrelation, queueMarked := c.opts.observer, c.queueCorrelation, c.queueMarked
	1128	+	c.queueMarked = false
1069	1129		wake := c.opts.wake
1070	1130		c.mu.Unlock()
1071	1131		stopTimer(timer)
	1132	+	if queueMarked && observer != nil {
	1133	+	observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, true))
	1134	+	}
1072	1135		if wake != nil && !headlessPreviewOnly {
1073	1136		wake(w)
1074	1137		}

 // renderCoordinatorOptions wires one coordinator instance.
 type renderCoordinatorOptions struct {
-	clock ports.Clock
+	clock    ports.Clock
+	observer ports.RuntimeObserver
 	// wake is the transitional composition target.
 	wake func(renderWake)
 	// ackReady reports whether the attachment may compose another output
 	pending       bool
 	pendingReset  bool
 	pendingUrgent bool
+	// queueMarked spans the coordinator's actual pending interval, from first
+	// enqueue until fire consumes the coalesced work.
+	queueMarked      bool
+	queueCorrelation ports.RuntimeCorrelation
 	// pendingPreview is accounted separately from the target primary wake:
 	// viewers may consume a coalesced target snapshot while that target is
 	// still ACK-blocked.
 	// ackDeferred records that an expired deadline or watchdog was blocked by
 	// output-window capacity. ACK notifications may flush only this state.
 	ackDeferred bool
+	// ackBlocked retains one correlation from the first rejected ACK-capacity
+	// probe until notifyAck releases that exact blocked interval.
+	ackBlocked     bool
+	ackCorrelation ports.RuntimeCorrelation
 	// deadlineDue closes the handoff between a timer worker receiving its tick
 	// and fire observing unavailable ACK capacity.
 	deadlineDue bool
 	onInvalidate := c.opts.onInvalidate
 	c.metrics.invalidations.Add(1)
 	wasPending, wasUrgent, wasPreviewPending := c.pending, c.pendingUrgent, c.pendingPreview
+	queueStart := false
+	var queueCorrelation ports.RuntimeCorrelation
 	if !wasPending {
 		c.ackDeferred = false
 		c.deadlineDue = false
+		if c.opts.observer != nil {
+			c.queueMarked = true
+			c.queueCorrelation = ports.NewRuntimeCorrelation()
+			queueCorrelation = c.queueCorrelation
+			queueStart = true
+		}
 	}
 	c.pending = true
 	c.pendingReset = c.pendingReset || inv.reset
 	}
 	clock := c.opts.clock
 	c.armed = c.armed || arm
+	observer := c.opts.observer
 	c.mu.Unlock()
+	if queueStart && observer != nil {
+		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueEnqueued, 0, true))
+	}
 	stopTimer(old)
 	if !arm || clock == nil {
 		if onInvalidate != nil {
 		return
 	}
 	gen := c.generation
+	observer, correlation, blocked := c.opts.observer, c.ackCorrelation, c.ackBlocked
+	c.ackBlocked = false
 	c.mu.Unlock()
+	if blocked && observer != nil {
+		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", correlation, ports.RuntimeACKBlockedEnd, 0, true))
+	}
 	c.fire(gen, false, false)
 }
 		retryTimer, c.retryTimer = c.retryTimer, nil
 		c.retryGen++
 	}
+	observer, queueCorrelation, queueMarked := c.opts.observer, c.queueCorrelation, c.queueMarked
+	ackCorrelation, ackBlocked := c.ackCorrelation, c.ackBlocked
+	c.queueMarked, c.ackBlocked = false, false
 	c.mu.Unlock()
 	stopTimer(timer)
 	stopTimer(resizeTimer)
 	stopTimer(retryTimer)
+	if queueMarked && observer != nil {
+		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, false))
+	}
+	if ackBlocked && observer != nil {
+		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, false))
+	}
 }
 // noteReplace hands the coordinator from old to replacement; callbacks
 		retryTimer, c.retryTimer = c.retryTimer, nil
 		c.retryGen++
 	}
+	observer, queueCorrelation, queueMarked := c.opts.observer, c.queueCorrelation, c.queueMarked
+	ackCorrelation, ackBlocked := c.ackCorrelation, c.ackBlocked
+	c.queueMarked, c.ackBlocked = false, false
 	c.mu.Unlock()
 	stopTimer(timer)
 	stopTimer(resizeTimer)
 	stopTimer(retryTimer)
+	if queueMarked && observer != nil {
+		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, false))
+	}
+	if ackBlocked && observer != nil {
+		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, false))
+	}
 }
 // notePark invalidates pending wakes when the attachment parks for resume.
 	c.armed = false
 	timer := c.detachNormalTimerLocked()
 	timers := c.detachSyncTimersLocked()
+	observer, queueCorrelation, queueMarked := c.opts.observer, c.queueCorrelation, c.queueMarked
+	ackCorrelation, ackBlocked := c.ackCorrelation, c.ackBlocked
+	c.queueMarked, c.ackBlocked = false, false
 	c.mu.Unlock()
+	if queueMarked && observer != nil {
+		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, false))
+	}
+	if ackBlocked && observer != nil {
+		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, false))
+	}
 	stopTimer(timer)
 	stopTimer(resizeTimer)
 	stopTimer(retryTimer)
 	if !headlessPreviewOnly && (!ready || (c.attachment != nil && !c.attachmentReady)) {
 		if !ready && deadline {
 			c.ackDeferred = true
+			if !c.ackBlocked && c.opts.observer != nil {
+				c.ackCorrelation = ports.NewRuntimeCorrelation()
+				c.ackBlocked = true
+				c.opts.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", c.ackCorrelation, ports.RuntimeACKBlockedStart, 0, true))
+			}
 		}
 		preview, previews := c.takePendingPreviewsLocked()
 		c.mu.Unlock()
 	}
 	timer := c.detachNormalTimerLocked()
 	c.pending, c.pendingReset, c.pendingUrgent, c.ackDeferred, c.deadlineDue, c.coalesced, c.armed = false, false, false, false, false, 0, false
+	observer, queueCorrelation, queueMarked := c.opts.observer, c.queueCorrelation, c.queueMarked
+	c.queueMarked = false
 	wake := c.opts.wake
 	c.mu.Unlock()
 	stopTimer(timer)
+	if queueMarked && observer != nil {
+		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, true))
+	}
 	if wake != nil && !headlessPreviewOnly {
 		wake(w)
 	}

M internal/usecase/daemon/render_pipeline.go +4 -0


@@ -306,7 +306,9 @@ func (d Daemon) emitFrame(sess session, ac attachedClient, state capturedRen
306	306		ac.sendMu.Unlock()
307	307		return false
308	308		}
	309	+	endDiff := d.runtimeSpan(ports.RuntimeDiffStart, ports.RuntimeDiffEnd, 0)
309	310		prepared, err := ac.output.prepare(composed.frame, composed.damage, composed.reset)
	311	+	endDiff(0, err == nil)
310	312		if err != nil {
311	313		ac.sess.mu.Unlock()
312	314		ac.sendMu.Unlock()

@@ -319,6 +321,7 @@ func (d Daemon) emitFrame(sess session, ac attachedClient, state capturedRen
319	321		var sendErr error
320	322		if len(data) > 0 {
321	323		sendTr = ac.transport()
	324	+	endEmit := d.runtimeSpan(ports.RuntimeEmitStart, ports.RuntimeEmitEnd, uint64(len(data)))
322	325		if sendTr == nil {
323	326		sendErr = errors.New("client transport is nil")
324	327		} else {

@@ -328,6 +331,7 @@ func (d Daemon) emitFrame(sess session, ac attachedClient, state capturedRen
328	331		}
329	332		sendErr = prepared.send(data, ac.echoAck.Load(), send)
330	333		}
	334	+	endEmit(uint64(len(data)), sendErr == nil)
331	335		}
332	336		if sendErr == nil {
333	337		// Publish only after output preparation and transport emission both


@@ -306,7 +306,9 @@ func (d Daemon) emitFrame(sess session, ac attachedClient, state capturedRen
306	ac.sendMu.Unlock()	306		ac.sendMu.Unlock()
307	return false	307		return false
308	}	308		}
		309	+	endDiff := d.runtimeSpan(ports.RuntimeDiffStart, ports.RuntimeDiffEnd, 0)
309	prepared, err := ac.output.prepare(composed.frame, composed.damage, composed.reset)	310		prepared, err := ac.output.prepare(composed.frame, composed.damage, composed.reset)
		311	+	endDiff(0, err == nil)
310	if err != nil {	312		if err != nil {
311	ac.sess.mu.Unlock()	313		ac.sess.mu.Unlock()
312	ac.sendMu.Unlock()	314		ac.sendMu.Unlock()

@@ -319,6 +321,7 @@ func (d Daemon) emitFrame(sess session, ac attachedClient, state capturedRen
319	var sendErr error	321		var sendErr error
320	if len(data) > 0 {	322		if len(data) > 0 {
321	sendTr = ac.transport()	323		sendTr = ac.transport()
		324	+	endEmit := d.runtimeSpan(ports.RuntimeEmitStart, ports.RuntimeEmitEnd, uint64(len(data)))
322	if sendTr == nil {	325		if sendTr == nil {
323	sendErr = errors.New("client transport is nil")	326		sendErr = errors.New("client transport is nil")
324	} else {	327		} else {

@@ -328,6 +331,7 @@ func (d Daemon) emitFrame(sess session, ac attachedClient, state capturedRen
328	}	331		}
329	sendErr = prepared.send(data, ac.echoAck.Load(), send)	332		sendErr = prepared.send(data, ac.echoAck.Load(), send)
330	}	333		}
		334	+	endEmit(uint64(len(data)), sendErr == nil)
331	}	335		}
332	if sendErr == nil {	336		if sendErr == nil {
333	// Publish only after output preparation and transport emission both	337		// Publish only after output preparation and transport emission both