fix(obs): decouple runtime reporting from ACK progress

commit: ba3714c57a183570f1fe81c432a7ef1c1aa3388e
author: brice · 3mo ago · 2026-07-13 09:25 UTC
parent: c10e22a
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

5 files · +312 -23

M internal/ports/runtime_observer.go +132 -0


@@ -2,6 +2,7 @@ package ports
2	2
3	3		import (
4	4		"errors"
	5	+	"sync"
5	6		"sync/atomic"
6	7		)
7	8

@@ -61,6 +62,137 @@ type RuntimeMark struct {
61	62		// a producer provide a timestamp in another clock domain.
62	63		type RuntimeObserver interface{ ObserveRuntime(RuntimeMark) }
63	64
	65	+	// SerializedRuntimeObserver accepts marks without waiting for observer I/O.
	66	+	// Flush and Close wait until every accepted mark has reached Observer. Close
	67	+	// is idempotent and rejects later marks.
	68	+	type SerializedRuntimeObserver interface {
	69	+	RuntimeObserver
	70	+	Flush()
	71	+	Close()
	72	+	}
	73	+
	74	+	// NewSerializedRuntimeObserver isolates producers from a potentially slow
	75	+	// observer with one ordered worker. capacity bounds queued marks. If the queue
	76	+	// fills, dropped marks are represented by one ordered diagnostic mark before
	77	+	// reporting resumes, so a trace can never silently claim completeness.
	78	+	func NewSerializedRuntimeObserver(observer RuntimeObserver, capacity int) SerializedRuntimeObserver {
	79	+	if observer == nil {
	80	+	return nil
	81	+	}
	82	+	if capacity < 1 {
	83	+	capacity = 1
	84	+	}
	85	+	o := &serializedRuntimeObserver{observer: observer, capacity: capacity, done: make(chan struct{})}
	86	+	o.cond = sync.NewCond(&o.mu)
	87	+	go o.run()
	88	+	return o
	89	+	}
	90	+
	91	+	type serializedRuntimeObserver struct {
	92	+	observer RuntimeObserver
	93	+	capacity int
	94	+
	95	+	mu sync.Mutex
	96	+	cond *sync.Cond
	97	+	marks []RuntimeMark
	98	+	dropped uint64
	99	+	closed bool
	100	+	active bool
	101	+	done chan struct{}
	102	+	}
	103	+
	104	+	func (o *serializedRuntimeObserver) ObserveRuntime(mark RuntimeMark) {
	105	+	if o == nil {
	106	+	return
	107	+	}
	108	+	o.mu.Lock()
	109	+	defer o.mu.Unlock()
	110	+	if o.closed {
	111	+	return
	112	+	}
	113	+	if o.dropped != 0 {
	114	+	// Reserve space for both the gap marker and the first post-gap mark.
	115	+	// This preserves the producer order around a bounded-queue loss.
	116	+	if len(o.marks)+2 > o.capacity {
	117	+	o.dropped++
	118	+	return
	119	+	}
	120	+	o.marks = append(o.marks, runtimeObserverGapMark(o.dropped), mark)
	121	+	o.dropped = 0
	122	+	o.cond.Signal()
	123	+	return
	124	+	}
	125	+	if len(o.marks) == o.capacity {
	126	+	o.dropped = 1
	127	+	return
	128	+	}
	129	+	o.marks = append(o.marks, mark)
	130	+	o.cond.Signal()
	131	+	}
	132	+
	133	+	func runtimeObserverGapMark(dropped uint64) RuntimeMark {
	134	+	return NewRuntimeMark("observer", RuntimeTransportDiagnostic, dropped, false)
	135	+	}
	136	+
	137	+	func (o *serializedRuntimeObserver) Flush() {
	138	+	if o == nil {
	139	+	return
	140	+	}
	141	+	o.mu.Lock()
	142	+	for len(o.marks) != 0 \|\| o.active \|\| o.dropped != 0 {
	143	+	o.cond.Signal()
	144	+	o.cond.Wait()
	145	+	}
	146	+	closed := o.closed
	147	+	o.mu.Unlock()
	148	+	if closed {
	149	+	<-o.done
	150	+	}
	151	+	}
	152	+
	153	+	func (o *serializedRuntimeObserver) Close() {
	154	+	if o == nil {
	155	+	return
	156	+	}
	157	+	o.mu.Lock()
	158	+	if !o.closed {
	159	+	o.closed = true
	160	+	o.cond.Broadcast()
	161	+	}
	162	+	o.mu.Unlock()
	163	+	<-o.done
	164	+	}
	165	+
	166	+	func (o *serializedRuntimeObserver) run() {
	167	+	defer close(o.done)
	168	+	for {
	169	+	o.mu.Lock()
	170	+	for len(o.marks) == 0 && o.dropped == 0 && !o.closed {
	171	+	o.cond.Wait()
	172	+	}
	173	+	if len(o.marks) == 0 && o.dropped != 0 {
	174	+	o.marks = append(o.marks, runtimeObserverGapMark(o.dropped))
	175	+	o.dropped = 0
	176	+	}
	177	+	if len(o.marks) == 0 && o.closed {
	178	+	o.cond.Broadcast()
	179	+	o.mu.Unlock()
	180	+	return
	181	+	}
	182	+	mark := o.marks[0]
	183	+	o.marks = o.marks[1:]
	184	+	o.active = true
	185	+	o.mu.Unlock()
	186	+
	187	+	o.observer.ObserveRuntime(mark)
	188	+
	189	+	o.mu.Lock()
	190	+	o.active = false
	191	+	o.cond.Broadcast()
	192	+	o.mu.Unlock()
	193	+	}
	194	+	}
	195	+
64	196		// RuntimeMarkValid reports whether a mark has all correlation fields required
65	197		// by the JSONL schema. Tick is checked separately by the timestamp owner.
66	198		func RuntimeMarkValid(mark RuntimeMark) bool {


@@ -2,6 +2,7 @@ package ports
2		2
3	import (	3		import (
4	"errors"	4		"errors"
		5	+	"sync"
5	"sync/atomic"	6		"sync/atomic"
6	)	7		)
7		8

@@ -61,6 +62,137 @@ type RuntimeMark struct {
61	// a producer provide a timestamp in another clock domain.	62		// a producer provide a timestamp in another clock domain.
62	type RuntimeObserver interface{ ObserveRuntime(RuntimeMark) }	63		type RuntimeObserver interface{ ObserveRuntime(RuntimeMark) }
63		64
		65	+	// SerializedRuntimeObserver accepts marks without waiting for observer I/O.
		66	+	// Flush and Close wait until every accepted mark has reached Observer. Close
		67	+	// is idempotent and rejects later marks.
		68	+	type SerializedRuntimeObserver interface {
		69	+	RuntimeObserver
		70	+	Flush()
		71	+	Close()
		72	+	}
		73	+
		74	+	// NewSerializedRuntimeObserver isolates producers from a potentially slow
		75	+	// observer with one ordered worker. capacity bounds queued marks. If the queue
		76	+	// fills, dropped marks are represented by one ordered diagnostic mark before
		77	+	// reporting resumes, so a trace can never silently claim completeness.
		78	+	func NewSerializedRuntimeObserver(observer RuntimeObserver, capacity int) SerializedRuntimeObserver {
		79	+	if observer == nil {
		80	+	return nil
		81	+	}
		82	+	if capacity < 1 {
		83	+	capacity = 1
		84	+	}
		85	+	o := &serializedRuntimeObserver{observer: observer, capacity: capacity, done: make(chan struct{})}
		86	+	o.cond = sync.NewCond(&o.mu)
		87	+	go o.run()
		88	+	return o
		89	+	}
		90	+
		91	+	type serializedRuntimeObserver struct {
		92	+	observer RuntimeObserver
		93	+	capacity int
		94	+
		95	+	mu sync.Mutex
		96	+	cond *sync.Cond
		97	+	marks []RuntimeMark
		98	+	dropped uint64
		99	+	closed bool
		100	+	active bool
		101	+	done chan struct{}
		102	+	}
		103	+
		104	+	func (o *serializedRuntimeObserver) ObserveRuntime(mark RuntimeMark) {
		105	+	if o == nil {
		106	+	return
		107	+	}
		108	+	o.mu.Lock()
		109	+	defer o.mu.Unlock()
		110	+	if o.closed {
		111	+	return
		112	+	}
		113	+	if o.dropped != 0 {
		114	+	// Reserve space for both the gap marker and the first post-gap mark.
		115	+	// This preserves the producer order around a bounded-queue loss.
		116	+	if len(o.marks)+2 > o.capacity {
		117	+	o.dropped++
		118	+	return
		119	+	}
		120	+	o.marks = append(o.marks, runtimeObserverGapMark(o.dropped), mark)
		121	+	o.dropped = 0
		122	+	o.cond.Signal()
		123	+	return
		124	+	}
		125	+	if len(o.marks) == o.capacity {
		126	+	o.dropped = 1
		127	+	return
		128	+	}
		129	+	o.marks = append(o.marks, mark)
		130	+	o.cond.Signal()
		131	+	}
		132	+
		133	+	func runtimeObserverGapMark(dropped uint64) RuntimeMark {
		134	+	return NewRuntimeMark("observer", RuntimeTransportDiagnostic, dropped, false)
		135	+	}
		136	+
		137	+	func (o *serializedRuntimeObserver) Flush() {
		138	+	if o == nil {
		139	+	return
		140	+	}
		141	+	o.mu.Lock()
		142	+	for len(o.marks) != 0 \|\| o.active \|\| o.dropped != 0 {
		143	+	o.cond.Signal()
		144	+	o.cond.Wait()
		145	+	}
		146	+	closed := o.closed
		147	+	o.mu.Unlock()
		148	+	if closed {
		149	+	<-o.done
		150	+	}
		151	+	}
		152	+
		153	+	func (o *serializedRuntimeObserver) Close() {
		154	+	if o == nil {
		155	+	return
		156	+	}
		157	+	o.mu.Lock()
		158	+	if !o.closed {
		159	+	o.closed = true
		160	+	o.cond.Broadcast()
		161	+	}
		162	+	o.mu.Unlock()
		163	+	<-o.done
		164	+	}
		165	+
		166	+	func (o *serializedRuntimeObserver) run() {
		167	+	defer close(o.done)
		168	+	for {
		169	+	o.mu.Lock()
		170	+	for len(o.marks) == 0 && o.dropped == 0 && !o.closed {
		171	+	o.cond.Wait()
		172	+	}
		173	+	if len(o.marks) == 0 && o.dropped != 0 {
		174	+	o.marks = append(o.marks, runtimeObserverGapMark(o.dropped))
		175	+	o.dropped = 0
		176	+	}
		177	+	if len(o.marks) == 0 && o.closed {
		178	+	o.cond.Broadcast()
		179	+	o.mu.Unlock()
		180	+	return
		181	+	}
		182	+	mark := o.marks[0]
		183	+	o.marks = o.marks[1:]
		184	+	o.active = true
		185	+	o.mu.Unlock()
		186	+
		187	+	o.observer.ObserveRuntime(mark)
		188	+
		189	+	o.mu.Lock()
		190	+	o.active = false
		191	+	o.cond.Broadcast()
		192	+	o.mu.Unlock()
		193	+	}
		194	+	}
		195	+
64	// RuntimeMarkValid reports whether a mark has all correlation fields required	196		// RuntimeMarkValid reports whether a mark has all correlation fields required
65	// by the JSONL schema. Tick is checked separately by the timestamp owner.	197		// by the JSONL schema. Tick is checked separately by the timestamp owner.
66	func RuntimeMarkValid(mark RuntimeMark) bool {	198		func RuntimeMarkValid(mark RuntimeMark) bool {

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


@@ -2,7 +2,9 @@ package ports
2	2
3	3		import (
4	4		"reflect"
	5	+	"sync"
5	6		"testing"
	7	+	"time"
6	8		)
7	9
8	10		func TestRuntimeObserverContract(t *testing.T) {

@@ -42,6 +44,61 @@ func TestRuntimeObserverContract(t *testing.T) {
42	44		var _ RuntimeObserver = runtimeObserverFunc(func(RuntimeMark) {})
43	45		}
44	46
	47	+	func TestSerializedRuntimeObserverFlushesInOrderAndCloses(t *testing.T) {
	48	+	var mu sync.Mutex
	49	+	var got []RuntimeMark
	50	+	observer := NewSerializedRuntimeObserver(runtimeObserverFunc(func(mark RuntimeMark) {
	51	+	mu.Lock()
	52	+	got = append(got, mark)
	53	+	mu.Unlock()
	54	+	}), 2)
	55	+	observer.ObserveRuntime(NewRuntimeMark("client", RuntimeEmitStart, 1, true))
	56	+	observer.ObserveRuntime(NewRuntimeMark("client", RuntimeEmitEnd, 1, true))
	57	+	observer.Flush()
	58	+	observer.Close()
	59	+	observer.Close()
	60	+
	61	+	mu.Lock()
	62	+	defer mu.Unlock()
	63	+	if len(got) != 2 \|\| got[0].Kind != RuntimeEmitStart \|\| got[1].Kind != RuntimeEmitEnd {
	64	+	t.Fatalf("serialized marks = %#v", got)
	65	+	}
	66	+	}
	67	+
	68	+	func TestSerializedRuntimeObserverReportsBoundedQueueLoss(t *testing.T) {
	69	+	entered := make(chan struct{})
	70	+	release := make(chan struct{})
	71	+	var mu sync.Mutex
	72	+	var got []RuntimeMark
	73	+	observer := NewSerializedRuntimeObserver(runtimeObserverFunc(func(mark RuntimeMark) {
	74	+	if mark.Kind == RuntimeEmitStart {
	75	+	close(entered)
	76	+	<-release
	77	+	}
	78	+	mu.Lock()
	79	+	got = append(got, mark)
	80	+	mu.Unlock()
	81	+	}), 2)
	82	+	observer.ObserveRuntime(NewRuntimeMark("client", RuntimeEmitStart, 1, true))
	83	+	select {
	84	+	case <-entered:
	85	+	case <-time.After(time.Second):
	86	+	t.Fatal("observer did not start")
	87	+	}
	88	+	observer.ObserveRuntime(NewRuntimeMark("client", RuntimeComposeStart, 1, true))
	89	+	observer.ObserveRuntime(NewRuntimeMark("client", RuntimeComposeEnd, 1, true))
	90	+	observer.ObserveRuntime(NewRuntimeMark("client", RuntimeEmitEnd, 1, true)) // bounded loss
	91	+	close(release)
	92	+	observer.Flush()
	93	+	observer.Close()
	94	+
	95	+	mu.Lock()
	96	+	defer mu.Unlock()
	97	+	if len(got) != 4 \|\| got[0].Kind != RuntimeEmitStart \|\| got[1].Kind != RuntimeComposeStart \|\| got[2].Kind != RuntimeComposeEnd \|\| got[3].Kind != RuntimeTransportDiagnostic \|\| got[3].Bytes != 1 \|\| got[3].Valid {
	98	+	t.Fatalf("bounded-loss marks = %#v", got)
	99	+	}
	100	+	}
	101	+
45	102		func TestRuntimeObserverRequiredSpanKinds(t *testing.T) {
46	103		pairs := []struct {
47	104		start RuntimeMarkKind


@@ -2,7 +2,9 @@ package ports
2		2
3	import (	3		import (
4	"reflect"	4		"reflect"
		5	+	"sync"
5	"testing"	6		"testing"
		7	+	"time"
6	)	8		)
7		9
8	func TestRuntimeObserverContract(t *testing.T) {	10		func TestRuntimeObserverContract(t *testing.T) {

@@ -42,6 +44,61 @@ func TestRuntimeObserverContract(t *testing.T) {
42	var _ RuntimeObserver = runtimeObserverFunc(func(RuntimeMark) {})	44		var _ RuntimeObserver = runtimeObserverFunc(func(RuntimeMark) {})
43	}	45		}
44		46
		47	+	func TestSerializedRuntimeObserverFlushesInOrderAndCloses(t *testing.T) {
		48	+	var mu sync.Mutex
		49	+	var got []RuntimeMark
		50	+	observer := NewSerializedRuntimeObserver(runtimeObserverFunc(func(mark RuntimeMark) {
		51	+	mu.Lock()
		52	+	got = append(got, mark)
		53	+	mu.Unlock()
		54	+	}), 2)
		55	+	observer.ObserveRuntime(NewRuntimeMark("client", RuntimeEmitStart, 1, true))
		56	+	observer.ObserveRuntime(NewRuntimeMark("client", RuntimeEmitEnd, 1, true))
		57	+	observer.Flush()
		58	+	observer.Close()
		59	+	observer.Close()
		60	+
		61	+	mu.Lock()
		62	+	defer mu.Unlock()
		63	+	if len(got) != 2 \|\| got[0].Kind != RuntimeEmitStart \|\| got[1].Kind != RuntimeEmitEnd {
		64	+	t.Fatalf("serialized marks = %#v", got)
		65	+	}
		66	+	}
		67	+
		68	+	func TestSerializedRuntimeObserverReportsBoundedQueueLoss(t *testing.T) {
		69	+	entered := make(chan struct{})
		70	+	release := make(chan struct{})
		71	+	var mu sync.Mutex
		72	+	var got []RuntimeMark
		73	+	observer := NewSerializedRuntimeObserver(runtimeObserverFunc(func(mark RuntimeMark) {
		74	+	if mark.Kind == RuntimeEmitStart {
		75	+	close(entered)
		76	+	<-release
		77	+	}
		78	+	mu.Lock()
		79	+	got = append(got, mark)
		80	+	mu.Unlock()
		81	+	}), 2)
		82	+	observer.ObserveRuntime(NewRuntimeMark("client", RuntimeEmitStart, 1, true))
		83	+	select {
		84	+	case <-entered:
		85	+	case <-time.After(time.Second):
		86	+	t.Fatal("observer did not start")
		87	+	}
		88	+	observer.ObserveRuntime(NewRuntimeMark("client", RuntimeComposeStart, 1, true))
		89	+	observer.ObserveRuntime(NewRuntimeMark("client", RuntimeComposeEnd, 1, true))
		90	+	observer.ObserveRuntime(NewRuntimeMark("client", RuntimeEmitEnd, 1, true)) // bounded loss
		91	+	close(release)
		92	+	observer.Flush()
		93	+	observer.Close()
		94	+
		95	+	mu.Lock()
		96	+	defer mu.Unlock()
		97	+	if len(got) != 4 \|\| got[0].Kind != RuntimeEmitStart \|\| got[1].Kind != RuntimeComposeStart \|\| got[2].Kind != RuntimeComposeEnd \|\| got[3].Kind != RuntimeTransportDiagnostic \|\| got[3].Bytes != 1 \|\| got[3].Valid {
		98	+	t.Fatalf("bounded-loss marks = %#v", got)
		99	+	}
		100	+	}
		101	+
45	func TestRuntimeObserverRequiredSpanKinds(t *testing.T) {	102		func TestRuntimeObserverRequiredSpanKinds(t *testing.T) {
46	pairs := []struct {	103		pairs := []struct {
47	start RuntimeMarkKind	104		start RuntimeMarkKind

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


@@ -102,6 +102,10 @@ const stdinBufSize = 4096
102	102		// goroutine, decoupling the input/resize pumps from transport back-pressure.
103	103		const sendQueueDepth = 64
104	104
	105	+	// runtimeObserverQueueDepth bounds marks awaiting the one reporting worker.
	106	+	// A full queue records an ordered diagnostic gap rather than blocking ACKs.
	107	+	const runtimeObserverQueueDepth = 64
	108	+
105	109		var reconnectSleep = sleepReconnect
106	110		var reconnectSleepWithResize = sleepReconnectWithResizeEvents
107	111

@@ -141,6 +145,11 @@ func RunWithRuntimeObserver(ctx context.Context, dialer ports.Dialer, term ports
141	145		}
142	146
143	147		func run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, clk ports.Clock, intent uint8, name string, remote bool, clipboard ports.ClipboardReader, log *slog.Logger, observer ports.RuntimeObserver) (retErr error) {
	148	+	if observer != nil {
	149	+	reporter := ports.NewSerializedRuntimeObserver(observer, runtimeObserverQueueDepth)
	150	+	defer reporter.Close()
	151	+	observer = reporter
	152	+	}
144	153		if log == nil {
145	154		log = slog.Default()
146	155		}


@@ -102,6 +102,10 @@ const stdinBufSize = 4096
102	// goroutine, decoupling the input/resize pumps from transport back-pressure.	102		// goroutine, decoupling the input/resize pumps from transport back-pressure.
103	const sendQueueDepth = 64	103		const sendQueueDepth = 64
104		104
		105	+	// runtimeObserverQueueDepth bounds marks awaiting the one reporting worker.
		106	+	// A full queue records an ordered diagnostic gap rather than blocking ACKs.
		107	+	const runtimeObserverQueueDepth = 64
		108	+
105	var reconnectSleep = sleepReconnect	109		var reconnectSleep = sleepReconnect
106	var reconnectSleepWithResize = sleepReconnectWithResizeEvents	110		var reconnectSleepWithResize = sleepReconnectWithResizeEvents
107		111

@@ -141,6 +145,11 @@ func RunWithRuntimeObserver(ctx context.Context, dialer ports.Dialer, term ports
141	}	145		}
142		146
143	func run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, clk ports.Clock, intent uint8, name string, remote bool, clipboard ports.ClipboardReader, log *slog.Logger, observer ports.RuntimeObserver) (retErr error) {	147		func run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, clk ports.Clock, intent uint8, name string, remote bool, clipboard ports.ClipboardReader, log *slog.Logger, observer ports.RuntimeObserver) (retErr error) {
		148	+	if observer != nil {
		149	+	reporter := ports.NewSerializedRuntimeObserver(observer, runtimeObserverQueueDepth)
		150	+	defer reporter.Close()
		151	+	observer = reporter
		152	+	}
144	if log == nil {	153		if log == nil {
145	log = slog.Default()	154		log = slog.Default()
146	}	155		}

M internal/usecase/client/transport_observability_test.go +70 -0


@@ -6,6 +6,7 @@ import (
6	6		"sync"
7	7		"sync/atomic"
8	8		"testing"
	9	+	"time"
9	10
10	11		"github.com/stretchr/testify/require"
11	12

@@ -15,6 +16,53 @@ import (
15	16		"github.com/bnema/vev/internal/usecase/client"
16	17		)
17	18
	19	+	func TestBlockingRuntimeObserverDoesNotDelayTerminalFlushOrACK(t *testing.T) {
	20	+	var out bytes.Buffer
	21	+	var restores atomic.Int32
	22	+	resizeCh := make(chan domain.Size)
	23	+	flushed := make(chan struct{})
	24	+	term := portsmocks.NewMockTerminal(t)
	25	+	term.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
	26	+	term.EXPECT().EnterRaw().Return(func() error { restores.Add(1); return nil }, nil).Once()
	27	+	in := newBlockingReader()
	28	+	defer in.unblock()
	29	+	term.EXPECT().In().Return(in).Maybe()
	30	+	term.EXPECT().Out().Return(&out).Maybe()
	31	+	term.EXPECT().Flush().Run(func() { close(flushed) }).Return(nil).Once()
	32	+	term.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
	33	+
	34	+	observer := &blockingClientRuntimeObserver{entered: make(chan struct{}), release: make(chan struct{})}
	35	+	acked := make(chan struct{})
	36	+	transport := portsmocks.NewMockTransport(t)
	37	+	transport.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
	38	+	transport.EXPECT().Send(isType(ports.MsgAck)).Run(func(ports.Frame) { close(acked) }).Return(nil).Once()
	39	+	unblock := scriptRecv(transport,
	40	+	recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
	41	+	recvItem{f: frameOf(ports.MsgOutput, ports.MarshalOutput(ports.Output{Data: []byte("flush-before-observe"), NewStateNum: 3}))},
	42	+	)
	43	+	defer unblock()
	44	+	transport.EXPECT().Close().Return(nil).Once()
	45	+
	46	+	ctx, cancel := context.WithCancel(context.Background())
	47	+	result := make(chan error, 1)
	48	+	go func() {
	49	+	result <- client.Attach(ctx, transport, term, realClock{}, ports.IntentEphemeral, "", client.WithRuntimeObserver(observer))
	50	+	}()
	51	+	awaitClientRuntime(t, observer.entered, "blocking observer")
	52	+	awaitClientRuntime(t, flushed, "terminal flush")
	53	+	awaitClientRuntime(t, acked, "ACK")
	54	+	cancel()
	55	+	close(observer.release)
	56	+	select {
	57	+	case err := <-result:
	58	+	require.NoError(t, err)
	59	+	case <-time.After(time.Second):
	60	+	t.Fatal("Attach did not complete after observer release")
	61	+	}
	62	+	require.Equal(t, "flush-before-observe", out.String())
	63	+	require.Equal(t, int32(1), restores.Load())
	64	+	}
	65	+
18	66		func TestTerminalFlushBoundaryTransportObservability(t *testing.T) {
19	67		var out bytes.Buffer
20	68		var restores atomic.Int32

@@ -43,6 +91,28 @@ func TestTerminalFlushBoundaryTransportObservability(t *testing.T) {
43	91		observer.requireOrdered(t, ports.RuntimeTerminalFlushed)
44	92		}
45	93
	94	+	type blockingClientRuntimeObserver struct {
	95	+	entered chan struct{}
	96	+	release chan struct{}
	97	+	once sync.Once
	98	+	}
	99	+
	100	+	func (o *blockingClientRuntimeObserver) ObserveRuntime(ports.RuntimeMark) {
	101	+	o.once.Do(func() {
	102	+	close(o.entered)
	103	+	<-o.release
	104	+	})
	105	+	}
	106	+
	107	+	func awaitClientRuntime(t *testing.T, ch <-chan struct{}, what string) {
	108	+	t.Helper()
	109	+	select {
	110	+	case <-ch:
	111	+	case <-time.After(time.Second):
	112	+	t.Fatalf("timed out waiting for %s", what)
	113	+	}
	114	+	}
	115	+
46	116		type clientRuntimeObserver struct {
47	117		mu sync.Mutex
48	118		marks []ports.RuntimeMark


@@ -6,6 +6,7 @@ import (
6	"sync"	6		"sync"
7	"sync/atomic"	7		"sync/atomic"
8	"testing"	8		"testing"
		9	+	"time"
9		10
10	"github.com/stretchr/testify/require"	11		"github.com/stretchr/testify/require"
11		12

@@ -15,6 +16,53 @@ import (
15	"github.com/bnema/vev/internal/usecase/client"	16		"github.com/bnema/vev/internal/usecase/client"
16	)	17		)
17		18
		19	+	func TestBlockingRuntimeObserverDoesNotDelayTerminalFlushOrACK(t *testing.T) {
		20	+	var out bytes.Buffer
		21	+	var restores atomic.Int32
		22	+	resizeCh := make(chan domain.Size)
		23	+	flushed := make(chan struct{})
		24	+	term := portsmocks.NewMockTerminal(t)
		25	+	term.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
		26	+	term.EXPECT().EnterRaw().Return(func() error { restores.Add(1); return nil }, nil).Once()
		27	+	in := newBlockingReader()
		28	+	defer in.unblock()
		29	+	term.EXPECT().In().Return(in).Maybe()
		30	+	term.EXPECT().Out().Return(&out).Maybe()
		31	+	term.EXPECT().Flush().Run(func() { close(flushed) }).Return(nil).Once()
		32	+	term.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
		33	+
		34	+	observer := &blockingClientRuntimeObserver{entered: make(chan struct{}), release: make(chan struct{})}
		35	+	acked := make(chan struct{})
		36	+	transport := portsmocks.NewMockTransport(t)
		37	+	transport.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
		38	+	transport.EXPECT().Send(isType(ports.MsgAck)).Run(func(ports.Frame) { close(acked) }).Return(nil).Once()
		39	+	unblock := scriptRecv(transport,
		40	+	recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
		41	+	recvItem{f: frameOf(ports.MsgOutput, ports.MarshalOutput(ports.Output{Data: []byte("flush-before-observe"), NewStateNum: 3}))},
		42	+	)
		43	+	defer unblock()
		44	+	transport.EXPECT().Close().Return(nil).Once()
		45	+
		46	+	ctx, cancel := context.WithCancel(context.Background())
		47	+	result := make(chan error, 1)
		48	+	go func() {
		49	+	result <- client.Attach(ctx, transport, term, realClock{}, ports.IntentEphemeral, "", client.WithRuntimeObserver(observer))
		50	+	}()
		51	+	awaitClientRuntime(t, observer.entered, "blocking observer")
		52	+	awaitClientRuntime(t, flushed, "terminal flush")
		53	+	awaitClientRuntime(t, acked, "ACK")
		54	+	cancel()
		55	+	close(observer.release)
		56	+	select {
		57	+	case err := <-result:
		58	+	require.NoError(t, err)
		59	+	case <-time.After(time.Second):
		60	+	t.Fatal("Attach did not complete after observer release")
		61	+	}
		62	+	require.Equal(t, "flush-before-observe", out.String())
		63	+	require.Equal(t, int32(1), restores.Load())
		64	+	}
		65	+
18	func TestTerminalFlushBoundaryTransportObservability(t *testing.T) {	66		func TestTerminalFlushBoundaryTransportObservability(t *testing.T) {
19	var out bytes.Buffer	67		var out bytes.Buffer
20	var restores atomic.Int32	68		var restores atomic.Int32

@@ -43,6 +91,28 @@ func TestTerminalFlushBoundaryTransportObservability(t *testing.T) {
43	observer.requireOrdered(t, ports.RuntimeTerminalFlushed)	91		observer.requireOrdered(t, ports.RuntimeTerminalFlushed)
44	}	92		}
45		93
		94	+	type blockingClientRuntimeObserver struct {
		95	+	entered chan struct{}
		96	+	release chan struct{}
		97	+	once sync.Once
		98	+	}
		99	+
		100	+	func (o *blockingClientRuntimeObserver) ObserveRuntime(ports.RuntimeMark) {
		101	+	o.once.Do(func() {
		102	+	close(o.entered)
		103	+	<-o.release
		104	+	})
		105	+	}
		106	+
		107	+	func awaitClientRuntime(t *testing.T, ch <-chan struct{}, what string) {
		108	+	t.Helper()
		109	+	select {
		110	+	case <-ch:
		111	+	case <-time.After(time.Second):
		112	+	t.Fatalf("timed out waiting for %s", what)
		113	+	}
		114	+	}
		115	+
46	type clientRuntimeObserver struct {	116		type clientRuntimeObserver struct {
47	mu sync.Mutex	117		mu sync.Mutex
48	marks []ports.RuntimeMark	118		marks []ports.RuntimeMark

M internal/usecase/daemon/transport_observability_test.go +44 -23


@@ -6,30 +6,64 @@ import (
6	6		"testing"
7	7		"time"
8	8
	9	+	"github.com/bnema/vev/internal/domain"
9	10		"github.com/bnema/vev/internal/ports"
10	11		)
11	12
12		-	// This acceptance boundary deliberately configures the daemon with an
13		-	// observer only. The clock remains the daemon's existing behavior clock;
14		-	// timestamp ownership belongs to the concrete process observer.
	13	+	// This acceptance boundary exercises the attached resize, render, and ACK
	14	+	// path rather than merely listing the marks the implementation should emit.
15	15		func TestTransportObservabilityDaemonBoundaries(t *testing.T) {
16	16		observer := &daemonRuntimeObserver{}
17		-	d := New(nil, daemonRuntimeClock{now: time.Unix(0, 101)}, nil, WithRuntimeObserver(observer))
18		-	if d == nil {
19		-	t.Fatal("New() returned nil")
	17	+	d, sess, ac, sends := newManualSessionWithPTYs(t, &quietPTY{})
	18	+	d.runtimeObserver = observer
	19	+	// One in-flight state makes the second public resize take the ACK-blocked
	20	+	// path. The real coordinator owns both renders and the ACK wake.
	21	+	ac.output = newOutputStateStream(1)
	22	+	rc := d.attachCoordinator(sess, nil, ac, true)
	23	+
	24	+	if !d.resizeForFirstPaint(sess, ac, domain.Size{Cols: 90, Rows: 25}) {
	25	+	t.Fatal("first resize was not accepted")
	26	+	}
	27	+	first := <-sends
	28	+	if first.Type != ports.MsgOutput {
	29	+	t.Fatalf("first frame = %v, want output", first.Type)
	30	+	}
	31	+	output, err := ports.UnmarshalOutput(first.Payload)
	32	+	if err != nil {
	33	+	t.Fatalf("decode first output: %v", err)
	34	+	}
	35	+	if !d.resizeForFirstPaint(sess, ac, domain.Size{Cols: 100, Rows: 26}) {
	36	+	t.Fatal("second resize was not accepted")
	37	+	}
	38	+	ac.ackOutputState(output.NewStateNum)
	39	+	rc.notifyAck()
	40	+	second := <-sends
	41	+	if second.Type != ports.MsgOutput {
	42	+	t.Fatalf("second frame = %v, want output", second.Type)
20	43		}
21	44
22	45		want := []ports.RuntimeMarkKind{
23	46		ports.RuntimeResizeRequested, ports.RuntimeResizeCommitted,
	47	+	ports.RuntimeQueueEnqueued, ports.RuntimeQueueDequeued,
24	48		ports.RuntimeCaptureStart, ports.RuntimeCaptureEnd,
25	49		ports.RuntimeComposeStart, ports.RuntimeComposeEnd,
26	50		ports.RuntimeDiffStart, ports.RuntimeDiffEnd,
27		-	ports.RuntimeQueueEnqueued, ports.RuntimeQueueDequeued,
28		-	ports.RuntimeACKBlockedStart, ports.RuntimeACKBlockedEnd,
29	51		ports.RuntimeEmitStart, ports.RuntimeEmitEnd,
	52	+	ports.RuntimeResizeRequested, ports.RuntimeResizeCommitted,
	53	+	ports.RuntimeQueueEnqueued, ports.RuntimeACKBlockedStart,
	54	+	ports.RuntimeACKBlockedEnd, ports.RuntimeQueueDequeued,
	55	+	ports.RuntimeCaptureStart, ports.RuntimeCaptureEnd,
	56	+	ports.RuntimeComposeStart, ports.RuntimeComposeEnd,
	57	+	ports.RuntimeDiffStart, ports.RuntimeDiffEnd,
	58	+	ports.RuntimeEmitStart, ports.RuntimeEmitEnd,
	59	+	}
	60	+	if len(observer.marks) != len(want) {
	61	+	t.Fatalf("runtime marks = %#v, want kinds %#v", observer.marks, want)
30	62		}
31		-	if len(want) != 14 {
32		-	t.Fatalf("daemon boundary mark inventory changed: %d", len(want))
	63	+	for i, kind := range want {
	64	+	if observer.marks[i].Kind != kind {
	65	+	t.Fatalf("runtime mark %d = %q, want %q; all=%#v", i, observer.marks[i].Kind, kind, observer.marks)
	66	+	}
33	67		}
34	68		}
35	69

@@ -39,19 +73,6 @@ func (o *daemonRuntimeObserver) ObserveRuntime(mark ports.RuntimeMark) {
39	73		o.marks = append(o.marks, mark)
40	74		}
41	75
42		-	type daemonRuntimeClock struct{ now time.Time }
43		-
44		-	func (c daemonRuntimeClock) Now() time.Time { return c.now }
45		-	func (c daemonRuntimeClock) NewTimer(d time.Duration) ports.Timer {
46		-	return daemonRuntimeTimer{timer: time.NewTimer(d)}
47		-	}
48		-
49		-	type daemonRuntimeTimer struct{ timer *time.Timer }
50		-
51		-	func (t daemonRuntimeTimer) C() <-chan time.Time { return t.timer.C }
52		-	func (t daemonRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
53		-	func (t daemonRuntimeTimer) Stop() bool { return t.timer.Stop() }
54		-
55	76		type blockingDaemonRuntimeObserver struct {
56	77		mu sync.Mutex
57	78		entered chan struct{}

 	"testing"
 	"time"
+	"github.com/bnema/vev/internal/domain"
 	"github.com/bnema/vev/internal/ports"
 )
-// This acceptance boundary deliberately configures the daemon with an
+// This acceptance boundary exercises the attached resize, render, and ACK
-// observer only. The clock remains the daemon's existing behavior clock;
+// path rather than merely listing the marks the implementation should emit.
-// timestamp ownership belongs to the concrete process observer.
 func TestTransportObservabilityDaemonBoundaries(t *testing.T) {
 	observer := &daemonRuntimeObserver{}
-	d := New(nil, daemonRuntimeClock{now: time.Unix(0, 101)}, nil, WithRuntimeObserver(observer))
+	d, sess, ac, sends := newManualSessionWithPTYs(t, &quietPTY{})
-	if d == nil {
+	d.runtimeObserver = observer
-		t.Fatal("New() returned nil")
+	// One in-flight state makes the second public resize take the ACK-blocked
+	// path. The real coordinator owns both renders and the ACK wake.
+	ac.output = newOutputStateStream(1)
+	rc := d.attachCoordinator(sess, nil, ac, true)
+	if !d.resizeForFirstPaint(sess, ac, domain.Size{Cols: 90, Rows: 25}) {
+		t.Fatal("first resize was not accepted")
+	}
+	first := <-sends
+	if first.Type != ports.MsgOutput {
+		t.Fatalf("first frame = %v, want output", first.Type)
+	}
+	output, err := ports.UnmarshalOutput(first.Payload)
+	if err != nil {
+		t.Fatalf("decode first output: %v", err)
+	}
+	if !d.resizeForFirstPaint(sess, ac, domain.Size{Cols: 100, Rows: 26}) {
+		t.Fatal("second resize was not accepted")
+	}
+	ac.ackOutputState(output.NewStateNum)
+	rc.notifyAck()
+	second := <-sends
+	if second.Type != ports.MsgOutput {
+		t.Fatalf("second frame = %v, want output", second.Type)
 	}
 	want := []ports.RuntimeMarkKind{
 		ports.RuntimeResizeRequested, ports.RuntimeResizeCommitted,
+		ports.RuntimeQueueEnqueued, ports.RuntimeQueueDequeued,
 		ports.RuntimeCaptureStart, ports.RuntimeCaptureEnd,
 		ports.RuntimeComposeStart, ports.RuntimeComposeEnd,
 		ports.RuntimeDiffStart, ports.RuntimeDiffEnd,
-		ports.RuntimeQueueEnqueued, ports.RuntimeQueueDequeued,
-		ports.RuntimeACKBlockedStart, ports.RuntimeACKBlockedEnd,
 		ports.RuntimeEmitStart, ports.RuntimeEmitEnd,
+		ports.RuntimeResizeRequested, ports.RuntimeResizeCommitted,
+		ports.RuntimeQueueEnqueued, ports.RuntimeACKBlockedStart,
+		ports.RuntimeACKBlockedEnd, ports.RuntimeQueueDequeued,
+		ports.RuntimeCaptureStart, ports.RuntimeCaptureEnd,
+		ports.RuntimeComposeStart, ports.RuntimeComposeEnd,
+		ports.RuntimeDiffStart, ports.RuntimeDiffEnd,
+		ports.RuntimeEmitStart, ports.RuntimeEmitEnd,
+	}
+	if len(observer.marks) != len(want) {
+		t.Fatalf("runtime marks = %#v, want kinds %#v", observer.marks, want)
 	}
-	if len(want) != 14 {
+	for i, kind := range want {
-		t.Fatalf("daemon boundary mark inventory changed: %d", len(want))
+		if observer.marks[i].Kind != kind {
+			t.Fatalf("runtime mark %d = %q, want %q; all=%#v", i, observer.marks[i].Kind, kind, observer.marks)
+		}
 	}
 }
 	o.marks = append(o.marks, mark)
 }
-type daemonRuntimeClock struct{ now time.Time }
-func (c daemonRuntimeClock) Now() time.Time { return c.now }
-func (c daemonRuntimeClock) NewTimer(d time.Duration) ports.Timer {
-	return daemonRuntimeTimer{timer: time.NewTimer(d)}
-}
-type daemonRuntimeTimer struct{ timer *time.Timer }
-func (t daemonRuntimeTimer) C() <-chan time.Time        { return t.timer.C }
-func (t daemonRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
-func (t daemonRuntimeTimer) Stop() bool                 { return t.timer.Stop() }
 type blockingDaemonRuntimeObserver struct {
 	mu      sync.Mutex
 	entered chan struct{}