3f30dae fix(obs): preserve runtime span correlation · vev

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


@@ -97,7 +97,7 @@ type RuntimeCorrelationInputs struct {
97	97
98	98		// NewRuntimeCorrelationObserver binds every mark from one OS process to its
99	99		// harness manifest scenario/run. Operation IDs remain producer-owned when
100		-	// supplied; missing IDs are assigned from this process-local sequence.
	100	+	// supplied; missing IDs are assigned from the process-wide allocator.
101	101		func NewRuntimeCorrelationObserver(observer RuntimeObserver, inputs RuntimeCorrelationInputs) (RuntimeObserver, error) {
102	102		if observer == nil \|\| inputs.Scenario == "" \|\| inputs.Run == 0 {
103	103		return nil, errors.New("invalid runtime correlation inputs")

@@ -108,7 +108,6 @@ func NewRuntimeCorrelationObserver(observer RuntimeObserver, inputs RuntimeCorre
108	108		type runtimeCorrelationObserver struct {
109	109		observer RuntimeObserver
110	110		inputs RuntimeCorrelationInputs
111		-	sequence atomic.Uint64
112	111		}
113	112
114	113		func (o *runtimeCorrelationObserver) ObserveRuntime(mark RuntimeMark) {

@@ -117,7 +116,7 @@ func (o *runtimeCorrelationObserver) ObserveRuntime(mark RuntimeMark) {
117	116		}
118	117		mark.Scenario, mark.Run = o.inputs.Scenario, o.inputs.Run
119	118		if mark.Sequence == 0 {
120		-	mark.Sequence = o.sequence.Add(1)
	119	+	mark.Sequence = runtimeMarkSequence.Add(1)
121	120		}
122	121		if mark.RequestID == 0 {
123	122		mark.RequestID = mark.Sequence

 // NewRuntimeCorrelationObserver binds every mark from one OS process to its
 // harness manifest scenario/run. Operation IDs remain producer-owned when
-// supplied; missing IDs are assigned from this process-local sequence.
+// supplied; missing IDs are assigned from the process-wide allocator.
 func NewRuntimeCorrelationObserver(observer RuntimeObserver, inputs RuntimeCorrelationInputs) (RuntimeObserver, error) {
 	if observer == nil || inputs.Scenario == "" || inputs.Run == 0 {
 		return nil, errors.New("invalid runtime correlation inputs")
 type runtimeCorrelationObserver struct {
 	observer RuntimeObserver
 	inputs   RuntimeCorrelationInputs
-	sequence atomic.Uint64
 }
 func (o *runtimeCorrelationObserver) ObserveRuntime(mark RuntimeMark) {
 	}
 	mark.Scenario, mark.Run = o.inputs.Scenario, o.inputs.Run
 	if mark.Sequence == 0 {
-		mark.Sequence = o.sequence.Add(1)
+		mark.Sequence = runtimeMarkSequence.Add(1)
 	}
 	if mark.RequestID == 0 {
 		mark.RequestID = mark.Sequence

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


@@ -88,6 +88,34 @@ func TestRuntimeCorrelationObserverUsesHarnessProcessInputs(t *testing.T) {
88	88		}
89	89		}
90	90
	91	+	func TestRuntimeCorrelationObserverAllocatesMissingIDsProcessWide(t *testing.T) {
	92	+	var got []RuntimeMark
	93	+	collect := runtimeObserverFunc(func(mark RuntimeMark) { got = append(got, mark) })
	94	+	first, err := NewRuntimeCorrelationObserver(collect, RuntimeCorrelationInputs{Scenario: "scenario", Run: 1})
	95	+	if err != nil {
	96	+	t.Fatal(err)
	97	+	}
	98	+	second, err := NewRuntimeCorrelationObserver(collect, RuntimeCorrelationInputs{Scenario: "scenario", Run: 1})
	99	+	if err != nil {
	100	+	t.Fatal(err)
	101	+	}
	102	+
	103	+	first.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "first", Kind: RuntimeEmitStart, Valid: true})
	104	+	second.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "second", Kind: RuntimeEmitEnd, Valid: true})
	105	+
	106	+	if len(got) != 2 {
	107	+	t.Fatalf("marks=%+v", got)
	108	+	}
	109	+	if got[0].Sequence == got[1].Sequence {
	110	+	t.Fatalf("wrapper-local sequence collision: %+v", got)
	111	+	}
	112	+	for _, mark := range got {
	113	+	if mark.RequestID != mark.Sequence \|\| mark.Epoch != mark.RequestID {
	114	+	t.Fatalf("missing-ID fallback changed: %+v", mark)
	115	+	}
	116	+	}
	117	+	}
	118	+
91	119		func TestRuntimeCorrelationObserverRejectsInvalidHarnessInputs(t *testing.T) {
92	120		for _, in := range []RuntimeCorrelationInputs{{}, {Scenario: "scenario"}} {
93	121		if _, err := NewRuntimeCorrelationObserver(runtimeObserverFunc(func(RuntimeMark) {}), in); err == nil {


@@ -88,6 +88,34 @@ func TestRuntimeCorrelationObserverUsesHarnessProcessInputs(t *testing.T) {
88	}	88		}
89	}	89		}
90		90
		91	+	func TestRuntimeCorrelationObserverAllocatesMissingIDsProcessWide(t *testing.T) {
		92	+	var got []RuntimeMark
		93	+	collect := runtimeObserverFunc(func(mark RuntimeMark) { got = append(got, mark) })
		94	+	first, err := NewRuntimeCorrelationObserver(collect, RuntimeCorrelationInputs{Scenario: "scenario", Run: 1})
		95	+	if err != nil {
		96	+	t.Fatal(err)
		97	+	}
		98	+	second, err := NewRuntimeCorrelationObserver(collect, RuntimeCorrelationInputs{Scenario: "scenario", Run: 1})
		99	+	if err != nil {
		100	+	t.Fatal(err)
		101	+	}
		102	+
		103	+	first.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "first", Kind: RuntimeEmitStart, Valid: true})
		104	+	second.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "second", Kind: RuntimeEmitEnd, Valid: true})
		105	+
		106	+	if len(got) != 2 {
		107	+	t.Fatalf("marks=%+v", got)
		108	+	}
		109	+	if got[0].Sequence == got[1].Sequence {
		110	+	t.Fatalf("wrapper-local sequence collision: %+v", got)
		111	+	}
		112	+	for _, mark := range got {
		113	+	if mark.RequestID != mark.Sequence \|\| mark.Epoch != mark.RequestID {
		114	+	t.Fatalf("missing-ID fallback changed: %+v", mark)
		115	+	}
		116	+	}
		117	+	}
		118	+
91	func TestRuntimeCorrelationObserverRejectsInvalidHarnessInputs(t *testing.T) {	119		func TestRuntimeCorrelationObserverRejectsInvalidHarnessInputs(t *testing.T) {
92	for _, in := range []RuntimeCorrelationInputs{{}, {Scenario: "scenario"}} {	120		for _, in := range []RuntimeCorrelationInputs{{}, {Scenario: "scenario"}} {
93	if _, err := NewRuntimeCorrelationObserver(runtimeObserverFunc(func(RuntimeMark) {}), in); err == nil {	121		if _, err := NewRuntimeCorrelationObserver(runtimeObserverFunc(func(RuntimeMark) {}), in); err == nil {

M internal/usecase/daemon/render_coordinator.go +17 -7


@@ -652,6 +652,9 @@ func (c *renderCoordinator) wakeCurrent(w renderWake) bool {
652	652		func (c renderCoordinator) noteDetach(ac attachedClient) {
653	653		c.mu.Lock()
654	654		var timer, resizeTimer, retryTimer ports.Timer
	655	+	var observer ports.RuntimeObserver
	656	+	var queueCorrelation, ackCorrelation ports.RuntimeCorrelation
	657	+	var queueMarked, ackBlocked bool
655	658		if c.attachment == ac {
656	659		c.advanceAttachmentEpochLocked(ac)
657	660		c.attachment = nil

@@ -677,10 +680,10 @@ func (c renderCoordinator) noteDetach(ac attachedClient) {
677	680		}
678	681		retryTimer, c.retryTimer = c.retryTimer, nil
679	682		c.retryGen++
	683	+	observer, queueCorrelation, queueMarked = c.opts.observer, c.queueCorrelation, c.queueMarked
	684	+	ackCorrelation, ackBlocked = c.ackCorrelation, c.ackBlocked
	685	+	c.queueMarked, c.ackBlocked = false, false
680	686		}
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
684	687		c.mu.Unlock()
685	688		stopTimer(timer)
686	689		stopTimer(resizeTimer)

@@ -701,6 +704,9 @@ func (c renderCoordinator) noteReplace(old, replacement attachedClient, readin
701	704		ready = readiness[0]
702	705		}
703	706		c.mu.Lock()
	707	+	var observer ports.RuntimeObserver
	708	+	var queueCorrelation, ackCorrelation ports.RuntimeCorrelation
	709	+	var queueMarked, ackBlocked bool
704	710		// A coordinator may be installed while replacing a legacy attachment
705	711		// which predated coordinator ownership. In that case nil has no pending
706	712		// identity to invalidate, and the replacement becomes the first bound

@@ -734,10 +740,10 @@ func (c renderCoordinator) noteReplace(old, replacement attachedClient, readin
734	740		}
735	741		retryTimer, c.retryTimer = c.retryTimer, nil
736	742		c.retryGen++
	743	+	observer, queueCorrelation, queueMarked = c.opts.observer, c.queueCorrelation, c.queueMarked
	744	+	ackCorrelation, ackBlocked = c.ackCorrelation, c.ackBlocked
	745	+	c.queueMarked, c.ackBlocked = false, false
737	746		}
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
741	747		c.mu.Unlock()
742	748		stopTimer(timer)
743	749		stopTimer(resizeTimer)

@@ -1133,13 +1139,17 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) {
1133	1139		timer := c.detachNormalTimerLocked()
1134	1140		c.pending, c.pendingReset, c.pendingUrgent, c.ackDeferred, c.deadlineDue, c.coalesced, c.armed = false, false, false, false, false, 0, false
1135	1141		observer, queueCorrelation, queueMarked := c.opts.observer, c.queueCorrelation, c.queueMarked
1136		-	c.queueMarked = false
	1142	+	ackCorrelation, ackBlocked := c.ackCorrelation, c.ackBlocked
	1143	+	c.queueMarked, c.ackBlocked = false, false
1137	1144		wake := c.opts.wake
1138	1145		c.mu.Unlock()
1139	1146		stopTimer(timer)
1140	1147		if queueMarked && observer != nil {
1141	1148		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, true))
1142	1149		}
	1150	+	if ackBlocked && observer != nil {
	1151	+	observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, true))
	1152	+	}
1143	1153		if wake != nil && !headlessPreviewOnly {
1144	1154		wake(w)
1145	1155		}

 func (c *renderCoordinator) noteDetach(ac *attachedClient) {
 	c.mu.Lock()
 	var timer, resizeTimer, retryTimer ports.Timer
+	var observer ports.RuntimeObserver
+	var queueCorrelation, ackCorrelation ports.RuntimeCorrelation
+	var queueMarked, ackBlocked bool
 	if c.attachment == ac {
 		c.advanceAttachmentEpochLocked(ac)
 		c.attachment = nil
 		}
 		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
 	}
-	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)
 		ready = readiness[0]
 	}
 	c.mu.Lock()
+	var observer ports.RuntimeObserver
+	var queueCorrelation, ackCorrelation ports.RuntimeCorrelation
+	var queueMarked, ackBlocked bool
 	// A coordinator may be installed while replacing a legacy attachment
 	// which predated coordinator ownership. In that case nil has no pending
 	// identity to invalidate, and the replacement becomes the first bound
 		}
 		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
 	}
-	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)
 	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
+	ackCorrelation, ackBlocked := c.ackCorrelation, c.ackBlocked
+	c.queueMarked, c.ackBlocked = false, 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 ackBlocked && observer != nil {
+		observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackCorrelation, ports.RuntimeACKBlockedEnd, 0, true))
+	}
 	if wake != nil && !headlessPreviewOnly {
 		wake(w)
 	}

M internal/usecase/daemon/render_pipeline.go +1 -1


@@ -288,7 +288,7 @@ func commitDamageReceipts(receipts []damageReceipt) {
288	288		// sendMu for the complete capture/compose/emit transaction.
289	289		func (d Daemon) emitFrame(sess session, ac attachedClient, state capturedRenderState, composed composedRenderFrame, batches ...*runtimeMarkBatch) bool {
290	290		var ownedMarks runtimeMarkBatch
291		-	marks := (*runtimeMarkBatch)(nil)
	291	+	var marks *runtimeMarkBatch
292	292		if len(batches) != 0 {
293	293		marks = batches[0]
294	294		} else {


@@ -288,7 +288,7 @@ func commitDamageReceipts(receipts []damageReceipt) {
288		// sendMu for the complete capture/compose/emit transaction.	288		// sendMu for the complete capture/compose/emit transaction.
289		func (d Daemon) emitFrame(sess session, ac attachedClient, state capturedRenderState, composed composedRenderFrame, batches ...*runtimeMarkBatch) bool {	289		func (d Daemon) emitFrame(sess session, ac attachedClient, state capturedRenderState, composed composedRenderFrame, batches ...*runtimeMarkBatch) bool {
290		var ownedMarks runtimeMarkBatch	290		var ownedMarks runtimeMarkBatch
291	-	marks := (*runtimeMarkBatch)(nil)	291	+	var marks *runtimeMarkBatch
292		if len(batches) != 0 {	292		if len(batches) != 0 {
293		marks = batches[0]	293		marks = batches[0]
294		} else {	294		} else {

M internal/usecase/daemon/transport_observability_test.go +61 -0


@@ -140,6 +140,67 @@ func TestTransportObservabilityBlockedRenderObserverReleasesArchitectureLocks(t
140	140
141	141		// ACK-blocked start is captured while coordinator state changes, then emitted
142	142		// after c.mu is unlocked. This closes the remaining coordinator-observer path.
	143	+	func TestTransportObservabilityACKBlockedSpansEndExactlyOnce(t *testing.T) {
	144	+	cases := []struct {
	145	+	name string
	146	+	finish func(renderCoordinator, bool)
	147	+	}{
	148	+	{
	149	+	name: "notify ACK",
	150	+	finish: func(rc renderCoordinator, ready bool) {
	151	+	*ready = true
	152	+	rc.notifyAck()
	153	+	},
	154	+	},
	155	+	{
	156	+	name: "successful consume",
	157	+	finish: func(rc renderCoordinator, ready bool) {
	158	+	*ready = true
	159	+	rc.fire(rc.generation, false, false)
	160	+	},
	161	+	},
	162	+	{
	163	+	name: "teardown",
	164	+	finish: func(rc renderCoordinator, ready bool) {
	165	+	rc.noteSessionTeardown()
	166	+	},
	167	+	},
	168	+	}
	169	+	for _, tt := range cases {
	170	+	t.Run(tt.name, func(t *testing.T) {
	171	+	observer := &daemonRuntimeObserver{}
	172	+	ready := false
	173	+	rc := newRenderCoordinator(renderCoordinatorOptions{
	174	+	observer: observer,
	175	+	ackReady: func() bool { return ready },
	176	+	})
	177	+	rc.attach(&attachedClient{})
	178	+	rc.invalidate(renderInvalidation{class: invalidateOutput})
	179	+	rc.fire(rc.generation, false, true)
	180	+	tt.finish(rc, &ready)
	181	+	// A late notification after consuming or tearing down must not emit a
	182	+	// second end for the already-cleared blocked interval.
	183	+	rc.notifyAck()
	184	+
	185	+	var starts, ends []ports.RuntimeMark
	186	+	for _, mark := range observer.marks {
	187	+	switch mark.Kind {
	188	+	case ports.RuntimeACKBlockedStart:
	189	+	starts = append(starts, mark)
	190	+	case ports.RuntimeACKBlockedEnd:
	191	+	ends = append(ends, mark)
	192	+	}
	193	+	}
	194	+	if len(starts) != 1 \|\| len(ends) != 1 {
	195	+	t.Fatalf("ACK-blocked span marks: starts=%+v ends=%+v all=%+v", starts, ends, observer.marks)
	196	+	}
	197	+	if starts[0].Sequence != ends[0].Sequence \|\| starts[0].RequestID != ends[0].RequestID \|\| starts[0].Epoch != ends[0].Epoch {
	198	+	t.Fatalf("ACK-blocked span correlation changed: start=%+v end=%+v", starts[0], ends[0])
	199	+	}
	200	+	})
	201	+	}
	202	+	}
	203	+
143	204		func TestTransportObservabilityBlockedCoordinatorObserverReleasesLock(t *testing.T) {
144	205		observer := newBlockingDaemonRuntimeObserver()
145	206		rc := newRenderCoordinator(renderCoordinatorOptions{


@@ -140,6 +140,67 @@ func TestTransportObservabilityBlockedRenderObserverReleasesArchitectureLocks(t
140		140
141	// ACK-blocked start is captured while coordinator state changes, then emitted	141		// ACK-blocked start is captured while coordinator state changes, then emitted
142	// after c.mu is unlocked. This closes the remaining coordinator-observer path.	142		// after c.mu is unlocked. This closes the remaining coordinator-observer path.
		143	+	func TestTransportObservabilityACKBlockedSpansEndExactlyOnce(t *testing.T) {
		144	+	cases := []struct {
		145	+	name string
		146	+	finish func(renderCoordinator, bool)
		147	+	}{
		148	+	{
		149	+	name: "notify ACK",
		150	+	finish: func(rc renderCoordinator, ready bool) {
		151	+	*ready = true
		152	+	rc.notifyAck()
		153	+	},
		154	+	},
		155	+	{
		156	+	name: "successful consume",
		157	+	finish: func(rc renderCoordinator, ready bool) {
		158	+	*ready = true
		159	+	rc.fire(rc.generation, false, false)
		160	+	},
		161	+	},
		162	+	{
		163	+	name: "teardown",
		164	+	finish: func(rc renderCoordinator, ready bool) {
		165	+	rc.noteSessionTeardown()
		166	+	},
		167	+	},
		168	+	}
		169	+	for _, tt := range cases {
		170	+	t.Run(tt.name, func(t *testing.T) {
		171	+	observer := &daemonRuntimeObserver{}
		172	+	ready := false
		173	+	rc := newRenderCoordinator(renderCoordinatorOptions{
		174	+	observer: observer,
		175	+	ackReady: func() bool { return ready },
		176	+	})
		177	+	rc.attach(&attachedClient{})
		178	+	rc.invalidate(renderInvalidation{class: invalidateOutput})
		179	+	rc.fire(rc.generation, false, true)
		180	+	tt.finish(rc, &ready)
		181	+	// A late notification after consuming or tearing down must not emit a
		182	+	// second end for the already-cleared blocked interval.
		183	+	rc.notifyAck()
		184	+
		185	+	var starts, ends []ports.RuntimeMark
		186	+	for _, mark := range observer.marks {
		187	+	switch mark.Kind {
		188	+	case ports.RuntimeACKBlockedStart:
		189	+	starts = append(starts, mark)
		190	+	case ports.RuntimeACKBlockedEnd:
		191	+	ends = append(ends, mark)
		192	+	}
		193	+	}
		194	+	if len(starts) != 1 \|\| len(ends) != 1 {
		195	+	t.Fatalf("ACK-blocked span marks: starts=%+v ends=%+v all=%+v", starts, ends, observer.marks)
		196	+	}
		197	+	if starts[0].Sequence != ends[0].Sequence \|\| starts[0].RequestID != ends[0].RequestID \|\| starts[0].Epoch != ends[0].Epoch {
		198	+	t.Fatalf("ACK-blocked span correlation changed: start=%+v end=%+v", starts[0], ends[0])
		199	+	}
		200	+	})
		201	+	}
		202	+	}
		203	+
143	func TestTransportObservabilityBlockedCoordinatorObserverReleasesLock(t *testing.T) {	204		func TestTransportObservabilityBlockedCoordinatorObserverReleasesLock(t *testing.T) {
144	observer := newBlockingDaemonRuntimeObserver()	205		observer := newBlockingDaemonRuntimeObserver()
145	rc := newRenderCoordinator(renderCoordinatorOptions{	206		rc := newRenderCoordinator(renderCoordinatorOptions{