d833ed1 fix(obs): keep ACK progress independent of observer I/O · vev

M internal/app/run.go +34 -1


@@ -22,6 +22,7 @@ import (
22	22		"path/filepath"
23	23		"strconv"
24	24		"strings"
	25	+	"sync"
25	26		"syscall"
26	27		"text/tabwriter"
27	28		"time"

@@ -270,7 +271,39 @@ func performanceTrace(clk ports.Clock) (ports.RuntimeObserver, io.Closer, error)
270	271		_ = closer.Close()
271	272		return nil, nil, err
272	273		}
273		-	return observer, closer, nil
	274	+	reporter := ports.NewSerializedRuntimeObserver(observer, runtimeTraceQueueDepth)
	275	+	return reporter, &runtimeTraceCloser{reporter: reporter, closer: closer}, nil
	276	+	}
	277	+
	278	+	// runtimeTraceQueueDepth bounds all process-local trace producer handoffs.
	279	+	// A full queue emits the serialized diagnostic gap rather than delaying a
	280	+	// terminal, transport, or ACK progress path.
	281	+	const runtimeTraceQueueDepth = 256
	282	+
	283	+	// runtimeTraceCloser owns the two-stage trace shutdown: drain the reporter
	284	+	// before closing the concrete timestamp/file owner. sync.Once makes every
	285	+	// deferred and explicit close path safe without closing shared workers twice.
	286	+	type runtimeTraceCloser struct {
	287	+	reporter ports.SerializedRuntimeObserver
	288	+	closer io.Closer
	289	+	once sync.Once
	290	+	err error
	291	+	}
	292	+
	293	+	func (c *runtimeTraceCloser) Close() error {
	294	+	if c == nil {
	295	+	return nil
	296	+	}
	297	+	c.once.Do(func() {
	298	+	if c.reporter != nil {
	299	+	c.reporter.Flush()
	300	+	c.reporter.Close()
	301	+	}
	302	+	if c.closer != nil {
	303	+	c.err = c.closer.Close()
	304	+	}
	305	+	})
	306	+	return c.err
274	307		}
275	308
276	309		func configureLogging(component logging.Component, rotateAtRuntime bool) (*slog.Logger, io.Closer, error) {


@@ -22,6 +22,7 @@ import (
22		"path/filepath"	22		"path/filepath"
23		"strconv"	23		"strconv"
24		"strings"	24		"strings"
			25	+	"sync"
25		"syscall"	26		"syscall"
26		"text/tabwriter"	27		"text/tabwriter"
27		"time"	28		"time"

@@ -270,7 +271,39 @@ func performanceTrace(clk ports.Clock) (ports.RuntimeObserver, io.Closer, error)
270		_ = closer.Close()	271		_ = closer.Close()
271		return nil, nil, err	272		return nil, nil, err
272		}	273		}
273	-	return observer, closer, nil	274	+	reporter := ports.NewSerializedRuntimeObserver(observer, runtimeTraceQueueDepth)
			275	+	return reporter, &runtimeTraceCloser{reporter: reporter, closer: closer}, nil
			276	+	}
			277	+
			278	+	// runtimeTraceQueueDepth bounds all process-local trace producer handoffs.
			279	+	// A full queue emits the serialized diagnostic gap rather than delaying a
			280	+	// terminal, transport, or ACK progress path.
			281	+	const runtimeTraceQueueDepth = 256
			282	+
			283	+	// runtimeTraceCloser owns the two-stage trace shutdown: drain the reporter
			284	+	// before closing the concrete timestamp/file owner. sync.Once makes every
			285	+	// deferred and explicit close path safe without closing shared workers twice.
			286	+	type runtimeTraceCloser struct {
			287	+	reporter ports.SerializedRuntimeObserver
			288	+	closer io.Closer
			289	+	once sync.Once
			290	+	err error
			291	+	}
			292	+
			293	+	func (c *runtimeTraceCloser) Close() error {
			294	+	if c == nil {
			295	+	return nil
			296	+	}
			297	+	c.once.Do(func() {
			298	+	if c.reporter != nil {
			299	+	c.reporter.Flush()
			300	+	c.reporter.Close()
			301	+	}
			302	+	if c.closer != nil {
			303	+	c.err = c.closer.Close()
			304	+	}
			305	+	})
			306	+	return c.err
274		}	307		}
275			308
276		func configureLogging(component logging.Component, rotateAtRuntime bool) (*slog.Logger, io.Closer, error) {	309		func configureLogging(component logging.Component, rotateAtRuntime bool) (*slog.Logger, io.Closer, error) {

M internal/app/transport_observability_test.go +6 -0


@@ -29,10 +29,16 @@ func TestPerformanceTraceUsesHarnessProcessMapping(t *testing.T) {
29	29		if observer == nil \|\| closer == nil {
30	30		t.Fatal("performance trace was not configured")
31	31		}
	32	+	if _, ok := observer.(ports.SerializedRuntimeObserver); !ok {
	33	+	t.Fatalf("performance trace observer = %T, want serialized owner", observer)
	34	+	}
32	35		observer.ObserveRuntime(ports.RuntimeMark{Schema: ports.RuntimeMarkSchema, Component: "daemon", Scenario: "runtime", Run: 1, Kind: ports.RuntimeEmitEnd, Valid: true})
33	36		if err := closer.Close(); err != nil {
34	37		t.Fatal(err)
35	38		}
	39	+	if err := closer.Close(); err != nil {
	40	+	t.Fatal(err)
	41	+	}
36	42		f, err := os.Open(path)
37	43		if err != nil {
38	44		t.Fatal(err)


@@ -29,10 +29,16 @@ func TestPerformanceTraceUsesHarnessProcessMapping(t *testing.T) {
29	if observer == nil \|\| closer == nil {	29		if observer == nil \|\| closer == nil {
30	t.Fatal("performance trace was not configured")	30		t.Fatal("performance trace was not configured")
31	}	31		}
		32	+	if _, ok := observer.(ports.SerializedRuntimeObserver); !ok {
		33	+	t.Fatalf("performance trace observer = %T, want serialized owner", observer)
		34	+	}
32	observer.ObserveRuntime(ports.RuntimeMark{Schema: ports.RuntimeMarkSchema, Component: "daemon", Scenario: "runtime", Run: 1, Kind: ports.RuntimeEmitEnd, Valid: true})	35		observer.ObserveRuntime(ports.RuntimeMark{Schema: ports.RuntimeMarkSchema, Component: "daemon", Scenario: "runtime", Run: 1, Kind: ports.RuntimeEmitEnd, Valid: true})
33	if err := closer.Close(); err != nil {	36		if err := closer.Close(); err != nil {
34	t.Fatal(err)	37		t.Fatal(err)
35	}	38		}
		39	+	if err := closer.Close(); err != nil {
		40	+	t.Fatal(err)
		41	+	}
36	f, err := os.Open(path)	42		f, err := os.Open(path)
37	if err != nil {	43		if err != nil {
38	t.Fatal(err)	44		t.Fatal(err)

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


@@ -88,6 +88,21 @@ func NewSerializedRuntimeObserver(observer RuntimeObserver, capacity int) Serial
88	88		return o
89	89		}
90	90
	91	+	// EnsureSerializedRuntimeObserver returns observer's existing serialized
	92	+	// owner when it has one. Otherwise it creates an owned worker. Callers close
	93	+	// the result only when owned is true, preventing nested workers and duplicate
	94	+	// lifecycle ownership when application wiring already serialized the process
	95	+	// observer.
	96	+	func EnsureSerializedRuntimeObserver(observer RuntimeObserver, capacity int) (serialized SerializedRuntimeObserver, owned bool) {
	97	+	if observer == nil {
	98	+	return nil, false
	99	+	}
	100	+	if serialized, ok := observer.(SerializedRuntimeObserver); ok {
	101	+	return serialized, false
	102	+	}
	103	+	return NewSerializedRuntimeObserver(observer, capacity), true
	104	+	}
	105	+
91	106		type serializedRuntimeObserver struct {
92	107		observer RuntimeObserver
93	108		capacity int


@@ -88,6 +88,21 @@ func NewSerializedRuntimeObserver(observer RuntimeObserver, capacity int) Serial
88	return o	88		return o
89	}	89		}
90		90
		91	+	// EnsureSerializedRuntimeObserver returns observer's existing serialized
		92	+	// owner when it has one. Otherwise it creates an owned worker. Callers close
		93	+	// the result only when owned is true, preventing nested workers and duplicate
		94	+	// lifecycle ownership when application wiring already serialized the process
		95	+	// observer.
		96	+	func EnsureSerializedRuntimeObserver(observer RuntimeObserver, capacity int) (serialized SerializedRuntimeObserver, owned bool) {
		97	+	if observer == nil {
		98	+	return nil, false
		99	+	}
		100	+	if serialized, ok := observer.(SerializedRuntimeObserver); ok {
		101	+	return serialized, false
		102	+	}
		103	+	return NewSerializedRuntimeObserver(observer, capacity), true
		104	+	}
		105	+
91	type serializedRuntimeObserver struct {	106		type serializedRuntimeObserver struct {
92	observer RuntimeObserver	107		observer RuntimeObserver
93	capacity int	108		capacity int

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


@@ -65,6 +65,15 @@ func TestSerializedRuntimeObserverFlushesInOrderAndCloses(t *testing.T) {
65	65		}
66	66		}
67	67
	68	+	func TestEnsureSerializedRuntimeObserverReusesExistingOwner(t *testing.T) {
	69	+	existing := NewSerializedRuntimeObserver(runtimeObserverFunc(func(RuntimeMark) {}), 1)
	70	+	reporter, owned := EnsureSerializedRuntimeObserver(existing, 1)
	71	+	if owned \|\| reporter != existing {
	72	+	t.Fatalf("EnsureSerializedRuntimeObserver() = (%T, %t), want existing unowned reporter", reporter, owned)
	73	+	}
	74	+	existing.Close()
	75	+	}
	76	+
68	77		func TestSerializedRuntimeObserverReportsBoundedQueueLoss(t *testing.T) {
69	78		entered := make(chan struct{})
70	79		release := make(chan struct{})


@@ -65,6 +65,15 @@ func TestSerializedRuntimeObserverFlushesInOrderAndCloses(t *testing.T) {
65	}	65		}
66	}	66		}
67		67
		68	+	func TestEnsureSerializedRuntimeObserverReusesExistingOwner(t *testing.T) {
		69	+	existing := NewSerializedRuntimeObserver(runtimeObserverFunc(func(RuntimeMark) {}), 1)
		70	+	reporter, owned := EnsureSerializedRuntimeObserver(existing, 1)
		71	+	if owned \|\| reporter != existing {
		72	+	t.Fatalf("EnsureSerializedRuntimeObserver() = (%T, %t), want existing unowned reporter", reporter, owned)
		73	+	}
		74	+	existing.Close()
		75	+	}
		76	+
68	func TestSerializedRuntimeObserverReportsBoundedQueueLoss(t *testing.T) {	77		func TestSerializedRuntimeObserverReportsBoundedQueueLoss(t *testing.T) {
69	entered := make(chan struct{})	78		entered := make(chan struct{})
70	release := make(chan struct{})	79		release := make(chan struct{})

M internal/usecase/client/client.go +7 -3


@@ -145,9 +145,13 @@ func RunWithRuntimeObserver(ctx context.Context, dialer ports.Dialer, term ports
145	145		}
146	146
147	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()
	148	+	if reporter, owned := ports.EnsureSerializedRuntimeObserver(observer, runtimeObserverQueueDepth); reporter != nil {
	149	+	if owned {
	150	+	defer func() {
	151	+	reporter.Flush()
	152	+	reporter.Close()
	153	+	}()
	154	+	}
151	155		observer = reporter
152	156		}
153	157		if log == nil {

 }
 func run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, clk ports.Clock, intent uint8, name string, remote bool, clipboard ports.ClipboardReader, log *slog.Logger, observer ports.RuntimeObserver) (retErr error) {
-	if observer != nil {
+	if reporter, owned := ports.EnsureSerializedRuntimeObserver(observer, runtimeObserverQueueDepth); reporter != nil {
-		reporter := ports.NewSerializedRuntimeObserver(observer, runtimeObserverQueueDepth)
+		if owned {
-		defer reporter.Close()
+			defer func() {
+				reporter.Flush()
+				reporter.Close()
+			}()
+		}
 		observer = reporter
 	}
 	if log == nil {

M internal/usecase/daemon/daemon.go +54 -23


@@ -22,8 +22,9 @@ import (
22	22		// Scheduler debounce bounds. Idle updates use the minimum for low latency;
23	23		// sustained floods step toward the maximum to reduce frame/syscall pressure.
24	24		const (
25		-	minDebounceInterval = 2 * time.Millisecond
26		-	maxSyncUpdateDuration = 500 * time.Millisecond
	25	+	minDebounceInterval = 2 * time.Millisecond
	26	+	maxSyncUpdateDuration = 500 * time.Millisecond
	27	+	runtimeObserverQueueDepth = 64
27	28		)
28	29
29	30		const debounceInterval = minDebounceInterval

@@ -92,26 +93,28 @@ type Daemon struct {
92	93		// interval between JRS validation and its committed hand-off.
93	94		beforeRecentSessionHandoff func()
94	95
95		-	ptys ports.PTYFactory
96		-	clock ports.Clock
97		-	log *slog.Logger
98		-	runtimeObserver ports.RuntimeObserver
99		-	baseEnv []string
100		-	shell string
101		-	shellArgs []string
102		-	persist *persist.Persister
103		-	persistEnabled bool
104		-	snaps ports.SnapshotStore
105		-	snapsEnabled bool
106		-	snapshotMarshal func(snapcodec.Session) ([]byte, error)
107		-	snapshotJobs chan *snapshotCapture
108		-	snapshotWorkerMu sync.Mutex
109		-	snapshotWorkerID uint64
110		-	snapshotWorkerCtx context.Context
111		-	snapshotWorkerCancel context.CancelFunc
112		-	snapshotWorkerDone chan struct{}
113		-	snapshotWorkerFlush chan struct{}
114		-	snapshotWorkerFinalWake chan struct{}
	96	+	ptys ports.PTYFactory
	97	+	clock ports.Clock
	98	+	log *slog.Logger
	99	+	runtimeObserver ports.RuntimeObserver
	100	+	runtimeObserverCloser ports.SerializedRuntimeObserver
	101	+	runtimeObserverCloseOnce sync.Once
	102	+	baseEnv []string
	103	+	shell string
	104	+	shellArgs []string
	105	+	persist *persist.Persister
	106	+	persistEnabled bool
	107	+	snaps ports.SnapshotStore
	108	+	snapsEnabled bool
	109	+	snapshotMarshal func(snapcodec.Session) ([]byte, error)
	110	+	snapshotJobs chan *snapshotCapture
	111	+	snapshotWorkerMu sync.Mutex
	112	+	snapshotWorkerID uint64
	113	+	snapshotWorkerCtx context.Context
	114	+	snapshotWorkerCancel context.CancelFunc
	115	+	snapshotWorkerDone chan struct{}
	116	+	snapshotWorkerFlush chan struct{}
	117	+	snapshotWorkerFinalWake chan struct{}
115	118		// snapshotFinalJobs retains immutable terminal captures rejected by the
116	119		// bounded regular queue. It is only populated during session teardown and
117	120		// drained by the same worker, so persistence remains outside daemon locks.

@@ -187,7 +190,34 @@ type Option func(*Daemon)
187	190		// WithRuntimeObserver enables opt-in process-local marks. It takes no clock:
188	191		// the concrete observer is the sole timestamp owner.
189	192		func WithRuntimeObserver(observer ports.RuntimeObserver) Option {
190		-	return func(d *Daemon) { d.runtimeObserver = observer }
	193	+	return func(d *Daemon) {
	194	+	// Options are applied before the daemon starts. Replacing an owned
	195	+	// observer here cannot race producers and must retire its worker.
	196	+	if d.runtimeObserverCloser != nil {
	197	+	d.runtimeObserverCloser.Close()
	198	+	d.runtimeObserverCloser = nil
	199	+	}
	200	+	reporter, owned := ports.EnsureSerializedRuntimeObserver(observer, runtimeObserverQueueDepth)
	201	+	d.runtimeObserver = reporter
	202	+	if owned {
	203	+	d.runtimeObserverCloser = reporter
	204	+	}
	205	+	}
	206	+	}
	207	+
	208	+	// closeRuntimeObserver drains the daemon-owned reporter before teardown
	209	+	// completes. Application-owned process reporters are closed by their app
	210	+	// lifecycle instead, so shared observers never receive a duplicate close.
	211	+	func (d *Daemon) closeRuntimeObserver() {
	212	+	if d == nil {
	213	+	return
	214	+	}
	215	+	d.runtimeObserverCloseOnce.Do(func() {
	216	+	if d.runtimeObserverCloser != nil {
	217	+	d.runtimeObserverCloser.Flush()
	218	+	d.runtimeObserverCloser.Close()
	219	+	}
	220	+	})
191	221		}
192	222
193	223		// WithShell overrides the command (and its args) each session spawns. The

@@ -363,6 +393,7 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt
363	393		// returns when the last session is removed or ctx is cancelled; on the latter
364	394		// path attached clients are detached with ReasonServerShutdown.
365	395		func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
	396	+	defer d.closeRuntimeObserver()
366	397		d.serveCtx, d.serveCancel = context.WithCancel(ctx)
367	398		defer d.serveCancel()
368	399		d.hardCtx, d.hardCancel = context.WithCancel(context.Background())

 // Scheduler debounce bounds. Idle updates use the minimum for low latency;
 // sustained floods step toward the maximum to reduce frame/syscall pressure.
 const (
-	minDebounceInterval   = 2 * time.Millisecond
+	minDebounceInterval       = 2 * time.Millisecond
-	maxSyncUpdateDuration = 500 * time.Millisecond
+	maxSyncUpdateDuration     = 500 * time.Millisecond
+	runtimeObserverQueueDepth = 64
 )
 const debounceInterval = minDebounceInterval
 	// interval between JRS validation and its committed hand-off.
 	beforeRecentSessionHandoff func()
-	ptys                    ports.PTYFactory
+	ptys                     ports.PTYFactory
-	clock                   ports.Clock
+	clock                    ports.Clock
-	log                     *slog.Logger
+	log                      *slog.Logger
-	runtimeObserver         ports.RuntimeObserver
+	runtimeObserver          ports.RuntimeObserver
-	baseEnv                 []string
+	runtimeObserverCloser    ports.SerializedRuntimeObserver
-	shell                   string
+	runtimeObserverCloseOnce sync.Once
-	shellArgs               []string
+	baseEnv                  []string
-	persist                 *persist.Persister
+	shell                    string
-	persistEnabled          bool
+	shellArgs                []string
-	snaps                   ports.SnapshotStore
+	persist                  *persist.Persister
-	snapsEnabled            bool
+	persistEnabled           bool
-	snapshotMarshal         func(snapcodec.Session) ([]byte, error)
+	snaps                    ports.SnapshotStore
-	snapshotJobs            chan *snapshotCapture
+	snapsEnabled             bool
-	snapshotWorkerMu        sync.Mutex
+	snapshotMarshal          func(snapcodec.Session) ([]byte, error)
-	snapshotWorkerID        uint64
+	snapshotJobs             chan *snapshotCapture
-	snapshotWorkerCtx       context.Context
+	snapshotWorkerMu         sync.Mutex
-	snapshotWorkerCancel    context.CancelFunc
+	snapshotWorkerID         uint64
-	snapshotWorkerDone      chan struct{}
+	snapshotWorkerCtx        context.Context
-	snapshotWorkerFlush     chan struct{}
+	snapshotWorkerCancel     context.CancelFunc
-	snapshotWorkerFinalWake chan struct{}
+	snapshotWorkerDone       chan struct{}
+	snapshotWorkerFlush      chan struct{}
+	snapshotWorkerFinalWake  chan struct{}
 	// snapshotFinalJobs retains immutable terminal captures rejected by the
 	// bounded regular queue. It is only populated during session teardown and
 	// drained by the same worker, so persistence remains outside daemon locks.
 // WithRuntimeObserver enables opt-in process-local marks. It takes no clock:
 // the concrete observer is the sole timestamp owner.
 func WithRuntimeObserver(observer ports.RuntimeObserver) Option {
-	return func(d *Daemon) { d.runtimeObserver = observer }
+	return func(d *Daemon) {
+		// Options are applied before the daemon starts. Replacing an owned
+		// observer here cannot race producers and must retire its worker.
+		if d.runtimeObserverCloser != nil {
+			d.runtimeObserverCloser.Close()
+			d.runtimeObserverCloser = nil
+		}
+		reporter, owned := ports.EnsureSerializedRuntimeObserver(observer, runtimeObserverQueueDepth)
+		d.runtimeObserver = reporter
+		if owned {
+			d.runtimeObserverCloser = reporter
+		}
+	}
+}
+// closeRuntimeObserver drains the daemon-owned reporter before teardown
+// completes. Application-owned process reporters are closed by their app
+// lifecycle instead, so shared observers never receive a duplicate close.
+func (d *Daemon) closeRuntimeObserver() {
+	if d == nil {
+		return
+	}
+	d.runtimeObserverCloseOnce.Do(func() {
+		if d.runtimeObserverCloser != nil {
+			d.runtimeObserverCloser.Flush()
+			d.runtimeObserverCloser.Close()
+		}
+	})
 }
 // WithShell overrides the command (and its args) each session spawns. The
 // returns when the last session is removed or ctx is cancelled; on the latter
 // path attached clients are detached with ReasonServerShutdown.
 func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
+	defer d.closeRuntimeObserver()
 	d.serveCtx, d.serveCancel = context.WithCancel(ctx)
 	defer d.serveCancel()
 	d.hardCtx, d.hardCancel = context.WithCancel(context.Background())

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


@@ -3,6 +3,7 @@ package daemon
3	3		import (
4	4		"runtime"
5	5		"sync"
	6	+	"sync/atomic"
6	7		"testing"
7	8		"time"
8	9

@@ -222,6 +223,67 @@ func TestTransportObservabilityACKBlockedSpansEndExactlyOnce(t *testing.T) {
222	223		}
223	224		}
224	225
	226	+	// A raw observer supplied through daemon configuration is serialized by the
	227	+	// daemon. Once Start has been published, a blocked End must not delay the ACK
	228	+	// that consumes deferred work and delivers its wake.
	229	+	func TestTransportObservabilityBlockedACKEndDoesNotDelayNotifyAck(t *testing.T) {
	230	+	observer := &blockingACKEndObserver{
	231	+	startObserved: make(chan struct{}),
	232	+	endEntered: make(chan struct{}),
	233	+	releaseEnd: make(chan struct{}),
	234	+	}
	235	+	d := New(nil, nil, nil, WithRuntimeObserver(observer))
	236	+	wakes := make(chan renderWake, 1)
	237	+	var ready atomic.Bool
	238	+	rc := newRenderCoordinator(renderCoordinatorOptions{
	239	+	observer: d.runtimeObserver,
	240	+	ackReady: ready.Load,
	241	+	wake: func(w renderWake) { wakes <- w },
	242	+	})
	243	+	rc.attach(&attachedClient{})
	244	+	rc.invalidate(renderInvalidation{class: invalidateOutput})
	245	+
	246	+	// Enter the ACK-blocked state and wait until Start has reached the injected
	247	+	// observer. End is the only deliberately blocked call in this test.
	248	+	rc.fire(rc.generation, false, true)
	249	+	awaitDaemonObserver(t, observer.startObserved, "ACK-blocked start observer")
	250	+
	251	+	ready.Store(true)
	252	+	ackDone := make(chan struct{})
	253	+	go func() {
	254	+	rc.notifyAck()
	255	+	close(ackDone)
	256	+	}()
	257	+	awaitDaemonObserver(t, observer.endEntered, "blocked ACK-blocked end observer")
	258	+	awaitDaemonObserver(t, ackDone, "nonblocking ACK progress")
	259	+	awaitWake(t, wakes)
	260	+	rc.mu.Lock()
	261	+	pending := rc.pending
	262	+	rc.mu.Unlock()
	263	+	if pending {
	264	+	t.Fatal("notifyAck left consumed ACK-deferred work pending")
	265	+	}
	266	+
	267	+	close(observer.releaseEnd)
	268	+	d.closeRuntimeObserver()
	269	+	}
	270	+
	271	+	type blockingACKEndObserver struct {
	272	+	startObserved chan struct{}
	273	+	endEntered chan struct{}
	274	+	releaseEnd chan struct{}
	275	+	}
	276	+
	277	+	func (o *blockingACKEndObserver) ObserveRuntime(mark ports.RuntimeMark) {
	278	+	switch mark.Kind {
	279	+	case ports.RuntimeACKBlockedStart:
	280	+	close(o.startObserved)
	281	+	case ports.RuntimeACKBlockedEnd:
	282	+	close(o.endEntered)
	283	+	<-o.releaseEnd
	284	+	}
	285	+	}
	286	+
225	287		func TestTransportObservabilityBlockedCoordinatorObserverReleasesLock(t *testing.T) {
226	288		observer := newBlockingDaemonRuntimeObserver()
227	289		rc := newRenderCoordinator(renderCoordinatorOptions{


@@ -3,6 +3,7 @@ package daemon
3	import (	3		import (
4	"runtime"	4		"runtime"
5	"sync"	5		"sync"
		6	+	"sync/atomic"
6	"testing"	7		"testing"
7	"time"	8		"time"
8		9

@@ -222,6 +223,67 @@ func TestTransportObservabilityACKBlockedSpansEndExactlyOnce(t *testing.T) {
222	}	223		}
223	}	224		}
224		225
		226	+	// A raw observer supplied through daemon configuration is serialized by the
		227	+	// daemon. Once Start has been published, a blocked End must not delay the ACK
		228	+	// that consumes deferred work and delivers its wake.
		229	+	func TestTransportObservabilityBlockedACKEndDoesNotDelayNotifyAck(t *testing.T) {
		230	+	observer := &blockingACKEndObserver{
		231	+	startObserved: make(chan struct{}),
		232	+	endEntered: make(chan struct{}),
		233	+	releaseEnd: make(chan struct{}),
		234	+	}
		235	+	d := New(nil, nil, nil, WithRuntimeObserver(observer))
		236	+	wakes := make(chan renderWake, 1)
		237	+	var ready atomic.Bool
		238	+	rc := newRenderCoordinator(renderCoordinatorOptions{
		239	+	observer: d.runtimeObserver,
		240	+	ackReady: ready.Load,
		241	+	wake: func(w renderWake) { wakes <- w },
		242	+	})
		243	+	rc.attach(&attachedClient{})
		244	+	rc.invalidate(renderInvalidation{class: invalidateOutput})
		245	+
		246	+	// Enter the ACK-blocked state and wait until Start has reached the injected
		247	+	// observer. End is the only deliberately blocked call in this test.
		248	+	rc.fire(rc.generation, false, true)
		249	+	awaitDaemonObserver(t, observer.startObserved, "ACK-blocked start observer")
		250	+
		251	+	ready.Store(true)
		252	+	ackDone := make(chan struct{})
		253	+	go func() {
		254	+	rc.notifyAck()
		255	+	close(ackDone)
		256	+	}()
		257	+	awaitDaemonObserver(t, observer.endEntered, "blocked ACK-blocked end observer")
		258	+	awaitDaemonObserver(t, ackDone, "nonblocking ACK progress")
		259	+	awaitWake(t, wakes)
		260	+	rc.mu.Lock()
		261	+	pending := rc.pending
		262	+	rc.mu.Unlock()
		263	+	if pending {
		264	+	t.Fatal("notifyAck left consumed ACK-deferred work pending")
		265	+	}
		266	+
		267	+	close(observer.releaseEnd)
		268	+	d.closeRuntimeObserver()
		269	+	}
		270	+
		271	+	type blockingACKEndObserver struct {
		272	+	startObserved chan struct{}
		273	+	endEntered chan struct{}
		274	+	releaseEnd chan struct{}
		275	+	}
		276	+
		277	+	func (o *blockingACKEndObserver) ObserveRuntime(mark ports.RuntimeMark) {
		278	+	switch mark.Kind {
		279	+	case ports.RuntimeACKBlockedStart:
		280	+	close(o.startObserved)
		281	+	case ports.RuntimeACKBlockedEnd:
		282	+	close(o.endEntered)
		283	+	<-o.releaseEnd
		284	+	}
		285	+	}
		286	+
225	func TestTransportObservabilityBlockedCoordinatorObserverReleasesLock(t *testing.T) {	287		func TestTransportObservabilityBlockedCoordinatorObserverReleasesLock(t *testing.T) {
226	observer := newBlockingDaemonRuntimeObserver()	288		observer := newBlockingDaemonRuntimeObserver()
227	rc := newRenderCoordinator(renderCoordinatorOptions{	289		rc := newRenderCoordinator(renderCoordinatorOptions{