fix(obs): close failed adapter spans

commit: e99dffb19a7f262b0bd960cbb80b3dc1b4619ea2
author: brice · 3mo ago · 2026-07-12 23:12 UTC
committer: brice · 3mo ago · 2026-07-12 23:16 UTC
parent: 1cb3d81
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +487 -11

M internal/adapters/dgram/transport.go +46 -1


@@ -177,6 +177,10 @@ type Transport struct {
177	177		afterLinkStateCommit func(ports.LinkState)
178	178		observe DiagnosticObserver
179	179		runtimeObserver ports.RuntimeObserver
	180	+	operationMu sync.Mutex
	181	+	operationCount int
	182	+	closing bool
	183	+	operationsDone chan struct{}
180	184		diagnosticCh chan Diagnostic
181	185		heartbeat time.Duration
182	186		resendAfter time.Duration

@@ -611,11 +615,16 @@ func (t *Transport) Recv() (ports.Frame, error) {
611	615		}
612	616		}
613	617		func (t *Transport) Close() error {
	618	+	done := t.beginShutdown()
614	619		t.closeWithError(errors.New("dgram: closed"))
615	620		t.mu.Lock()
616	621		pc := t.pc
617	622		t.mu.Unlock()
618		-	return pc.Close()
	623	+	err := pc.Close()
	624	+	if done != nil {
	625	+	<-done
	626	+	}
	627	+	return err
619	628		}
620	629
621	630		func (t *Transport) Peer() net.Addr { t.mu.Lock(); defer t.mu.Unlock(); return t.peer }

@@ -1618,6 +1627,9 @@ func (t *Transport) beginRuntimeOperation(start ports.RuntimeMarkKind, bytes uin
1618	1627		if t.runtimeObserver == nil {
1619	1628		return func(bool) {}
1620	1629		}
	1630	+	if !t.beginObservedOperation() {
	1631	+	return func(bool) {}
	1632	+	}
1621	1633		correlation := ports.NewRuntimeCorrelation()
1622	1634		t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, start, bytes, true))
1623	1635		end := ports.RuntimeAdapterSendEnd

@@ -1625,8 +1637,41 @@ func (t *Transport) beginRuntimeOperation(start ports.RuntimeMarkKind, bytes uin
1625	1637		end = ports.RuntimeAdapterReceiveEnd
1626	1638		}
1627	1639		return func(valid bool) {
	1640	+	defer t.finishObservedOperation()
1628	1641		t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, end, bytes, valid))
1629	1642		}
1630	1643		}
1631	1644
	1645	+	func (t *Transport) beginObservedOperation() bool {
	1646	+	t.operationMu.Lock()
	1647	+	defer t.operationMu.Unlock()
	1648	+	if t.closing {
	1649	+	return false
	1650	+	}
	1651	+	if t.operationCount == 0 {
	1652	+	t.operationsDone = make(chan struct{})
	1653	+	}
	1654	+	t.operationCount++
	1655	+	return true
	1656	+	}
	1657	+
	1658	+	func (t *Transport) finishObservedOperation() {
	1659	+	t.operationMu.Lock()
	1660	+	defer t.operationMu.Unlock()
	1661	+	t.operationCount--
	1662	+	if t.operationCount == 0 {
	1663	+	close(t.operationsDone)
	1664	+	}
	1665	+	}
	1666	+
	1667	+	func (t *Transport) beginShutdown() <-chan struct{} {
	1668	+	t.operationMu.Lock()
	1669	+	defer t.operationMu.Unlock()
	1670	+	t.closing = true
	1671	+	if t.operationCount == 0 {
	1672	+	return nil
	1673	+	}
	1674	+	return t.operationsDone
	1675	+	}
	1676	+
1632	1677		func (*Transport) DatagramTransport() {}


@@ -177,6 +177,10 @@ type Transport struct {
177		afterLinkStateCommit func(ports.LinkState)	177		afterLinkStateCommit func(ports.LinkState)
178		observe DiagnosticObserver	178		observe DiagnosticObserver
179		runtimeObserver ports.RuntimeObserver	179		runtimeObserver ports.RuntimeObserver
			180	+	operationMu sync.Mutex
			181	+	operationCount int
			182	+	closing bool
			183	+	operationsDone chan struct{}
180		diagnosticCh chan Diagnostic	184		diagnosticCh chan Diagnostic
181		heartbeat time.Duration	185		heartbeat time.Duration
182		resendAfter time.Duration	186		resendAfter time.Duration

@@ -611,11 +615,16 @@ func (t *Transport) Recv() (ports.Frame, error) {
611		}	615		}
612		}	616		}
613		func (t *Transport) Close() error {	617		func (t *Transport) Close() error {
			618	+	done := t.beginShutdown()
614		t.closeWithError(errors.New("dgram: closed"))	619		t.closeWithError(errors.New("dgram: closed"))
615		t.mu.Lock()	620		t.mu.Lock()
616		pc := t.pc	621		pc := t.pc
617		t.mu.Unlock()	622		t.mu.Unlock()
618	-	return pc.Close()	623	+	err := pc.Close()
			624	+	if done != nil {
			625	+	<-done
			626	+	}
			627	+	return err
619		}	628		}
620			629
621		func (t *Transport) Peer() net.Addr { t.mu.Lock(); defer t.mu.Unlock(); return t.peer }	630		func (t *Transport) Peer() net.Addr { t.mu.Lock(); defer t.mu.Unlock(); return t.peer }

@@ -1618,6 +1627,9 @@ func (t *Transport) beginRuntimeOperation(start ports.RuntimeMarkKind, bytes uin
1618		if t.runtimeObserver == nil {	1627		if t.runtimeObserver == nil {
1619		return func(bool) {}	1628		return func(bool) {}
1620		}	1629		}
			1630	+	if !t.beginObservedOperation() {
			1631	+	return func(bool) {}
			1632	+	}
1621		correlation := ports.NewRuntimeCorrelation()	1633		correlation := ports.NewRuntimeCorrelation()
1622		t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, start, bytes, true))	1634		t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, start, bytes, true))
1623		end := ports.RuntimeAdapterSendEnd	1635		end := ports.RuntimeAdapterSendEnd

@@ -1625,8 +1637,41 @@ func (t *Transport) beginRuntimeOperation(start ports.RuntimeMarkKind, bytes uin
1625		end = ports.RuntimeAdapterReceiveEnd	1637		end = ports.RuntimeAdapterReceiveEnd
1626		}	1638		}
1627		return func(valid bool) {	1639		return func(valid bool) {
			1640	+	defer t.finishObservedOperation()
1628		t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, end, bytes, valid))	1641		t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, end, bytes, valid))
1629		}	1642		}
1630		}	1643		}
1631			1644
			1645	+	func (t *Transport) beginObservedOperation() bool {
			1646	+	t.operationMu.Lock()
			1647	+	defer t.operationMu.Unlock()
			1648	+	if t.closing {
			1649	+	return false
			1650	+	}
			1651	+	if t.operationCount == 0 {
			1652	+	t.operationsDone = make(chan struct{})
			1653	+	}
			1654	+	t.operationCount++
			1655	+	return true
			1656	+	}
			1657	+
			1658	+	func (t *Transport) finishObservedOperation() {
			1659	+	t.operationMu.Lock()
			1660	+	defer t.operationMu.Unlock()
			1661	+	t.operationCount--
			1662	+	if t.operationCount == 0 {
			1663	+	close(t.operationsDone)
			1664	+	}
			1665	+	}
			1666	+
			1667	+	func (t *Transport) beginShutdown() <-chan struct{} {
			1668	+	t.operationMu.Lock()
			1669	+	defer t.operationMu.Unlock()
			1670	+	t.closing = true
			1671	+	if t.operationCount == 0 {
			1672	+	return nil
			1673	+	}
			1674	+	return t.operationsDone
			1675	+	}
			1676	+
1632		func (*Transport) DatagramTransport() {}	1677		func (*Transport) DatagramTransport() {}

M internal/adapters/dgram/transport_observability_test.go +108 -0

@@ -1,9 +1,13 @@
1	1		package dgram
2	2
3	3		import (
	4	+	"encoding/json"
	5	+	"errors"
	6	+	"net"
4	7		"os"
5	8		"reflect"
6	9		"strings"
	10	+	"sync"
7	11		"testing"
8	12		"time"
9	13

@@ -11,6 +15,110 @@ import (
11	15		"github.com/bnema/vev/internal/ports"
12	16		)
13	17
	18	+	func TestTransportObservabilityDgramFailedCloseEndsReceiveAndSend(t *testing.T) {
	19	+	behaviorClock := newManualClock(time.Unix(0, 417))
	20	+	link := newSimulatedLink(behaviorClock, packetPolicy{})
	21	+	aPC, _ := newPairWithCapacity(link, 64)
	22	+	closeErr := errors.New("packet connection close failed")
	23	+	tracePath := t.TempDir() + "/dgram-close.jsonl"
	24	+	observer, closer, err := observability.NewJSONL(tracePath, dgramObserverClock{now: time.Unix(0, 521)}, "dgram-process")
	25	+	if err != nil {
	26	+	t.Fatalf("NewJSONL() error = %v", err)
	27	+	}
	28	+	receiveStarted := make(chan struct{})
	29	+	signalingObserver := &dgramStartObserver{RuntimeObserver: observer, receiveStarted: receiveStarted}
	30	+	transport, err := NewTransportWithOptions(&dgramCloseFailPC{fakePC: aPC, err: closeErr}, testAddr("b"), key(), 1, 2, Options{Clock: behaviorClock, ResendAfter: time.Hour, Heartbeat: time.Hour}, WithRuntimeObserver(signalingObserver))
	31	+	if err != nil {
	32	+	t.Fatalf("NewTransportWithOptions() error = %v", err)
	33	+	}
	34	+	if err := transport.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("write failure")}); err == nil {
	35	+	t.Fatal("Send() error = nil on packet write failure")
	36	+	}
	37	+	recvDone := make(chan error, 1)
	38	+	go func() { _, err := transport.Recv(); recvDone <- err }()
	39	+	<-receiveStarted
	40	+	if err := transport.Close(); !errors.Is(err, closeErr) {
	41	+	t.Fatalf("Close() error = %v, want %v", err, closeErr)
	42	+	}
	43	+	if err := <-recvDone; err == nil {
	44	+	t.Fatal("Recv() error = nil after failed Close()")
	45	+	}
	46	+	if err := closer.Close(); err != nil {
	47	+	t.Fatalf("trace Close() error = %v", err)
	48	+	}
	49	+	assertDgramFailedSpanPairs(t, tracePath)
	50	+	}
	51	+
	52	+	type dgramStartObserver struct {
	53	+	ports.RuntimeObserver
	54	+	receiveStarted chan struct{}
	55	+	once sync.Once
	56	+	}
	57	+
	58	+	func (o *dgramStartObserver) ObserveRuntime(mark ports.RuntimeMark) {
	59	+	if mark.Kind == ports.RuntimeAdapterReceiveStart {
	60	+	o.once.Do(func() { close(o.receiveStarted) })
	61	+	}
	62	+	o.RuntimeObserver.ObserveRuntime(mark)
	63	+	}
	64	+
	65	+	type dgramCloseFailPC struct {
	66	+	*fakePC
	67	+	err error
	68	+	}
	69	+
	70	+	func (p *dgramCloseFailPC) WriteTo([]byte, net.Addr) (int, error) {
	71	+	return 0, errors.New("packet write failed")
	72	+	}
	73	+
	74	+	func (p *dgramCloseFailPC) Close() error {
	75	+	_ = p.fakePC.Close()
	76	+	return p.err
	77	+	}
	78	+
	79	+	func assertDgramFailedSpanPairs(t *testing.T, path string) {
	80	+	t.Helper()
	81	+	data, err := os.ReadFile(path)
	82	+	if err != nil {
	83	+	t.Fatalf("ReadFile(trace): %v", err)
	84	+	}
	85	+	type mark struct {
	86	+	ProcessID string `json:"process_id"`
	87	+	Scenario string `json:"scenario"`
	88	+	Run uint64 `json:"run"`
	89	+	Sequence uint64 `json:"sequence"`
	90	+	RequestID uint64 `json:"request_id"`
	91	+	Epoch uint64 `json:"epoch"`
	92	+	Kind string `json:"kind"`
	93	+	Valid bool `json:"valid"`
	94	+	}
	95	+	type spanKey struct {
	96	+	processID, scenario string
	97	+	run, sequence, requestID, epoch uint64
	98	+	}
	99	+	starts := make(map[spanKey]mark)
	100	+	for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
	101	+	var m mark
	102	+	if err := json.Unmarshal([]byte(line), &m); err != nil {
	103	+	t.Fatalf("Unmarshal(trace): %v", err)
	104	+	}
	105	+	key := spanKey{m.ProcessID, m.Scenario, m.Run, m.Sequence, m.RequestID, m.Epoch}
	106	+	switch m.Kind {
	107	+	case "adapter_receive_start", "adapter_send_start":
	108	+	starts[key] = m
	109	+	case "adapter_receive_end", "adapter_send_end":
	110	+	start, ok := starts[key]
	111	+	if !ok \|\| strings.TrimSuffix(start.Kind, "_start") != strings.TrimSuffix(m.Kind, "_end") \|\| m.Valid {
	112	+	t.Fatalf("unmatched or valid failed end: start=%#v end=%#v", start, m)
	113	+	}
	114	+	delete(starts, key)
	115	+	}
	116	+	}
	117	+	if len(starts) != 0 {
	118	+	t.Fatalf("unmatched adapter starts: %#v", starts)
	119	+	}
	120	+	}
	121	+
14	122		func TestTransportObservabilityDgramKeepsBehaviorClockSeparate(t *testing.T) {
15	123		behaviorClock := newManualClock(time.Unix(0, 401))
16	124		link := newSimulatedLink(behaviorClock, packetPolicy{})

@@ -1,9 +1,13 @@
1	package dgram	1		package dgram
2		2
3	import (	3		import (
		4	+	"encoding/json"
		5	+	"errors"
		6	+	"net"
4	"os"	7		"os"
5	"reflect"	8		"reflect"
6	"strings"	9		"strings"
		10	+	"sync"
7	"testing"	11		"testing"
8	"time"	12		"time"
9		13

@@ -11,6 +15,110 @@ import (
11	"github.com/bnema/vev/internal/ports"	15		"github.com/bnema/vev/internal/ports"
12	)	16		)
13		17
		18	+	func TestTransportObservabilityDgramFailedCloseEndsReceiveAndSend(t *testing.T) {
		19	+	behaviorClock := newManualClock(time.Unix(0, 417))
		20	+	link := newSimulatedLink(behaviorClock, packetPolicy{})
		21	+	aPC, _ := newPairWithCapacity(link, 64)
		22	+	closeErr := errors.New("packet connection close failed")
		23	+	tracePath := t.TempDir() + "/dgram-close.jsonl"
		24	+	observer, closer, err := observability.NewJSONL(tracePath, dgramObserverClock{now: time.Unix(0, 521)}, "dgram-process")
		25	+	if err != nil {
		26	+	t.Fatalf("NewJSONL() error = %v", err)
		27	+	}
		28	+	receiveStarted := make(chan struct{})
		29	+	signalingObserver := &dgramStartObserver{RuntimeObserver: observer, receiveStarted: receiveStarted}
		30	+	transport, err := NewTransportWithOptions(&dgramCloseFailPC{fakePC: aPC, err: closeErr}, testAddr("b"), key(), 1, 2, Options{Clock: behaviorClock, ResendAfter: time.Hour, Heartbeat: time.Hour}, WithRuntimeObserver(signalingObserver))
		31	+	if err != nil {
		32	+	t.Fatalf("NewTransportWithOptions() error = %v", err)
		33	+	}
		34	+	if err := transport.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("write failure")}); err == nil {
		35	+	t.Fatal("Send() error = nil on packet write failure")
		36	+	}
		37	+	recvDone := make(chan error, 1)
		38	+	go func() { _, err := transport.Recv(); recvDone <- err }()
		39	+	<-receiveStarted
		40	+	if err := transport.Close(); !errors.Is(err, closeErr) {
		41	+	t.Fatalf("Close() error = %v, want %v", err, closeErr)
		42	+	}
		43	+	if err := <-recvDone; err == nil {
		44	+	t.Fatal("Recv() error = nil after failed Close()")
		45	+	}
		46	+	if err := closer.Close(); err != nil {
		47	+	t.Fatalf("trace Close() error = %v", err)
		48	+	}
		49	+	assertDgramFailedSpanPairs(t, tracePath)
		50	+	}
		51	+
		52	+	type dgramStartObserver struct {
		53	+	ports.RuntimeObserver
		54	+	receiveStarted chan struct{}
		55	+	once sync.Once
		56	+	}
		57	+
		58	+	func (o *dgramStartObserver) ObserveRuntime(mark ports.RuntimeMark) {
		59	+	if mark.Kind == ports.RuntimeAdapterReceiveStart {
		60	+	o.once.Do(func() { close(o.receiveStarted) })
		61	+	}
		62	+	o.RuntimeObserver.ObserveRuntime(mark)
		63	+	}
		64	+
		65	+	type dgramCloseFailPC struct {
		66	+	*fakePC
		67	+	err error
		68	+	}
		69	+
		70	+	func (p *dgramCloseFailPC) WriteTo([]byte, net.Addr) (int, error) {
		71	+	return 0, errors.New("packet write failed")
		72	+	}
		73	+
		74	+	func (p *dgramCloseFailPC) Close() error {
		75	+	_ = p.fakePC.Close()
		76	+	return p.err
		77	+	}
		78	+
		79	+	func assertDgramFailedSpanPairs(t *testing.T, path string) {
		80	+	t.Helper()
		81	+	data, err := os.ReadFile(path)
		82	+	if err != nil {
		83	+	t.Fatalf("ReadFile(trace): %v", err)
		84	+	}
		85	+	type mark struct {
		86	+	ProcessID string `json:"process_id"`
		87	+	Scenario string `json:"scenario"`
		88	+	Run uint64 `json:"run"`
		89	+	Sequence uint64 `json:"sequence"`
		90	+	RequestID uint64 `json:"request_id"`
		91	+	Epoch uint64 `json:"epoch"`
		92	+	Kind string `json:"kind"`
		93	+	Valid bool `json:"valid"`
		94	+	}
		95	+	type spanKey struct {
		96	+	processID, scenario string
		97	+	run, sequence, requestID, epoch uint64
		98	+	}
		99	+	starts := make(map[spanKey]mark)
		100	+	for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
		101	+	var m mark
		102	+	if err := json.Unmarshal([]byte(line), &m); err != nil {
		103	+	t.Fatalf("Unmarshal(trace): %v", err)
		104	+	}
		105	+	key := spanKey{m.ProcessID, m.Scenario, m.Run, m.Sequence, m.RequestID, m.Epoch}
		106	+	switch m.Kind {
		107	+	case "adapter_receive_start", "adapter_send_start":
		108	+	starts[key] = m
		109	+	case "adapter_receive_end", "adapter_send_end":
		110	+	start, ok := starts[key]
		111	+	if !ok \|\| strings.TrimSuffix(start.Kind, "_start") != strings.TrimSuffix(m.Kind, "_end") \|\| m.Valid {
		112	+	t.Fatalf("unmatched or valid failed end: start=%#v end=%#v", start, m)
		113	+	}
		114	+	delete(starts, key)
		115	+	}
		116	+	}
		117	+	if len(starts) != 0 {
		118	+	t.Fatalf("unmatched adapter starts: %#v", starts)
		119	+	}
		120	+	}
		121	+
14	func TestTransportObservabilityDgramKeepsBehaviorClockSeparate(t *testing.T) {	122		func TestTransportObservabilityDgramKeepsBehaviorClockSeparate(t *testing.T) {
15	behaviorClock := newManualClock(time.Unix(0, 401))	123		behaviorClock := newManualClock(time.Unix(0, 401))
16	link := newSimulatedLink(behaviorClock, packetPolicy{})	124		link := newSimulatedLink(behaviorClock, packetPolicy{})

M internal/adapters/ipc/transport.go +51 -4


@@ -5,6 +5,7 @@ import (
5	5		"errors"
6	6		"io"
7	7		"net"
	8	+	"sync"
8	9
9	10		"github.com/bnema/vev/internal/ports"
10	11		)

@@ -44,8 +45,12 @@ func WithRuntimeObserver(observer ports.RuntimeObserver) Option {
44	45		}
45	46
46	47		type unixTransport struct {
47		-	conn net.Conn
48		-	observer ports.RuntimeObserver
	48	+	conn net.Conn
	49	+	observer ports.RuntimeObserver
	50	+	operationMu sync.Mutex
	51	+	operationCount int
	52	+	closing bool
	53	+	operationsDone chan struct{}
49	54
50	55		// readBuf is reused across Recv calls (grow-once strategy): it grows
51	56		// to fit the largest frame seen so far and is never shrunk. The Frame

@@ -131,6 +136,9 @@ func (t *unixTransport) beginOperation(start ports.RuntimeMarkKind, bytes uint64
131	136		if t.observer == nil {
132	137		return func(bool) {}
133	138		}
	139	+	if !t.beginObservedOperation() {
	140	+	return func(bool) {}
	141	+	}
134	142		correlation := ports.NewRuntimeCorrelation()
135	143		t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("ipc", correlation, start, bytes, true))
136	144		end := ports.RuntimeAdapterSendEnd

@@ -138,11 +146,50 @@ func (t *unixTransport) beginOperation(start ports.RuntimeMarkKind, bytes uint64
138	146		end = ports.RuntimeAdapterReceiveEnd
139	147		}
140	148		return func(valid bool) {
	149	+	defer t.finishObservedOperation()
141	150		t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("ipc", correlation, end, bytes, valid))
142	151		}
143	152		}
144	153
145		-	// Close closes the underlying connection.
	154	+	func (t *unixTransport) beginObservedOperation() bool {
	155	+	t.operationMu.Lock()
	156	+	defer t.operationMu.Unlock()
	157	+	if t.closing {
	158	+	return false
	159	+	}
	160	+	if t.operationCount == 0 {
	161	+	t.operationsDone = make(chan struct{})
	162	+	}
	163	+	t.operationCount++
	164	+	return true
	165	+	}
	166	+
	167	+	func (t *unixTransport) finishObservedOperation() {
	168	+	t.operationMu.Lock()
	169	+	defer t.operationMu.Unlock()
	170	+	t.operationCount--
	171	+	if t.operationCount == 0 {
	172	+	close(t.operationsDone)
	173	+	}
	174	+	}
	175	+
	176	+	func (t *unixTransport) beginShutdown() <-chan struct{} {
	177	+	t.operationMu.Lock()
	178	+	defer t.operationMu.Unlock()
	179	+	t.closing = true
	180	+	if t.operationCount == 0 {
	181	+	return nil
	182	+	}
	183	+	return t.operationsDone
	184	+	}
	185	+
	186	+	// Close closes the underlying connection and waits for any interrupted
	187	+	// carriage operation to write its matching failed end mark.
146	188		func (t *unixTransport) Close() error {
147		-	return t.conn.Close()
	189	+	done := t.beginShutdown()
	190	+	err := t.conn.Close()
	191	+	if done != nil {
	192	+	<-done
	193	+	}
	194	+	return err
148	195		}


@@ -5,6 +5,7 @@ import (
5		"errors"	5		"errors"
6		"io"	6		"io"
7		"net"	7		"net"
			8	+	"sync"
8			9
9		"github.com/bnema/vev/internal/ports"	10		"github.com/bnema/vev/internal/ports"
10		)	11		)

@@ -44,8 +45,12 @@ func WithRuntimeObserver(observer ports.RuntimeObserver) Option {
44		}	45		}
45			46
46		type unixTransport struct {	47		type unixTransport struct {
47	-	conn net.Conn	48	+	conn net.Conn
48	-	observer ports.RuntimeObserver	49	+	observer ports.RuntimeObserver
			50	+	operationMu sync.Mutex
			51	+	operationCount int
			52	+	closing bool
			53	+	operationsDone chan struct{}
49			54
50		// readBuf is reused across Recv calls (grow-once strategy): it grows	55		// readBuf is reused across Recv calls (grow-once strategy): it grows
51		// to fit the largest frame seen so far and is never shrunk. The Frame	56		// to fit the largest frame seen so far and is never shrunk. The Frame

@@ -131,6 +136,9 @@ func (t *unixTransport) beginOperation(start ports.RuntimeMarkKind, bytes uint64
131		if t.observer == nil {	136		if t.observer == nil {
132		return func(bool) {}	137		return func(bool) {}
133		}	138		}
			139	+	if !t.beginObservedOperation() {
			140	+	return func(bool) {}
			141	+	}
134		correlation := ports.NewRuntimeCorrelation()	142		correlation := ports.NewRuntimeCorrelation()
135		t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("ipc", correlation, start, bytes, true))	143		t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("ipc", correlation, start, bytes, true))
136		end := ports.RuntimeAdapterSendEnd	144		end := ports.RuntimeAdapterSendEnd

@@ -138,11 +146,50 @@ func (t *unixTransport) beginOperation(start ports.RuntimeMarkKind, bytes uint64
138		end = ports.RuntimeAdapterReceiveEnd	146		end = ports.RuntimeAdapterReceiveEnd
139		}	147		}
140		return func(valid bool) {	148		return func(valid bool) {
			149	+	defer t.finishObservedOperation()
141		t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("ipc", correlation, end, bytes, valid))	150		t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("ipc", correlation, end, bytes, valid))
142		}	151		}
143		}	152		}
144			153
145	-	// Close closes the underlying connection.	154	+	func (t *unixTransport) beginObservedOperation() bool {
			155	+	t.operationMu.Lock()
			156	+	defer t.operationMu.Unlock()
			157	+	if t.closing {
			158	+	return false
			159	+	}
			160	+	if t.operationCount == 0 {
			161	+	t.operationsDone = make(chan struct{})
			162	+	}
			163	+	t.operationCount++
			164	+	return true
			165	+	}
			166	+
			167	+	func (t *unixTransport) finishObservedOperation() {
			168	+	t.operationMu.Lock()
			169	+	defer t.operationMu.Unlock()
			170	+	t.operationCount--
			171	+	if t.operationCount == 0 {
			172	+	close(t.operationsDone)
			173	+	}
			174	+	}
			175	+
			176	+	func (t *unixTransport) beginShutdown() <-chan struct{} {
			177	+	t.operationMu.Lock()
			178	+	defer t.operationMu.Unlock()
			179	+	t.closing = true
			180	+	if t.operationCount == 0 {
			181	+	return nil
			182	+	}
			183	+	return t.operationsDone
			184	+	}
			185	+
			186	+	// Close closes the underlying connection and waits for any interrupted
			187	+	// carriage operation to write its matching failed end mark.
146		func (t *unixTransport) Close() error {	188		func (t *unixTransport) Close() error {
147	-	return t.conn.Close()	189	+	done := t.beginShutdown()
			190	+	err := t.conn.Close()
			191	+	if done != nil {
			192	+	<-done
			193	+	}
			194	+	return err
148		}	195		}

M internal/adapters/ipc/transport_observability_test.go +110 -0

@@ -1,6 +1,7 @@
1	1		package ipc
2	2
3	3		import (
	4	+	"encoding/json"
4	5		"net"
5	6		"os"
6	7		"reflect"

@@ -13,6 +14,115 @@ import (
13	14		"github.com/bnema/vev/internal/ports"
14	15		)
15	16
	17	+	func TestTransportObservabilityIPCEOFAndCloseEndFailedSpans(t *testing.T) {
	18	+	trace := t.TempDir() + "/ipc-shutdown.jsonl"
	19	+	observer, closer, err := observability.NewJSONL(trace, ipcRuntimeClock{now: time.Unix(0, 223)}, "ipc-process")
	20	+	if err != nil {
	21	+	t.Fatalf("NewJSONL() error = %v", err)
	22	+	}
	23	+
	24	+	// EOF from the peer and a locally closed blocked receive are distinct
	25	+	// shutdown paths; both must record a failed, correlated end.
	26	+	for _, closeLocal := range []bool{false, true} {
	27	+	left, right := net.Pipe()
	28	+	var started <-chan struct{}
	29	+	conn := left
	30	+	if closeLocal {
	31	+	observedConn := &ipcReadStartedConn{Conn: left, started: make(chan struct{})}
	32	+	conn, started = observedConn, observedConn.started
	33	+	}
	34	+	transport := NewTransport(conn, WithRuntimeObserver(observer))
	35	+	recvDone := make(chan error, 1)
	36	+	go func() { _, err := transport.Recv(); recvDone <- err }()
	37	+	if closeLocal {
	38	+	<-started
	39	+	if err := transport.Close(); err != nil {
	40	+	t.Fatalf("Close() error = %v", err)
	41	+	}
	42	+	} else if err := right.Close(); err != nil {
	43	+	t.Fatalf("peer Close() error = %v", err)
	44	+	}
	45	+	if err := <-recvDone; err == nil {
	46	+	t.Fatal("Recv() error = nil after transport shutdown")
	47	+	}
	48	+	_ = left.Close()
	49	+	_ = right.Close()
	50	+	}
	51	+
	52	+	// A peer shutdown also makes Send fail after its start mark.
	53	+	left, right := net.Pipe()
	54	+	transport := NewTransport(left, WithRuntimeObserver(observer))
	55	+	_ = right.Close()
	56	+	if err := transport.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("failed")}); err == nil {
	57	+	t.Fatal("Send() error = nil after peer close")
	58	+	}
	59	+	_ = transport.Close()
	60	+	if err := closer.Close(); err != nil {
	61	+	t.Fatalf("trace Close() error = %v", err)
	62	+	}
	63	+	assertIPCFailedSpanPairs(t, trace)
	64	+	}
	65	+
	66	+	type ipcReadStartedConn struct {
	67	+	net.Conn
	68	+	started chan struct{}
	69	+	once sync.Once
	70	+	}
	71	+
	72	+	func (c *ipcReadStartedConn) Read(p []byte) (int, error) {
	73	+	c.once.Do(func() { close(c.started) })
	74	+	return c.Conn.Read(p)
	75	+	}
	76	+
	77	+	func assertIPCFailedSpanPairs(t *testing.T, path string) {
	78	+	t.Helper()
	79	+	data, err := os.ReadFile(path)
	80	+	if err != nil {
	81	+	t.Fatalf("ReadFile(trace): %v", err)
	82	+	}
	83	+	type mark struct {
	84	+	ProcessID string `json:"process_id"`
	85	+	Scenario string `json:"scenario"`
	86	+	Run uint64 `json:"run"`
	87	+	Sequence uint64 `json:"sequence"`
	88	+	RequestID uint64 `json:"request_id"`
	89	+	Epoch uint64 `json:"epoch"`
	90	+	Kind string `json:"kind"`
	91	+	Valid bool `json:"valid"`
	92	+	}
	93	+	type spanKey struct {
	94	+	processID, scenario string
	95	+	run, sequence, requestID, epoch uint64
	96	+	}
	97	+	starts := make(map[spanKey]mark)
	98	+	for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
	99	+	var m mark
	100	+	if err := json.Unmarshal([]byte(line), &m); err != nil {
	101	+	t.Fatalf("Unmarshal(trace): %v", err)
	102	+	}
	103	+	key := spanKey{m.ProcessID, m.Scenario, m.Run, m.Sequence, m.RequestID, m.Epoch}
	104	+	switch m.Kind {
	105	+	case "adapter_receive_start", "adapter_send_start":
	106	+	if _, exists := starts[key]; exists {
	107	+	t.Fatalf("duplicate adapter start for %#v", key)
	108	+	}
	109	+	starts[key] = m
	110	+	case "adapter_receive_end", "adapter_send_end":
	111	+	start, ok := starts[key]
	112	+	if !ok \|\| strings.TrimSuffix(start.Kind, "_start") != strings.TrimSuffix(m.Kind, "_end") {
	113	+	t.Fatalf("unmatched adapter end %#v", m)
	114	+	}
	115	+	if m.Valid {
	116	+	t.Fatalf("failed adapter end is valid: %#v", m)
	117	+	}
	118	+	delete(starts, key)
	119	+	}
	120	+	}
	121	+	if len(starts) != 0 {
	122	+	t.Fatalf("unmatched adapter starts: %#v", starts)
	123	+	}
	124	+	}
	125	+
16	126		func TestTransportObservabilityIPCMarksCarriageWithoutChangingBytes(t *testing.T) {
17	127		trace := t.TempDir() + "/ipc.jsonl"
18	128		observer, closer, err := observability.NewJSONL(trace, ipcRuntimeClock{now: time.Unix(0, 211)}, "ipc-process")

@@ -1,6 +1,7 @@
1	package ipc	1		package ipc
2		2
3	import (	3		import (
		4	+	"encoding/json"
4	"net"	5		"net"
5	"os"	6		"os"
6	"reflect"	7		"reflect"

@@ -13,6 +14,115 @@ import (
13	"github.com/bnema/vev/internal/ports"	14		"github.com/bnema/vev/internal/ports"
14	)	15		)
15		16
		17	+	func TestTransportObservabilityIPCEOFAndCloseEndFailedSpans(t *testing.T) {
		18	+	trace := t.TempDir() + "/ipc-shutdown.jsonl"
		19	+	observer, closer, err := observability.NewJSONL(trace, ipcRuntimeClock{now: time.Unix(0, 223)}, "ipc-process")
		20	+	if err != nil {
		21	+	t.Fatalf("NewJSONL() error = %v", err)
		22	+	}
		23	+
		24	+	// EOF from the peer and a locally closed blocked receive are distinct
		25	+	// shutdown paths; both must record a failed, correlated end.
		26	+	for _, closeLocal := range []bool{false, true} {
		27	+	left, right := net.Pipe()
		28	+	var started <-chan struct{}
		29	+	conn := left
		30	+	if closeLocal {
		31	+	observedConn := &ipcReadStartedConn{Conn: left, started: make(chan struct{})}
		32	+	conn, started = observedConn, observedConn.started
		33	+	}
		34	+	transport := NewTransport(conn, WithRuntimeObserver(observer))
		35	+	recvDone := make(chan error, 1)
		36	+	go func() { _, err := transport.Recv(); recvDone <- err }()
		37	+	if closeLocal {
		38	+	<-started
		39	+	if err := transport.Close(); err != nil {
		40	+	t.Fatalf("Close() error = %v", err)
		41	+	}
		42	+	} else if err := right.Close(); err != nil {
		43	+	t.Fatalf("peer Close() error = %v", err)
		44	+	}
		45	+	if err := <-recvDone; err == nil {
		46	+	t.Fatal("Recv() error = nil after transport shutdown")
		47	+	}
		48	+	_ = left.Close()
		49	+	_ = right.Close()
		50	+	}
		51	+
		52	+	// A peer shutdown also makes Send fail after its start mark.
		53	+	left, right := net.Pipe()
		54	+	transport := NewTransport(left, WithRuntimeObserver(observer))
		55	+	_ = right.Close()
		56	+	if err := transport.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("failed")}); err == nil {
		57	+	t.Fatal("Send() error = nil after peer close")
		58	+	}
		59	+	_ = transport.Close()
		60	+	if err := closer.Close(); err != nil {
		61	+	t.Fatalf("trace Close() error = %v", err)
		62	+	}
		63	+	assertIPCFailedSpanPairs(t, trace)
		64	+	}
		65	+
		66	+	type ipcReadStartedConn struct {
		67	+	net.Conn
		68	+	started chan struct{}
		69	+	once sync.Once
		70	+	}
		71	+
		72	+	func (c *ipcReadStartedConn) Read(p []byte) (int, error) {
		73	+	c.once.Do(func() { close(c.started) })
		74	+	return c.Conn.Read(p)
		75	+	}
		76	+
		77	+	func assertIPCFailedSpanPairs(t *testing.T, path string) {
		78	+	t.Helper()
		79	+	data, err := os.ReadFile(path)
		80	+	if err != nil {
		81	+	t.Fatalf("ReadFile(trace): %v", err)
		82	+	}
		83	+	type mark struct {
		84	+	ProcessID string `json:"process_id"`
		85	+	Scenario string `json:"scenario"`
		86	+	Run uint64 `json:"run"`
		87	+	Sequence uint64 `json:"sequence"`
		88	+	RequestID uint64 `json:"request_id"`
		89	+	Epoch uint64 `json:"epoch"`
		90	+	Kind string `json:"kind"`
		91	+	Valid bool `json:"valid"`
		92	+	}
		93	+	type spanKey struct {
		94	+	processID, scenario string
		95	+	run, sequence, requestID, epoch uint64
		96	+	}
		97	+	starts := make(map[spanKey]mark)
		98	+	for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
		99	+	var m mark
		100	+	if err := json.Unmarshal([]byte(line), &m); err != nil {
		101	+	t.Fatalf("Unmarshal(trace): %v", err)
		102	+	}
		103	+	key := spanKey{m.ProcessID, m.Scenario, m.Run, m.Sequence, m.RequestID, m.Epoch}
		104	+	switch m.Kind {
		105	+	case "adapter_receive_start", "adapter_send_start":
		106	+	if _, exists := starts[key]; exists {
		107	+	t.Fatalf("duplicate adapter start for %#v", key)
		108	+	}
		109	+	starts[key] = m
		110	+	case "adapter_receive_end", "adapter_send_end":
		111	+	start, ok := starts[key]
		112	+	if !ok \|\| strings.TrimSuffix(start.Kind, "_start") != strings.TrimSuffix(m.Kind, "_end") {
		113	+	t.Fatalf("unmatched adapter end %#v", m)
		114	+	}
		115	+	if m.Valid {
		116	+	t.Fatalf("failed adapter end is valid: %#v", m)
		117	+	}
		118	+	delete(starts, key)
		119	+	}
		120	+	}
		121	+	if len(starts) != 0 {
		122	+	t.Fatalf("unmatched adapter starts: %#v", starts)
		123	+	}
		124	+	}
		125	+
16	func TestTransportObservabilityIPCMarksCarriageWithoutChangingBytes(t *testing.T) {	126		func TestTransportObservabilityIPCMarksCarriageWithoutChangingBytes(t *testing.T) {
17	trace := t.TempDir() + "/ipc.jsonl"	127		trace := t.TempDir() + "/ipc.jsonl"
18	observer, closer, err := observability.NewJSONL(trace, ipcRuntimeClock{now: time.Unix(0, 211)}, "ipc-process")	128		observer, closer, err := observability.NewJSONL(trace, ipcRuntimeClock{now: time.Unix(0, 211)}, "ipc-process")

M internal/adapters/sshstdio/transport.go +63 -6


@@ -61,11 +61,15 @@ func newTransport(r io.Reader, w io.Writer, closeFn closeFunc, eofErr eofErrFunc
61	61		}
62	62
63	63		type transport struct {
64		-	r io.Reader
65		-	w io.Writer
66		-	close closeFunc
67		-	eofErr eofErrFunc
68		-	observer ports.RuntimeObserver
	64	+	r io.Reader
	65	+	w io.Writer
	66	+	close closeFunc
	67	+	eofErr eofErrFunc
	68	+	observer ports.RuntimeObserver
	69	+	operationMu sync.Mutex
	70	+	operationCount int
	71	+	closing bool
	72	+	operationsDone chan struct{}
69	73
70	74		mu sync.Mutex
71	75		readBuf []byte

@@ -130,6 +134,9 @@ func (t *transport) beginOperation(start ports.RuntimeMarkKind, bytes uint64) fu
130	134		if t.observer == nil {
131	135		return func(bool) {}
132	136		}
	137	+	if !t.beginObservedOperation() {
	138	+	return func(bool) {}
	139	+	}
133	140		correlation := ports.NewRuntimeCorrelation()
134	141		t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("sshstdio", correlation, start, bytes, true))
135	142		end := ports.RuntimeAdapterSendEnd

@@ -137,6 +144,7 @@ func (t *transport) beginOperation(start ports.RuntimeMarkKind, bytes uint64) fu
137	144		end = ports.RuntimeAdapterReceiveEnd
138	145		}
139	146		return func(valid bool) {
	147	+	defer t.finishObservedOperation()
140	148		t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("sshstdio", correlation, end, bytes, valid))
141	149		}
142	150		}

@@ -151,7 +159,56 @@ func (t *transport) mapEOFError(err error) error {
151	159		return err
152	160		}
153	161
154		-	func (t *transport) Close() error { return t.close() }
	162	+	func (t *transport) Close() error {
	163	+	done := t.beginShutdown()
	164	+	// Closing the reader first releases a Recv blocked in io.ReadFull even if
	165	+	// the subprocess shutdown callback reports an error. This lets Recv emit
	166	+	// its failed end mark before the process observer is closed.
	167	+	var readErr error
	168	+	if r, ok := t.r.(io.Closer); ok {
	169	+	readErr = r.Close()
	170	+	}
	171	+	closeErr := t.close()
	172	+	if done != nil {
	173	+	<-done
	174	+	}
	175	+	if closeErr != nil {
	176	+	return closeErr
	177	+	}
	178	+	return readErr
	179	+	}
	180	+
	181	+	func (t *transport) beginObservedOperation() bool {
	182	+	t.operationMu.Lock()
	183	+	defer t.operationMu.Unlock()
	184	+	if t.closing {
	185	+	return false
	186	+	}
	187	+	if t.operationCount == 0 {
	188	+	t.operationsDone = make(chan struct{})
	189	+	}
	190	+	t.operationCount++
	191	+	return true
	192	+	}
	193	+
	194	+	func (t *transport) finishObservedOperation() {
	195	+	t.operationMu.Lock()
	196	+	defer t.operationMu.Unlock()
	197	+	t.operationCount--
	198	+	if t.operationCount == 0 {
	199	+	close(t.operationsDone)
	200	+	}
	201	+	}
	202	+
	203	+	func (t *transport) beginShutdown() <-chan struct{} {
	204	+	t.operationMu.Lock()
	205	+	defer t.operationMu.Unlock()
	206	+	t.closing = true
	207	+	if t.operationCount == 0 {
	208	+	return nil
	209	+	}
	210	+	return t.operationsDone
	211	+	}
155	212
156	213		type processWaiter struct {
157	214		cmd *exec.Cmd


@@ -61,11 +61,15 @@ func newTransport(r io.Reader, w io.Writer, closeFn closeFunc, eofErr eofErrFunc
61		}	61		}
62			62
63		type transport struct {	63		type transport struct {
64	-	r io.Reader	64	+	r io.Reader
65	-	w io.Writer	65	+	w io.Writer
66	-	close closeFunc	66	+	close closeFunc
67	-	eofErr eofErrFunc	67	+	eofErr eofErrFunc
68	-	observer ports.RuntimeObserver	68	+	observer ports.RuntimeObserver
			69	+	operationMu sync.Mutex
			70	+	operationCount int
			71	+	closing bool
			72	+	operationsDone chan struct{}
69			73
70		mu sync.Mutex	74		mu sync.Mutex
71		readBuf []byte	75		readBuf []byte

@@ -130,6 +134,9 @@ func (t *transport) beginOperation(start ports.RuntimeMarkKind, bytes uint64) fu
130		if t.observer == nil {	134		if t.observer == nil {
131		return func(bool) {}	135		return func(bool) {}
132		}	136		}
			137	+	if !t.beginObservedOperation() {
			138	+	return func(bool) {}
			139	+	}
133		correlation := ports.NewRuntimeCorrelation()	140		correlation := ports.NewRuntimeCorrelation()
134		t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("sshstdio", correlation, start, bytes, true))	141		t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("sshstdio", correlation, start, bytes, true))
135		end := ports.RuntimeAdapterSendEnd	142		end := ports.RuntimeAdapterSendEnd

@@ -137,6 +144,7 @@ func (t *transport) beginOperation(start ports.RuntimeMarkKind, bytes uint64) fu
137		end = ports.RuntimeAdapterReceiveEnd	144		end = ports.RuntimeAdapterReceiveEnd
138		}	145		}
139		return func(valid bool) {	146		return func(valid bool) {
			147	+	defer t.finishObservedOperation()
140		t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("sshstdio", correlation, end, bytes, valid))	148		t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("sshstdio", correlation, end, bytes, valid))
141		}	149		}
142		}	150		}

@@ -151,7 +159,56 @@ func (t *transport) mapEOFError(err error) error {
151		return err	159		return err
152		}	160		}
153			161
154	-	func (t *transport) Close() error { return t.close() }	162	+	func (t *transport) Close() error {
			163	+	done := t.beginShutdown()
			164	+	// Closing the reader first releases a Recv blocked in io.ReadFull even if
			165	+	// the subprocess shutdown callback reports an error. This lets Recv emit
			166	+	// its failed end mark before the process observer is closed.
			167	+	var readErr error
			168	+	if r, ok := t.r.(io.Closer); ok {
			169	+	readErr = r.Close()
			170	+	}
			171	+	closeErr := t.close()
			172	+	if done != nil {
			173	+	<-done
			174	+	}
			175	+	if closeErr != nil {
			176	+	return closeErr
			177	+	}
			178	+	return readErr
			179	+	}
			180	+
			181	+	func (t *transport) beginObservedOperation() bool {
			182	+	t.operationMu.Lock()
			183	+	defer t.operationMu.Unlock()
			184	+	if t.closing {
			185	+	return false
			186	+	}
			187	+	if t.operationCount == 0 {
			188	+	t.operationsDone = make(chan struct{})
			189	+	}
			190	+	t.operationCount++
			191	+	return true
			192	+	}
			193	+
			194	+	func (t *transport) finishObservedOperation() {
			195	+	t.operationMu.Lock()
			196	+	defer t.operationMu.Unlock()
			197	+	t.operationCount--
			198	+	if t.operationCount == 0 {
			199	+	close(t.operationsDone)
			200	+	}
			201	+	}
			202	+
			203	+	func (t *transport) beginShutdown() <-chan struct{} {
			204	+	t.operationMu.Lock()
			205	+	defer t.operationMu.Unlock()
			206	+	t.closing = true
			207	+	if t.operationCount == 0 {
			208	+	return nil
			209	+	}
			210	+	return t.operationsDone
			211	+	}
155			212
156		type processWaiter struct {	213		type processWaiter struct {
157		cmd *exec.Cmd	214		cmd *exec.Cmd

M internal/adapters/sshstdio/transport_observability_test.go +109 -0


@@ -2,8 +2,11 @@ package sshstdio
2	2
3	3		import (
4	4		"bytes"
	5	+	"encoding/json"
	6	+	"errors"
5	7		"os"
6	8		"strings"
	9	+	"sync"
7	10		"testing"
8	11		"time"
9	12

@@ -11,6 +14,112 @@ import (
11	14		"github.com/bnema/vev/internal/ports"
12	15		)
13	16
	17	+	func TestTransportObservabilitySSHStdioEOFEndsReceive(t *testing.T) {
	18	+	tracePath := t.TempDir() + "/ssh-eof.jsonl"
	19	+	observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 309)}, "ssh-process")
	20	+	if err != nil {
	21	+	t.Fatalf("NewJSONL() error = %v", err)
	22	+	}
	23	+	if _, err := NewTransport(bytes.NewReader(nil), &bytes.Buffer{}, nil, WithRuntimeObserver(observer)).Recv(); err == nil {
	24	+	t.Fatal("Recv() error = nil at EOF")
	25	+	}
	26	+	if err := closer.Close(); err != nil {
	27	+	t.Fatalf("trace Close() error = %v", err)
	28	+	}
	29	+	assertSSHReceiveFailurePair(t, tracePath)
	30	+	}
	31	+
	32	+	func TestTransportObservabilitySSHStdioCloseEndsBlockedReceive(t *testing.T) {
	33	+	tracePath := t.TempDir() + "/ssh-close.jsonl"
	34	+	observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 311)}, "ssh-process")
	35	+	if err != nil {
	36	+	t.Fatalf("NewJSONL() error = %v", err)
	37	+	}
	38	+	reader := &sshShutdownReader{entered: make(chan struct{}), done: make(chan struct{})}
	39	+	shutdownErr := errors.New("ssh shutdown failed")
	40	+	transport := NewTransport(reader, &bytes.Buffer{}, func() error { return shutdownErr }, WithRuntimeObserver(observer))
	41	+	recvDone := make(chan error, 1)
	42	+	go func() { _, err := transport.Recv(); recvDone <- err }()
	43	+	<-reader.entered
	44	+	if err := transport.Close(); !errors.Is(err, shutdownErr) {
	45	+	t.Fatalf("Close() error = %v, want %v", err, shutdownErr)
	46	+	}
	47	+	if err := <-recvDone; err == nil {
	48	+	t.Fatal("Recv() error = nil after Close()")
	49	+	}
	50	+	if err := closer.Close(); err != nil {
	51	+	t.Fatalf("trace Close() error = %v", err)
	52	+	}
	53	+	assertSSHReceiveFailurePair(t, tracePath)
	54	+	}
	55	+
	56	+	type sshShutdownReader struct {
	57	+	entered chan struct{}
	58	+	done chan struct{}
	59	+	once sync.Once
	60	+	}
	61	+
	62	+	func (r *sshShutdownReader) Read([]byte) (int, error) {
	63	+	r.once.Do(func() { close(r.entered) })
	64	+	<-r.done
	65	+	return 0, errors.New("reader closed")
	66	+	}
	67	+
	68	+	func (r *sshShutdownReader) Close() error {
	69	+	r.once.Do(func() { close(r.entered) })
	70	+	close(r.done)
	71	+	return nil
	72	+	}
	73	+
	74	+	func assertSSHReceiveFailurePair(t *testing.T, path string) {
	75	+	t.Helper()
	76	+	data, err := os.ReadFile(path)
	77	+	if err != nil {
	78	+	t.Fatalf("ReadFile(trace): %v", err)
	79	+	}
	80	+	var starts, ends []struct {
	81	+	ProcessID string `json:"process_id"`
	82	+	Scenario string `json:"scenario"`
	83	+	Run uint64 `json:"run"`
	84	+	Sequence uint64 `json:"sequence"`
	85	+	RequestID uint64 `json:"request_id"`
	86	+	Epoch uint64 `json:"epoch"`
	87	+	Kind string `json:"kind"`
	88	+	Valid bool `json:"valid"`
	89	+	}
	90	+	for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
	91	+	var mark struct {
	92	+	ProcessID string `json:"process_id"`
	93	+	Scenario string `json:"scenario"`
	94	+	Run uint64 `json:"run"`
	95	+	Sequence uint64 `json:"sequence"`
	96	+	RequestID uint64 `json:"request_id"`
	97	+	Epoch uint64 `json:"epoch"`
	98	+	Kind string `json:"kind"`
	99	+	Valid bool `json:"valid"`
	100	+	}
	101	+	if err := json.Unmarshal([]byte(line), &mark); err != nil {
	102	+	t.Fatalf("Unmarshal(trace): %v", err)
	103	+	}
	104	+	switch mark.Kind {
	105	+	case "adapter_receive_start":
	106	+	starts = append(starts, mark)
	107	+	case "adapter_receive_end":
	108	+	ends = append(ends, mark)
	109	+	}
	110	+	}
	111	+	if len(starts) != 1 \|\| len(ends) != 1 {
	112	+	t.Fatalf("receive spans start=%d end=%d, want exactly one pair", len(starts), len(ends))
	113	+	}
	114	+	start, end := starts[0], ends[0]
	115	+	if start.ProcessID != end.ProcessID \|\| start.Scenario != end.Scenario \|\| start.Run != end.Run \|\| start.Sequence != end.Sequence \|\| start.RequestID != end.RequestID \|\| start.Epoch != end.Epoch {
	116	+	t.Fatalf("mismatched receive correlation: start=%+v end=%+v", start, end)
	117	+	}
	118	+	if !start.Valid \|\| end.Valid {
	119	+	t.Fatalf("receive validity start=%t end=%t, want true/false", start.Valid, end.Valid)
	120	+	}
	121	+	}
	122	+
14	123		func TestTransportObservabilitySSHStdioPreservesCarriage(t *testing.T) {
15	124		tracePath := t.TempDir() + "/ssh.jsonl"
16	125		observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 307)}, "ssh-process")


@@ -2,8 +2,11 @@ package sshstdio
2		2
3	import (	3		import (
4	"bytes"	4		"bytes"
		5	+	"encoding/json"
		6	+	"errors"
5	"os"	7		"os"
6	"strings"	8		"strings"
		9	+	"sync"
7	"testing"	10		"testing"
8	"time"	11		"time"
9		12

@@ -11,6 +14,112 @@ import (
11	"github.com/bnema/vev/internal/ports"	14		"github.com/bnema/vev/internal/ports"
12	)	15		)
13		16
		17	+	func TestTransportObservabilitySSHStdioEOFEndsReceive(t *testing.T) {
		18	+	tracePath := t.TempDir() + "/ssh-eof.jsonl"
		19	+	observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 309)}, "ssh-process")
		20	+	if err != nil {
		21	+	t.Fatalf("NewJSONL() error = %v", err)
		22	+	}
		23	+	if _, err := NewTransport(bytes.NewReader(nil), &bytes.Buffer{}, nil, WithRuntimeObserver(observer)).Recv(); err == nil {
		24	+	t.Fatal("Recv() error = nil at EOF")
		25	+	}
		26	+	if err := closer.Close(); err != nil {
		27	+	t.Fatalf("trace Close() error = %v", err)
		28	+	}
		29	+	assertSSHReceiveFailurePair(t, tracePath)
		30	+	}
		31	+
		32	+	func TestTransportObservabilitySSHStdioCloseEndsBlockedReceive(t *testing.T) {
		33	+	tracePath := t.TempDir() + "/ssh-close.jsonl"
		34	+	observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 311)}, "ssh-process")
		35	+	if err != nil {
		36	+	t.Fatalf("NewJSONL() error = %v", err)
		37	+	}
		38	+	reader := &sshShutdownReader{entered: make(chan struct{}), done: make(chan struct{})}
		39	+	shutdownErr := errors.New("ssh shutdown failed")
		40	+	transport := NewTransport(reader, &bytes.Buffer{}, func() error { return shutdownErr }, WithRuntimeObserver(observer))
		41	+	recvDone := make(chan error, 1)
		42	+	go func() { _, err := transport.Recv(); recvDone <- err }()
		43	+	<-reader.entered
		44	+	if err := transport.Close(); !errors.Is(err, shutdownErr) {
		45	+	t.Fatalf("Close() error = %v, want %v", err, shutdownErr)
		46	+	}
		47	+	if err := <-recvDone; err == nil {
		48	+	t.Fatal("Recv() error = nil after Close()")
		49	+	}
		50	+	if err := closer.Close(); err != nil {
		51	+	t.Fatalf("trace Close() error = %v", err)
		52	+	}
		53	+	assertSSHReceiveFailurePair(t, tracePath)
		54	+	}
		55	+
		56	+	type sshShutdownReader struct {
		57	+	entered chan struct{}
		58	+	done chan struct{}
		59	+	once sync.Once
		60	+	}
		61	+
		62	+	func (r *sshShutdownReader) Read([]byte) (int, error) {
		63	+	r.once.Do(func() { close(r.entered) })
		64	+	<-r.done
		65	+	return 0, errors.New("reader closed")
		66	+	}
		67	+
		68	+	func (r *sshShutdownReader) Close() error {
		69	+	r.once.Do(func() { close(r.entered) })
		70	+	close(r.done)
		71	+	return nil
		72	+	}
		73	+
		74	+	func assertSSHReceiveFailurePair(t *testing.T, path string) {
		75	+	t.Helper()
		76	+	data, err := os.ReadFile(path)
		77	+	if err != nil {
		78	+	t.Fatalf("ReadFile(trace): %v", err)
		79	+	}
		80	+	var starts, ends []struct {
		81	+	ProcessID string `json:"process_id"`
		82	+	Scenario string `json:"scenario"`
		83	+	Run uint64 `json:"run"`
		84	+	Sequence uint64 `json:"sequence"`
		85	+	RequestID uint64 `json:"request_id"`
		86	+	Epoch uint64 `json:"epoch"`
		87	+	Kind string `json:"kind"`
		88	+	Valid bool `json:"valid"`
		89	+	}
		90	+	for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
		91	+	var mark struct {
		92	+	ProcessID string `json:"process_id"`
		93	+	Scenario string `json:"scenario"`
		94	+	Run uint64 `json:"run"`
		95	+	Sequence uint64 `json:"sequence"`
		96	+	RequestID uint64 `json:"request_id"`
		97	+	Epoch uint64 `json:"epoch"`
		98	+	Kind string `json:"kind"`
		99	+	Valid bool `json:"valid"`
		100	+	}
		101	+	if err := json.Unmarshal([]byte(line), &mark); err != nil {
		102	+	t.Fatalf("Unmarshal(trace): %v", err)
		103	+	}
		104	+	switch mark.Kind {
		105	+	case "adapter_receive_start":
		106	+	starts = append(starts, mark)
		107	+	case "adapter_receive_end":
		108	+	ends = append(ends, mark)
		109	+	}
		110	+	}
		111	+	if len(starts) != 1 \|\| len(ends) != 1 {
		112	+	t.Fatalf("receive spans start=%d end=%d, want exactly one pair", len(starts), len(ends))
		113	+	}
		114	+	start, end := starts[0], ends[0]
		115	+	if start.ProcessID != end.ProcessID \|\| start.Scenario != end.Scenario \|\| start.Run != end.Run \|\| start.Sequence != end.Sequence \|\| start.RequestID != end.RequestID \|\| start.Epoch != end.Epoch {
		116	+	t.Fatalf("mismatched receive correlation: start=%+v end=%+v", start, end)
		117	+	}
		118	+	if !start.Valid \|\| end.Valid {
		119	+	t.Fatalf("receive validity start=%t end=%t, want true/false", start.Valid, end.Valid)
		120	+	}
		121	+	}
		122	+
14	func TestTransportObservabilitySSHStdioPreservesCarriage(t *testing.T) {	123		func TestTransportObservabilitySSHStdioPreservesCarriage(t *testing.T) {
15	tracePath := t.TempDir() + "/ssh.jsonl"	124		tracePath := t.TempDir() + "/ssh.jsonl"
16	observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 307)}, "ssh-process")	125		observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 307)}, "ssh-process")