1package daemon
2
3import (
4 "os"
5 "path/filepath"
6 "sync"
7 "sync/atomic"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/mock"
12 "github.com/stretchr/testify/require"
13
14 "github.com/bnema/vev/internal/domain"
15 "github.com/bnema/vev/internal/ports"
16 portsmocks "github.com/bnema/vev/internal/ports/mocks"
17 "github.com/bnema/vev/internal/usecase/keys"
18 "github.com/bnema/vev/internal/usecase/layout"
19)
20
21// --- coordinator harness ------------------------------------------------------
22
23// coordinatorMockClock uses generated mocks while retaining deterministic timer
24// channels and recorded deadlines for coordinator scheduling assertions.
25type coordinatorMockClock struct {
26 clock *portsmocks.MockClock
27 timers chan *coordinatorMockTimer
28 handshakeTimers chan *coordinatorMockTimer
29}
30
31type coordinatorMockTimer struct {
32 mock *portsmocks.MockTimer
33 ch chan time.Time
34 duration time.Duration
35 stopped chan struct{}
36 stopOnce sync.Once
37}
38
39func newCoordinatorMockClock(t *testing.T, capacity int) *coordinatorMockClock {
40 t.Helper()
41 clk := &coordinatorMockClock{
42 clock: portsmocks.NewMockClock(t),
43 timers: make(chan *coordinatorMockTimer, capacity),
44 handshakeTimers: make(chan *coordinatorMockTimer, 1),
45 }
46 clk.clock.EXPECT().Now().Return(time.Time{}).Maybe()
47 clk.clock.EXPECT().NewTimer(mock.MatchedBy(func(d time.Duration) bool {
48 return d == ports.HandshakeTimeout ||
49 d == urgentRenderDeadline ||
50 (d >= minOutputRenderDeadline && d <= maxOutputRenderDeadline) ||
51 d == maxSyncUpdateDuration ||
52 d == time.Second ||
53 d == 15*time.Minute
54 })).RunAndReturn(func(d time.Duration) ports.Timer {
55 timer := &coordinatorMockTimer{
56 mock: portsmocks.NewMockTimer(t),
57 duration: d,
58 stopped: make(chan struct{}),
59 }
60 timer.ch = make(chan time.Time, 1)
61 // A normal deadline reads C both in its worker and in the inert-clock
62 // guard; a watchdog reads it in its worker. The generated expectation
63 // makes every channel access observable without a hand-written Timer.
64 timer.mock.EXPECT().C().Maybe().Return((<-chan time.Time)(timer.ch))
65 timer.mock.EXPECT().Stop().Run(func() {
66 timer.stopOnce.Do(func() { close(timer.stopped) })
67 }).Maybe().Return(true)
68 if d == ports.HandshakeTimeout {
69 clk.handshakeTimers <- timer
70 } else {
71 clk.timers <- timer
72 }
73 return timer.mock
74 }).Maybe()
75 return clk
76}
77
78// TestRenderCoordinatorTimerCallbacksReenterCoordinator is deliberately a
79// deadlock regression: every external timer method synchronously reads
80// coordinator metadata. Before two-phase timer ownership this blocked while
81// c.mu was held during normal arming, urgent replacement, sync arming, and
82// detach cancellation.
83// Detach cancels and stops a selected worker but never synchronously joins it;
84// the stale worker must be rejected when its external readiness probe returns.
85func TestRenderCoordinatorDetachDoesNotJoinSelectedDeadlineWorker(t *testing.T) {
86 clock := portsmocks.NewMockClock(t)
87 timer := portsmocks.NewMockTimer(t)
88 timerC := make(chan time.Time, 1)
89 stopCalled := make(chan struct{})
90 clock.EXPECT().NewTimer(minOutputRenderDeadline).Return(timer).Once()
91 timer.EXPECT().C().Return((<-chan time.Time)(timerC)).Once()
92 timer.EXPECT().Stop().Run(func() { close(stopCalled) }).Return(true).Once()
93
94 entered := make(chan struct{})
95 release := make(chan struct{})
96 rc := newRenderCoordinator(renderCoordinatorOptions{
97 clock: clock,
98 ackReady: func() bool {
99 close(entered)
100 <-release
101 return true
102 },
103 })
104 ac := &attachedClient{}
105 rc.attach(ac)
106 rc.invalidate(renderInvalidation{class: invalidateOutput})
107 timerC <- time.Time{}
108 <-entered
109
110 detached := make(chan struct{})
111 go func() {
112 rc.noteDetach(ac)
113 close(detached)
114 }()
115 <-stopCalled
116 awaitTestCompletion(t, detached, "detach synchronously waited for a selected timer worker")
117 close(release)
118}
119
120func TestRenderCoordinatorTimerCallbacksReenterCoordinator(t *testing.T) {
121 clock := portsmocks.NewMockClock(t)
122 var rc *renderCoordinator
123 clock.EXPECT().NewTimer(mock.Anything).RunAndReturn(func(time.Duration) ports.Timer {
124 timer := portsmocks.NewMockTimer(t)
125 ch := make(chan time.Time)
126 timer.EXPECT().C().Run(func() {
127 _ = rc.resizeSnapshot()
128 _ = rc.burstMetricsSnapshot()
129 }).Return((<-chan time.Time)(ch)).Once()
130 timer.EXPECT().Stop().Run(func() {
131 _ = rc.resizeSnapshot()
132 _ = rc.burstMetricsSnapshot()
133 }).Return(true).Once()
134 return timer
135 }).Times(3)
136 rc = newRenderCoordinator(renderCoordinatorOptions{clock: clock})
137 ac := &attachedClient{}
138 rc.attach(ac)
139
140 rc.invalidate(renderInvalidation{class: invalidateOutput}) // normal arm
141 rc.invalidate(renderInvalidation{class: invalidateUrgent}) // urgent promotion
142 rc.noteSyncBegin(nil, 1) // sync arm
143 rc.noteDetach(ac) // cancel normal deadline only
144 rc.noteSyncPaneRemoved(nil) // pane destruction cancels sync watchdog
145}
146
147// coordinatorHarness wires one coordinator to generated clock/timer mocks and
148// recording hooks. Every assertion is channel- or counter-based; nothing sleeps.
149type coordinatorHarness struct {
150 clk *coordinatorMockClock
151 wakes chan renderWake
152 previews chan renderWake
153 ackReady atomic.Bool
154 syncActive atomic.Bool
155 rc *renderCoordinator
156}
157
158func newCoordinatorHarness(t *testing.T) *coordinatorHarness {
159 t.Helper()
160 h := &coordinatorHarness{
161 clk: newCoordinatorMockClock(t, 16),
162 wakes: make(chan renderWake, 16),
163 previews: make(chan renderWake, 16),
164 }
165 h.ackReady.Store(true)
166 h.rc = newRenderCoordinator(renderCoordinatorOptions{
167 clock: h.clk.clock,
168 wake: func(w renderWake) { h.wakes <- w },
169 ackReady: func() bool { return h.ackReady.Load() },
170 syncActive: func() bool { return h.syncActive.Load() },
171 })
172 return h
173}
174
175// armedTimers drains every deadline armed so far. Arming happens
176// synchronously inside the coordinator call, so a non-blocking drain is
177// deterministic.
178func (h *coordinatorHarness) armedTimers(t *testing.T) []*coordinatorMockTimer {
179 t.Helper()
180 var timers []*coordinatorMockTimer
181 for {
182 select {
183 case tm := <-h.clk.timers:
184 timers = append(timers, tm)
185 default:
186 return timers
187 }
188 }
189}
190
191// Coordinator calls and fake-clock advancement are synchronous test steps.
192// These helpers deliberately never wait on wall time: a missing event is a
193// deterministic contract failure, not a slow behavior to poll for.
194func awaitWake(t *testing.T, ch chan renderWake) renderWake {
195 t.Helper()
196 wake := awaitTestValue(t, ch, "coordinator did not publish a wake after fake-clock advancement")
197 wake.attachmentLeases = nil
198 wake.generation = 0
199 return wake
200}
201
202func requireNoWake(t *testing.T, ch chan renderWake) {
203 t.Helper()
204 select {
205 case w := <-ch:
206 t.Fatalf("unexpected coordinator wake: %+v", w)
207 default:
208 }
209}
210
211func awaitInvalidation(t *testing.T, ch chan renderInvalidation) renderInvalidation {
212 t.Helper()
213 return awaitTestValue(t, ch, "producer did not publish a coordinator invalidation")
214}
215
216func requireNoInvalidation(t *testing.T, ch chan renderInvalidation) {
217 t.Helper()
218 select {
219 case inv := <-ch:
220 t.Fatalf("unexpected coordinator invalidation: %+v", inv)
221 default:
222 }
223}
224
225func captureResizeCallbackDone(t *testing.T, rc *renderCoordinator) <-chan struct{} {
226 t.Helper()
227 rc.mu.Lock()
228 defer rc.mu.Unlock()
229 if rc.resizeLane.token == nil {
230 t.Fatal("coordinator did not publish a resize callback completion")
231 }
232 return rc.resizeLane.token.done
233}
234
235func awaitHandshakeTimer(t *testing.T, clk *coordinatorMockClock) *coordinatorMockTimer {
236 t.Helper()
237 select {
238 case tm := <-clk.handshakeTimers:
239 return tm
240 default:
241 t.Fatal("handshake did not synchronously arm a fake-clock timer")
242 return nil
243 }
244}
245
246func awaitCoordinatorScheduledTimer(t *testing.T, clk *coordinatorMockClock) *coordinatorMockTimer {
247 t.Helper()
248 select {
249 case tm := <-clk.timers:
250 return tm
251 default:
252 t.Fatal("coordinator did not synchronously arm a fake-clock timer")
253 return nil
254 }
255}
256
257func awaitLatestCoordinatorTimer(t *testing.T, clk *coordinatorMockClock) *coordinatorMockTimer {
258 t.Helper()
259 timer := awaitCoordinatorScheduledTimer(t, clk)
260 for {
261 select {
262 case timer = <-clk.timers:
263 default:
264 return timer
265 }
266 }
267}
268
269func requireNoCoordinatorOutputFrame(t *testing.T, sends chan ports.Frame) {
270 t.Helper()
271 for {
272 select {
273 case frame := <-sends:
274 if frame.Type == ports.MsgRoutePosition {
275 continue
276 }
277 t.Fatalf("unexpected output frame: %+v", frame)
278 default:
279 return
280 }
281 }
282}
283
284// --- wake coalescing and deadlines --------------------------------------------
285
286func TestRenderCoordinatorCoalescesWakesUnderOneDeadline(t *testing.T) {
287 cases := []struct {
288 name string
289 invalidations []renderInvalidation
290 wantTimers int
291 wantMin time.Duration
292 wantMax time.Duration
293 wantWake renderWake
294 }{
295 {
296 name: "urgent transition wakes within two milliseconds",
297 invalidations: []renderInvalidation{{class: invalidateUrgent, reset: true, producer: "input.go"}},
298 wantTimers: 1,
299 wantMax: urgentRenderDeadline,
300 wantWake: renderWake{reset: true, urgent: true, coalesced: 1},
301 },
302 {
303 name: "output burst coalesces into one adaptive deadline",
304 invalidations: []renderInvalidation{
305 {class: invalidateOutput, producer: "render.go"},
306 {class: invalidateOutput, producer: "render.go"},
307 {class: invalidateOutput, producer: "render.go"},
308 },
309 wantTimers: 1,
310 wantMin: minOutputRenderDeadline,
311 wantMax: maxOutputRenderDeadline,
312 wantWake: renderWake{coalesced: 3},
313 },
314 {
315 name: "full redraw stays sticky across later incremental transitions",
316 invalidations: []renderInvalidation{
317 {class: invalidateOutput},
318 {class: invalidateOutput, reset: true},
319 {class: invalidateOutput},
320 },
321 wantTimers: 1,
322 wantMin: minOutputRenderDeadline,
323 wantMax: maxOutputRenderDeadline,
324 wantWake: renderWake{reset: true, coalesced: 3},
325 },
326 {
327 name: "urgent transition tightens a pending output deadline",
328 invalidations: []renderInvalidation{
329 {class: invalidateOutput},
330 {class: invalidateUrgent},
331 },
332 wantTimers: 2,
333 wantMax: urgentRenderDeadline,
334 wantWake: renderWake{urgent: true, coalesced: 2},
335 },
336 }
337 for _, tc := range cases {
338 t.Run(tc.name, func(t *testing.T) {
339 h := newCoordinatorHarness(t)
340 for _, inv := range tc.invalidations {
341 h.rc.invalidate(inv)
342 }
343 timers := h.armedTimers(t)
344 require.Lenf(t, timers, tc.wantTimers,
345 "cap-1 coalescing must arm exactly %d deadline timer(s) for %d invalidations",
346 tc.wantTimers, len(tc.invalidations))
347 last := timers[len(timers)-1]
348 require.GreaterOrEqual(t, last.duration, tc.wantMin)
349 require.LessOrEqual(t, last.duration, tc.wantMax)
350
351 last.ch <- time.Time{}
352 require.Equal(t, tc.wantWake, awaitWake(t, h.wakes),
353 "one wake must deliver the latest coalesced state")
354 requireNoWake(t, h.wakes)
355 })
356 }
357}
358
359func TestRenderCoordinatorAdaptsOutputDeadlineAfterBurstAndDecays(t *testing.T) {
360 h := newCoordinatorHarness(t)
361
362 // A burst establishes pressure without extending its already-armed first
363 // deadline; the following output deadline grows to the 16ms ceiling.
364 for range 9 {
365 h.rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"})
366 }
367 burst := h.armedTimers(t)
368 require.Len(t, burst, 1)
369 require.Equal(t, minOutputRenderDeadline, burst[0].duration)
370 burst[0].ch <- time.Time{}
371 awaitWake(t, h.wakes)
372
373 h.rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"})
374 atCap := h.armedTimers(t)
375 require.Len(t, atCap, 1)
376 require.Equal(t, maxOutputRenderDeadline, atCap[0].duration,
377 "burst pressure must be capped at the 16ms output deadline")
378 atCap[0].ch <- time.Time{}
379 awaitWake(t, h.wakes)
380
381 // Quiet singleton batches decay pressure toward the 8ms floor.
382 var previous = maxOutputRenderDeadline
383 for range 8 {
384 h.rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"})
385 timers := h.armedTimers(t)
386 require.Len(t, timers, 1)
387 require.LessOrEqual(t, timers[0].duration, previous)
388 require.GreaterOrEqual(t, timers[0].duration, minOutputRenderDeadline)
389 previous = timers[0].duration
390 timers[0].ch <- time.Time{}
391 awaitWake(t, h.wakes)
392 }
393 require.Equal(t, minOutputRenderDeadline, previous)
394}
395
396func TestRenderCoordinatorUrgentDeadlineCannotBeExtended(t *testing.T) {
397 h := newCoordinatorHarness(t)
398 h.rc.invalidate(renderInvalidation{class: invalidateOutput})
399 h.rc.invalidate(renderInvalidation{class: invalidateUrgent})
400 for range 32 {
401 h.rc.invalidate(renderInvalidation{class: invalidateOutput})
402 h.rc.invalidate(renderInvalidation{class: invalidateUrgent})
403 }
404 timers := h.armedTimers(t)
405 require.Len(t, timers, 2, "only the urgent promotion may replace the output timer")
406 require.Equal(t, urgentRenderDeadline, timers[len(timers)-1].duration)
407}
408
409// --- ACK gating ----------------------------------------------------------------
410
411func TestRenderCoordinatorAckReadinessIsAttachmentScoped(t *testing.T) {
412 slow, healthy := &attachedClient{}, &attachedClient{}
413 var slowReady atomic.Bool
414 wakes := make(chan renderWake, 2)
415 rc := newRenderCoordinator(renderCoordinatorOptions{
416 ackReadyFor: func(ac *attachedClient) bool { return ac == healthy || slowReady.Load() },
417 wake: func(w renderWake) { wakes <- w },
418 })
419 rc.attach(slow)
420 rc.attach(healthy)
421 rc.invalidate(renderInvalidation{class: invalidateUrgent, reset: true, producer: "test"})
422 rc.fireCurrent(false)
423
424 wake := awaitTestValue(t, wakes, "healthy attachment did not receive an independent wake")
425 require.Contains(t, wake.attachmentLeases, healthy)
426 _, slowSelected := wake.attachmentLeases[slow]
427 require.False(t, slowSelected)
428 rc.mu.Lock()
429 require.True(t, rc.pending, "slow attachment must retain the shared mutation until its ACK window opens")
430 rc.mu.Unlock()
431
432 slowReady.Store(true)
433 rc.notifyAckForLease(rc.attachmentLease(slow))
434 wake = awaitTestValue(t, wakes, "slow attachment did not receive its deferred wake")
435 require.Contains(t, wake.attachmentLeases, slow)
436 rc.mu.Lock()
437 require.False(t, rc.pending, "the shared mutation must clear after every attachment catches up")
438 rc.mu.Unlock()
439}
440
441func TestRenderCoordinatorAckReadinessReentersWithoutBlockingResize(t *testing.T) {
442 owner := &attachedClient{}
443 wakes := make(chan renderWake, 1)
444 ackEntered := make(chan struct{})
445 resizeDone := make(chan uint64, 1)
446 fireDone := make(chan struct{})
447
448 var rc *renderCoordinator
449 rc = newRenderCoordinator(renderCoordinatorOptions{
450 ackReady: func() bool {
451 close(ackEntered)
452 // The readiness probe models the output send path reading coordinator
453 // metadata after sendMu is held. It must never run under c.mu.
454 _ = rc.resizeSnapshot()
455 return true
456 },
457 wake: func(w renderWake) { wakes <- w },
458 })
459 rc.attach(owner)
460 rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"})
461
462 go func() {
463 rc.fire(1, false, true)
464 close(fireDone)
465 }()
466 awaitTestCompletion(t, ackEntered, "ack readiness was not evaluated")
467
468 go func() { resizeDone <- rc.recordResizeRequest(domain.Size{Cols: 120, Rows: 40}, owner) }()
469 epoch := awaitTestValue(t, resizeDone, "ack readiness re-entry blocked resize metadata")
470 require.Equal(t, uint64(1), epoch)
471 wake := awaitWake(t, wakes)
472 require.Nil(t, wake.lease, "shared wakes do not carry an attachment lease")
473 require.Equal(t, renderWake{coalesced: 1}, wake)
474 awaitTestCompletion(t, fireDone, "fire did not return after wake")
475}
476
477func TestRenderCoordinatorAckDoesNotBypassAnUnexpiredDeadline(t *testing.T) {
478 h := newCoordinatorHarness(t)
479 h.rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"})
480 timer := awaitCoordinatorScheduledTimer(t, h.clk)
481 require.GreaterOrEqual(t, timer.duration, minOutputRenderDeadline)
482
483 // An unrelated ACK must not turn a normal output deadline into an
484 // immediate wake or cancel its timer.
485 h.rc.notifyAck()
486 requireNoWake(t, h.wakes)
487 timer.mock.AssertNotCalled(t, "Stop")
488
489 timer.ch <- time.Time{}
490 awaitWake(t, h.wakes)
491 requireNoWake(t, h.wakes)
492}
493
494func TestRenderCoordinatorAckFlushesOnlyExpiredAckDeferredWork(t *testing.T) {
495 t.Run("expired deadline flushes exactly once after readiness", func(t *testing.T) {
496 h := newCoordinatorHarness(t)
497 h.ackReady.Store(false)
498 h.rc.invalidate(renderInvalidation{class: invalidateOutput, reset: true, producer: "render.go"})
499 timer := awaitCoordinatorScheduledTimer(t, h.clk)
500 timer.ch <- time.Time{}
501 requireNoWake(t, h.wakes)
502
503 // Repeated ACKs while capacity remains unavailable are no-ops.
504 h.rc.notifyAck()
505 requireNoWake(t, h.wakes)
506 h.ackReady.Store(true)
507 h.rc.notifyAck()
508 w := awaitWake(t, h.wakes)
509 require.True(t, w.reset)
510 requireNoWake(t, h.wakes)
511 h.rc.notifyAck()
512 requireNoWake(t, h.wakes)
513 })
514
515 t.Run("lifecycle clears deferred work and urgent explicit fires stay immediate", func(t *testing.T) {
516 h := newCoordinatorHarness(t)
517 h.ackReady.Store(false)
518 h.rc.invalidate(renderInvalidation{class: invalidateOutput})
519 timer := awaitCoordinatorScheduledTimer(t, h.clk)
520 timer.ch <- time.Time{}
521 requireNoWake(t, h.wakes)
522 h.rc.beginSessionTeardown().finish()
523 h.ackReady.Store(true)
524 h.rc.notifyAck()
525 requireNoWake(t, h.wakes)
526
527 immediate := newCoordinatorHarness(t)
528 immediate.rc.invalidate(renderInvalidation{class: invalidateUrgent})
529 immediate.rc.fireCurrent(false)
530 w := awaitWake(t, immediate.wakes)
531 require.True(t, w.urgent)
532 })
533}
534
535func TestRenderCoordinatorAckGateBlocksCompositionUntilAck(t *testing.T) {
536 h := newCoordinatorHarness(t)
537 h.ackReady.Store(false)
538
539 h.rc.invalidate(renderInvalidation{class: invalidateOutput})
540 h.rc.invalidate(renderInvalidation{class: invalidateOutput, reset: true})
541 h.rc.invalidate(renderInvalidation{class: invalidateUrgent})
542 for _, timer := range h.armedTimers(t) {
543 timer.ch <- time.Time{}
544 }
545 requireNoWake(t, h.wakes)
546
547 h.ackReady.Store(true)
548 h.rc.notifyAck()
549 w := awaitWake(t, h.wakes)
550 require.True(t, w.reset, "the deferred full redraw must stay sticky through the ack gate")
551 require.Equal(t, 3, w.coalesced, "the post-ack wake must carry every deferred transition")
552 requireNoWake(t, h.wakes)
553
554 h.rc.notifyAck()
555 requireNoWake(t, h.wakes)
556}
557
558// --- synchronized output ---------------------------------------------------------
559
560func TestRenderCoordinatorConcurrentSynchronizedPaneBatches(t *testing.T) {
561 t.Run("end orders, watchdogs, and stale callbacks are isolated per pane", func(t *testing.T) {
562 h := newCoordinatorHarness(t)
563 firstForced, secondForced := make(chan struct{}, 1), make(chan struct{}, 1)
564 firstPane, secondPane := &pane{}, &pane{}
565
566 // Distinct stable pane owners must remain independent.
567 h.rc.noteSyncBegin(firstPane, 11, func() { firstForced <- struct{}{} })
568 firstWatchdog := h.armedTimers(t)[0]
569 h.rc.noteSyncBegin(secondPane, 22, func() { secondForced <- struct{}{} })
570 secondWatchdog := h.armedTimers(t)[0]
571 h.rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"})
572
573 // Completing either pane leaves the aggregate gate closed.
574 h.rc.noteSyncEnd(firstPane, 11)
575 requireNoWake(t, h.wakes)
576 // A stale callback for the completed pane must not force it or wake.
577 firstWatchdog.ch <- time.Time{}
578 select {
579 case <-firstForced:
580 t.Fatal("stale watchdog forced a completed pane")
581 default:
582 }
583 requireNoWake(t, h.wakes)
584
585 // The remaining pane's watchdog forces only its own batch, then the
586 // aggregate completion produces exactly one urgent wake.
587 secondWatchdog.ch <- time.Time{}
588 <-secondForced
589 w := awaitWake(t, h.wakes)
590 require.True(t, w.watchdog)
591 require.True(t, w.urgent, "aggregate completion must wake urgently")
592 requireNoWake(t, h.wakes)
593 })
594
595 t.Run("later pane can end first without releasing an earlier pane", func(t *testing.T) {
596 h := newCoordinatorHarness(t)
597 firstForced := make(chan struct{}, 1)
598 firstPane, secondPane := &pane{}, &pane{}
599 h.rc.noteSyncBegin(firstPane, 11, func() { firstForced <- struct{}{} })
600 firstWatchdog := h.armedTimers(t)[0]
601 h.rc.noteSyncBegin(secondPane, 22)
602 h.rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"})
603
604 h.rc.noteSyncEnd(secondPane, 22)
605 requireNoWake(t, h.wakes)
606 firstWatchdog.ch <- time.Time{}
607 <-firstForced
608 w := awaitWake(t, h.wakes)
609 require.True(t, w.watchdog)
610 require.True(t, w.urgent, "final pane completion must wake urgently")
611 requireNoWake(t, h.wakes)
612 })
613
614 t.Run("pane removal cancels only that pane watchdog", func(t *testing.T) {
615 h := newCoordinatorHarness(t)
616 firstPane, secondPane := &pane{}, &pane{}
617 firstForced, secondForced := make(chan struct{}, 1), make(chan struct{}, 1)
618 h.rc.noteSyncBegin(firstPane, 11, func() { firstForced <- struct{}{} })
619 firstWatchdog := h.armedTimers(t)[0]
620 h.rc.noteSyncBegin(secondPane, 22, func() { secondForced <- struct{}{} })
621 secondWatchdog := h.armedTimers(t)[0]
622 h.rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"})
623
624 h.rc.noteSyncPaneRemoved(firstPane)
625 firstWatchdog.ch <- time.Time{}
626 select {
627 case <-firstForced:
628 t.Fatal("removed pane watchdog forced output")
629 default:
630 }
631 requireNoWake(t, h.wakes)
632 secondWatchdog.ch <- time.Time{}
633 <-secondForced
634 require.True(t, awaitWake(t, h.wakes).watchdog)
635 })
636}
637
638func TestRenderCoordinatorSyncGateFollowsDynamicRenderability(t *testing.T) {
639 t.Run("inactive batch that becomes active gates activation output until end", func(t *testing.T) {
640 h := newCoordinatorHarness(t)
641 var visible atomic.Bool
642 p := &pane{}
643 h.rc.noteSyncBeginWithRenderability(p, 1, visible.Load)
644 visible.Store(true)
645 h.rc.invalidate(renderInvalidation{class: invalidateUrgent, producer: "activation"})
646 for _, timer := range h.armedTimers(t) {
647 timer.ch <- time.Time{}
648 }
649 requireNoWake(t, h.wakes)
650
651 h.rc.noteSyncEnd(p, 1)
652 w := awaitWake(t, h.wakes)
653 require.True(t, w.urgent)
654 requireNoWake(t, h.wakes)
655 })
656
657 t.Run("hidden batch does not stall newly active output or wake on late end", func(t *testing.T) {
658 h := newCoordinatorHarness(t)
659 var visible atomic.Bool
660 visible.Store(true)
661 p := &pane{}
662 h.rc.noteSyncBeginWithRenderability(p, 1, visible.Load)
663 h.rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "batch"})
664 visible.Store(false)
665 h.rc.invalidate(renderInvalidation{class: invalidateUrgent, producer: "new-active-tab"})
666 timers := h.armedTimers(t)
667 require.NotEmpty(t, timers)
668 timers[len(timers)-1].ch <- time.Time{}
669 w := awaitWake(t, h.wakes)
670 require.False(t, w.watchdog)
671 h.rc.noteSyncEnd(p, 1)
672 requireNoWake(t, h.wakes)
673 })
674}
675
676func TestRenderCoordinatorSyncGateRetriesWhenBatchPublishesDuringPredicate(t *testing.T) {
677 h := newCoordinatorHarness(t)
678 first, second := &pane{}, &pane{}
679 gateEvaluated := make(chan struct{})
680 releaseGate := make(chan struct{})
681 var hookOnce sync.Once
682 h.rc.opts.afterSyncGateEvaluated = func() {
683 hookOnce.Do(func() {
684 close(gateEvaluated)
685 <-releaseGate
686 })
687 }
688 h.rc.noteSyncBeginWithRenderability(first, 1, func() bool { return false })
689 h.rc.invalidate(renderInvalidation{class: invalidateUrgent, producer: "test"})
690
691 fireDone := make(chan struct{})
692 go func() {
693 h.rc.fireCurrent(false)
694 close(fireDone)
695 }()
696 <-gateEvaluated
697 // This publication is deliberately between an open predicate result and
698 // pending consumption. A stale gate result must be rejected and retried.
699 h.rc.noteSyncBeginWithRenderability(second, 2, func() bool { return true })
700 close(releaseGate)
701 awaitTestCompletion(t, fireDone, "fire did not finish after retrying the registry snapshot")
702
703 requireNoWake(t, h.wakes)
704 h.rc.noteSyncEnd(second, 2)
705 require.Equal(t, renderWake{urgent: true, coalesced: 1}, awaitWake(t, h.wakes))
706}
707
708func TestRenderCoordinatorSynchronizedOutput(t *testing.T) {
709 t.Run("completion flushes pending state in one wake", func(t *testing.T) {
710 h := newCoordinatorHarness(t)
711 h.syncActive.Store(true)
712 h.rc.noteSyncBegin(nil, 1)
713 watchdogs := h.armedTimers(t)
714 require.NotEmpty(t, watchdogs, "sync begin must arm the completion watchdog")
715 require.Equal(t, maxSyncUpdateDuration, watchdogs[len(watchdogs)-1].duration)
716
717 h.rc.invalidate(renderInvalidation{class: invalidateOutput, reset: true})
718 timers := h.armedTimers(t)
719 require.Len(t, timers, 1)
720 h.rc.mu.Lock()
721 normalWorkerDone := h.rc.normalLane.token.done
722 h.rc.mu.Unlock()
723 require.NotNil(t, normalWorkerDone)
724 timers[0].ch <- time.Time{}
725 requireWorkerExit(t, normalWorkerDone)
726 requireNoWake(t, h.wakes)
727
728 h.syncActive.Store(false)
729 h.rc.noteSyncEnd(nil, 1)
730 w := awaitWake(t, h.wakes)
731 require.True(t, w.reset)
732 requireNoWake(t, h.wakes)
733 })
734
735 t.Run("watchdog ends a wedged batch before it flushes", func(t *testing.T) {
736 h := newCoordinatorHarness(t)
737 h.syncActive.Store(true)
738 forced := make(chan struct{}, 1)
739 h.rc.noteSyncBegin(nil, 7, func() { forced <- struct{}{} })
740 watchdogs := h.armedTimers(t)
741 require.NotEmpty(t, watchdogs, "sync begin must arm the completion watchdog")
742 h.rc.invalidate(renderInvalidation{class: invalidateOutput})
743
744 watchdogs[0].ch <- time.Time{}
745 <-forced
746 w := awaitWake(t, h.wakes)
747 require.True(t, w.watchdog, "a wedged synchronized batch must be force-flushed by the watchdog")
748 requireNoWake(t, h.wakes)
749 })
750
751 t.Run("watchdog retains gate while force closes the VT batch", func(t *testing.T) {
752 h := newCoordinatorHarness(t)
753 entered := make(chan struct{})
754 release := make(chan struct{})
755 h.rc.noteSyncBegin(nil, 1, func() {
756 close(entered)
757 <-release
758 })
759 watchdogs := h.armedTimers(t)
760 require.NotEmpty(t, watchdogs)
761 h.rc.invalidate(renderInvalidation{class: invalidateOutput})
762 watchdogs[0].ch <- time.Time{}
763 <-entered
764
765 // A concurrent deadline/fire sees the retained batch gate and cannot
766 // publish the partial VT state while force is blocked.
767 h.rc.fireCurrent(false)
768 requireNoWake(t, h.wakes)
769 close(release)
770 w := awaitWake(t, h.wakes)
771 require.True(t, w.watchdog)
772 require.True(t, w.urgent)
773 requireNoWake(t, h.wakes)
774 })
775
776 t.Run("watchdog flushes after the snapshotted generation is replaced", func(t *testing.T) {
777 h := newCoordinatorHarness(t)
778 h.rc.invalidate(renderInvalidation{class: invalidateOutput})
779 h.rc.mu.Lock()
780 staleGeneration := h.rc.normalLane.generation
781 h.rc.mu.Unlock()
782
783 // Simulate a new urgent publication after fireCurrent(true) snapshots its
784 // generation but before fire validates it.
785 h.rc.invalidate(renderInvalidation{class: invalidateUrgent})
786 h.rc.fire(staleGeneration, true, true)
787
788 w := awaitWake(t, h.wakes)
789 require.True(t, w.watchdog)
790 require.True(t, w.urgent)
791 require.Equal(t, 2, w.coalesced)
792 requireNoWake(t, h.wakes)
793 })
794
795 t.Run("stale watchdog generation cannot wake a completed batch", func(t *testing.T) {
796 h := newCoordinatorHarness(t)
797 h.syncActive.Store(true)
798 h.rc.noteSyncBegin(nil, 1)
799 watchdogs := h.armedTimers(t)
800 require.NotEmpty(t, watchdogs, "sync begin must arm the completion watchdog")
801 h.syncActive.Store(false)
802 h.rc.noteSyncEnd(nil, 1)
803
804 watchdogs[0].ch <- time.Time{}
805 requireNoWake(t, h.wakes)
806 })
807}
808
809// --- preview subscription --------------------------------------------------------
810
811func TestRenderCoordinatorPreviewSubscription(t *testing.T) {
812 h := newCoordinatorHarness(t)
813 viewer := &attachedClient{}
814 h.rc.subscribePreviewFor(viewer, 1, func(w renderWake) { h.previews <- w })
815
816 h.rc.invalidate(renderInvalidation{class: invalidateUrgent})
817 timers := h.armedTimers(t)
818 require.NotEmpty(t, timers, "invalidate must arm a deadline timer")
819 timers[len(timers)-1].ch <- time.Time{}
820 owner := awaitWake(t, h.wakes)
821 preview := awaitWake(t, h.previews)
822 require.Equal(t, owner, preview, "a subscribed preview observes the same coalesced wake")
823 requireNoWake(t, h.previews)
824
825 h.rc.teardownPreviewFor(viewer, 1)
826 h.rc.invalidate(renderInvalidation{class: invalidateUrgent})
827 timers = h.armedTimers(t)
828 require.NotEmpty(t, timers, "invalidate must arm a deadline timer")
829 timers[len(timers)-1].ch <- time.Time{}
830 awaitWake(t, h.wakes)
831 requireNoWake(t, h.previews)
832}
833
834// --- lifecycle and stale callbacks -----------------------------------------------
835
836func TestRenderCoordinatorPreviewWakesDoNotWaitForTargetAck(t *testing.T) {
837 cases := []struct {
838 name string
839 attach bool
840 }{
841 {name: "attached cross-session target blocked on ack", attach: true},
842 {name: "headless cross-session target blocked on ack"},
843 }
844 for _, tc := range cases {
845 t.Run(tc.name, func(t *testing.T) {
846 h := newCoordinatorHarness(t)
847 h.ackReady.Store(false)
848 if tc.attach {
849 h.rc.attach(&attachedClient{})
850 }
851 viewer := &attachedClient{}
852 h.rc.subscribePreviewFor(viewer, 1, func(w renderWake) { h.previews <- w })
853
854 h.rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "target output"})
855 timers := h.armedTimers(t)
856 require.Len(t, timers, 1)
857 timers[0].ch <- time.Time{}
858
859 preview := awaitWake(t, h.previews)
860 require.Equal(t, renderWake{coalesced: 1}, preview,
861 "the viewer preview must receive target output without target ACK capacity")
862 requireNoWake(t, h.previews)
863 if tc.attach {
864 requireNoWake(t, h.wakes)
865 } else {
866 wake := awaitWake(t, h.wakes)
867 require.Equal(t, renderWake{coalesced: 1}, wake)
868 require.Nil(t, wake.lease, "shared wakes do not carry an attachment lease")
869 }
870
871 h.ackReady.Store(true)
872 h.rc.notifyAck()
873 if tc.attach {
874 wake := awaitWake(t, h.wakes)
875 require.Equal(t, renderWake{coalesced: 1}, wake,
876 "the target frame remains pending for its ACK")
877 }
878 requireNoWake(t, h.previews)
879 })
880 }
881}
882
883func TestRenderCoordinatorDetachedTargetPublishesPreviewOnlyInvalidations(t *testing.T) {
884 h := newCoordinatorHarness(t)
885 target, viewer := &attachedClient{}, &attachedClient{}
886 h.ackReady.Store(false)
887 h.rc.attach(target)
888 h.rc.subscribePreviewFor(viewer, 1, func(w renderWake) { h.previews <- w })
889 h.rc.noteDetach(target)
890
891 // A detached target only accepts PTY/session-owned invalidations. The old
892 // attachment remains stale and must not revive its render path.
893 h.rc.invalidateForAttachment(target, renderInvalidation{class: invalidateOutput, producer: "stale attachment"})
894 require.Empty(t, h.armedTimers(t))
895
896 for _, producer := range []string{"first PTY output", "second PTY output"} {
897 h.rc.invalidate(renderInvalidation{class: invalidateOutput, producer: producer})
898 timers := h.armedTimers(t)
899 require.Len(t, timers, 1)
900 timers[0].ch <- time.Time{}
901 preview := awaitWake(t, h.previews)
902 require.Equal(t, renderWake{coalesced: 1}, preview)
903 wake := awaitWake(t, h.wakes)
904 require.Equal(t, renderWake{coalesced: 1}, wake)
905 require.Nil(t, wake.lease, "shared wakes do not carry an attachment lease")
906
907 h.rc.mu.Lock()
908 require.False(t, h.rc.pending)
909 require.False(t, h.rc.pendingPreview)
910 require.False(t, h.rc.ackDeferred)
911 require.False(t, h.rc.deadlineDue)
912 require.Zero(t, h.rc.coalesced)
913 require.Empty(t, h.rc.attachments)
914 h.rc.mu.Unlock()
915 }
916
917 // A later attach retains the normal first-paint reset path.
918 replacement := &attachedClient{}
919 h.ackReady.Store(true)
920 h.rc.attach(replacement)
921 h.rc.invalidateForAttachment(replacement, renderInvalidation{class: invalidateUrgent, reset: true, producer: "first paint"})
922 timers := h.armedTimers(t)
923 require.Len(t, timers, 1)
924 timers[0].ch <- time.Time{}
925 wake := awaitWake(t, h.wakes)
926 require.True(t, wake.reset)
927 require.Nil(t, wake.lease, "shared wakes do not carry an attachment lease")
928}
929
930func TestRenderCoordinatorPreviewLifecycleDropsStaleTargetWakes(t *testing.T) {
931 cases := []struct {
932 name string
933 transition func(*renderCoordinator, *attachedClient, *attachedClient)
934 }{
935 {"target detach", func(rc *renderCoordinator, target, _ *attachedClient) { rc.noteDetach(target) }},
936 {"target detach and attach", func(rc *renderCoordinator, target, replacement *attachedClient) {
937 rc.noteDetach(target)
938 rc.attach(replacement)
939 }},
940 {"target teardown", func(rc *renderCoordinator, _ *attachedClient, _ *attachedClient) { rc.beginSessionTeardown().finish() }},
941 }
942 for _, tc := range cases {
943 t.Run(tc.name, func(t *testing.T) {
944 h := newCoordinatorHarness(t)
945 target, replacement, viewer := &attachedClient{}, &attachedClient{}, &attachedClient{}
946 h.rc.attach(target)
947 h.rc.subscribePreviewFor(viewer, 1, func(w renderWake) { h.previews <- w })
948 h.rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "target output"})
949 stale := h.armedTimers(t)
950 require.Len(t, stale, 1)
951
952 tc.transition(h.rc, target, replacement)
953 stale[0].ch <- time.Time{}
954 requireNoWake(t, h.wakes)
955 requireNoWake(t, h.previews)
956 })
957 }
958}
959
960func TestRenderCoordinatorPreviewSubscriptionsAreIndependent(t *testing.T) {
961 h := newCoordinatorHarness(t)
962 one, two := &attachedClient{}, &attachedClient{}
963 first, second := make(chan renderWake, 1), make(chan renderWake, 1)
964 h.rc.subscribePreviewFor(one, 1, func(w renderWake) { first <- w })
965 h.rc.subscribePreviewFor(two, 1, func(w renderWake) { second <- w })
966 h.rc.attach(&attachedClient{})
967 h.rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "preview"})
968 last := h.armedTimers(t)
969 last[len(last)-1].ch <- time.Time{}
970 awaitWake(t, first)
971 awaitWake(t, second)
972
973 h.rc.teardownPreviewFor(one, 1)
974 h.rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "preview"})
975 last = h.armedTimers(t)
976 last[len(last)-1].ch <- time.Time{}
977 requireNoWake(t, first)
978 awaitWake(t, second)
979}
980
981func TestRenderCoordinatorLifecycleDropsStaleWakes(t *testing.T) {
982 cases := []struct {
983 name string
984 teardown func(rc *renderCoordinator, owner *attachedClient)
985 }{
986 {"detach", func(rc *renderCoordinator, owner *attachedClient) { rc.noteDetach(owner) }},
987 {"session teardown", func(rc *renderCoordinator, _ *attachedClient) { rc.beginSessionTeardown().finish() }},
988 }
989 for _, tc := range cases {
990 t.Run(tc.name, func(t *testing.T) {
991 h := newCoordinatorHarness(t)
992 owner := &attachedClient{}
993 h.rc.attach(owner)
994 h.rc.invalidate(renderInvalidation{class: invalidateOutput})
995 stale := h.armedTimers(t)
996 require.NotEmpty(t, stale, "invalidate must arm a deadline timer")
997
998 tc.teardown(h.rc, owner)
999 for _, timer := range stale {
1000 timer.ch <- time.Time{}
1001 }
1002 requireNoWake(t, h.wakes)
1003
1004 h.rc.invalidate(renderInvalidation{class: invalidateUrgent})
1005 for _, timer := range h.armedTimers(t) {
1006 timer.ch <- time.Time{}
1007 }
1008 requireNoWake(t, h.wakes)
1009 })
1010 }
1011
1012 t.Run("replacement attachment wakes independently", func(t *testing.T) {
1013 h := newCoordinatorHarness(t)
1014 owner := &attachedClient{}
1015 replacement := &attachedClient{}
1016 h.rc.attach(owner)
1017 h.rc.invalidate(renderInvalidation{class: invalidateOutput})
1018 stale := h.armedTimers(t)
1019 require.NotEmpty(t, stale, "invalidate must arm a deadline timer")
1020
1021 h.rc.noteDetach(owner)
1022 h.rc.attach(replacement)
1023 for _, timer := range stale {
1024 timer.ch <- time.Time{}
1025 }
1026 requireNoWake(t, h.wakes)
1027
1028 h.rc.invalidate(renderInvalidation{class: invalidateUrgent, reset: true})
1029 fresh := h.armedTimers(t)
1030 require.NotEmpty(t, fresh, "a replacement attachment must arm its own deadline")
1031 fresh[len(fresh)-1].ch <- time.Time{}
1032 w := awaitWake(t, h.wakes)
1033 require.True(t, w.reset, "the replacement starts from an independent full state")
1034 requireNoWake(t, h.wakes)
1035 })
1036}
1037
1038// --- resize metadata ownership -----------------------------------------------------
1039
1040func TestRenderCoordinatorResizeMetadata(t *testing.T) {
1041 sz := func(cols, rows int) domain.Size { return domain.Size{Cols: cols, Rows: rows} }
1042 cases := []struct {
1043 name string
1044 run func(t *testing.T, rc *renderCoordinator, owner, replacement *attachedClient)
1045 }{
1046 {
1047 name: "latest request wins with strictly increasing epochs",
1048 run: func(t *testing.T, rc *renderCoordinator, owner, _ *attachedClient) {
1049 require.Equal(t, uint64(1), rc.recordResizeRequest(sz(100, 30), owner))
1050 require.Equal(t, uint64(2), rc.recordResizeRequest(sz(110, 32), owner))
1051 require.Equal(t, uint64(3), rc.recordResizeRequest(sz(120, 40), owner))
1052 snap := rc.resizeSnapshot()
1053 require.Equal(t, sz(120, 40), snap.size)
1054 require.Same(t, owner, snap.source)
1055 require.Equal(t, uint64(3), snap.epoch)
1056
1057 // Resize state is exclusively coordinator-owned.
1058 require.Zero(t, snap.committed)
1059 },
1060 },
1061 {
1062 name: "duplicate sizes still advance the epoch deterministically",
1063 run: func(t *testing.T, rc *renderCoordinator, owner, _ *attachedClient) {
1064 require.Equal(t, uint64(1), rc.recordResizeRequest(sz(120, 40), owner))
1065 require.Equal(t, uint64(2), rc.recordResizeRequest(sz(120, 40), owner))
1066 snap := rc.resizeSnapshot()
1067 require.Equal(t, sz(120, 40), snap.size)
1068 require.Equal(t, uint64(2), snap.epoch)
1069 },
1070 },
1071 {
1072 name: "stale resize callbacks preserve latest metadata through lifecycle transitions",
1073 run: func(t *testing.T, rc *renderCoordinator, owner, replacement *attachedClient) {
1074 type lifecycleCase struct {
1075 name string
1076 installNew bool
1077 }
1078 for _, tc := range []lifecycleCase{
1079 {name: "detach"},
1080 {name: "park"},
1081 {name: "session teardown"},
1082 {name: "replacement", installNew: true},
1083 } {
1084 t.Run(tc.name, func(t *testing.T) {
1085 h := newCoordinatorHarness(t)
1086 staleOwner := &attachedClient{}
1087 freshOwner := &attachedClient{}
1088 h.rc.attach(staleOwner)
1089 require.Equal(t, uint64(1), h.rc.recordResizeRequest(sz(120, 40), staleOwner))
1090
1091 // This models a stale callback which captured the old
1092 // attachment before lifecycle ownership changed.
1093 staleCallback := func() uint64 {
1094 return h.rc.recordResizeRequest(sz(80, 20), staleOwner)
1095 }
1096
1097 switch tc.name {
1098 case "detach":
1099 h.rc.noteDetach(staleOwner)
1100 case "park":
1101 h.rc.noteDetach(staleOwner)
1102 case "session teardown":
1103 h.rc.beginSessionTeardown().finish()
1104 case "replacement":
1105 h.rc.noteDetach(staleOwner)
1106 h.rc.attach(freshOwner)
1107 require.Equal(t, uint64(2), h.rc.recordResizeRequest(sz(100, 50), freshOwner))
1108 }
1109
1110 require.Zero(t, staleCallback(), "stale callback must not advance the resize epoch")
1111 snap := h.rc.resizeSnapshot()
1112 if tc.installNew {
1113 require.Equal(t, sz(100, 50), snap.size)
1114 require.Equal(t, uint64(2), snap.epoch)
1115 require.Same(t, freshOwner, snap.source)
1116 return
1117 }
1118 require.Equal(t, sz(120, 40), snap.size)
1119 require.Equal(t, uint64(1), snap.epoch)
1120 require.Same(t, staleOwner, snap.source,
1121 "the stale callback must not replace the recorded attachment identity")
1122 })
1123 }
1124 },
1125 },
1126 }
1127 for _, tc := range cases {
1128 t.Run(tc.name, func(t *testing.T) {
1129 h := newCoordinatorHarness(t)
1130 owner := &attachedClient{}
1131 replacement := &attachedClient{}
1132 h.rc.attach(owner)
1133 tc.run(t, h.rc, owner, replacement)
1134 })
1135 }
1136}
1137
1138// --- producer fan-in ------------------------------------------------------------
1139
1140// producerFiles is the current production direct-paint inventory. Every file
1141// gets exactly one exercised state transition in TestProducerInvalidations.
1142var producerFiles = []string{
1143 "attention.go", "client.go", "copymode.go", "floating.go", "input.go",
1144 "palette.go", "pane_actions.go", "picker.go", "prompt.go", "render.go",
1145 "session.go",
1146}
1147
1148func TestProducerInvalidations(t *testing.T) {
1149 cases := []struct {
1150 file string
1151 name string
1152 tabs int
1153 producer string
1154 run func(t *testing.T, d *Daemon, sess *session, ac *attachedClient)
1155 }{
1156 {
1157 file: "attention.go",
1158 name: "attention repaint",
1159 run: func(t *testing.T, d *Daemon, sess *session, ac *attachedClient) {
1160 d.repaintAttachedClients(sess)
1161 },
1162 },
1163 {
1164 file: "client.go",
1165 name: "client theme application",
1166 run: func(t *testing.T, d *Daemon, sess *session, ac *attachedClient) {
1167 d.applyTheme(sess, ac, ports.Theme{TrueColor: true})
1168 },
1169 },
1170 {
1171 file: "copymode.go",
1172 name: "copy mode entry",
1173 run: func(t *testing.T, d *Daemon, sess *session, ac *attachedClient) {
1174 d.enterCopyMode(sess, ac)
1175 },
1176 },
1177 {
1178 file: "floating.go",
1179 name: "floating toggle to visible",
1180 run: func(t *testing.T, d *Daemon, sess *session, ac *attachedClient) {
1181 fp := newPaneWithStableID(layout.PaneID("floating"), "float-producer", newQuietPTY(), domain.Size{Cols: 40, Rows: 10})
1182 tb := testAttachmentTab(sess)
1183 tb.mu.Lock()
1184 tb.floating.pane = fp
1185 tb.floating.state = floatingHidden
1186 tb.floating.generation = 1
1187 tb.mu.Unlock()
1188 require.NoError(t, d.toggleFloating(sess, ac))
1189 },
1190 },
1191 {
1192 file: "input.go",
1193 name: "jump attention switches a local tab",
1194 tabs: 2,
1195 producer: "attention.go",
1196 run: func(t *testing.T, d *Daemon, sess *session, ac *attachedClient) {
1197 sess.tabs[1].attention = true
1198 daemonKeyHandler{d: d, ac: ac}.Action(keys.ActionJumpAttention, nil)
1199 require.Equal(t, 1, testAttachmentTabIndex(sess))
1200 },
1201 },
1202 {
1203 file: "palette.go",
1204 name: "palette entry",
1205 run: func(t *testing.T, d *Daemon, sess *session, ac *attachedClient) {
1206 d.enterPalette(sess, ac)
1207 },
1208 },
1209 {
1210 file: "pane_actions.go",
1211 name: "pane split",
1212 run: func(t *testing.T, d *Daemon, sess *session, ac *attachedClient) {
1213 require.NoError(t, d.splitPane(sess, ac, layout.Right))
1214 },
1215 },
1216 {
1217 file: "picker.go",
1218 name: "picker entry",
1219 run: func(t *testing.T, d *Daemon, sess *session, ac *attachedClient) {
1220 d.enterPicker(sess, ac)
1221 },
1222 },
1223 {
1224 file: "prompt.go",
1225 name: "prompt entry",
1226 run: func(t *testing.T, d *Daemon, sess *session, ac *attachedClient) {
1227 d.enterPrompt(sess, ac, "rename", "", func(string) error { return nil })
1228 },
1229 },
1230 {
1231 file: "render.go",
1232 name: "retained resize dispatch",
1233 run: func(t *testing.T, d *Daemon, sess *session, ac *attachedClient) {
1234 clk := newCoordinatorMockClock(t, 2)
1235 // The coordinator owns deadline time; install the deterministic
1236 // clock on the already-attached coordinator rather than changing
1237 // Daemon's construction-time clock afterwards.
1238 sess.renderCoordinator().opts.clock = clk.clock
1239 d.resize(sess, ac, domain.Size{Cols: 100, Rows: 26})
1240 timer := awaitCoordinatorScheduledTimer(t, clk)
1241 done := captureResizeCallbackDone(t, sess.renderCoordinator())
1242 timer.ch <- time.Time{}
1243 awaitTestCompletion(t, done, "resize callback did not complete")
1244 },
1245 },
1246 {
1247 file: "session.go",
1248 name: "tab close repaint",
1249 tabs: 2,
1250 run: func(t *testing.T, d *Daemon, sess *session, ac *attachedClient) {
1251 sess.mu.Lock()
1252 tb := sess.tabs[1]
1253 sess.mu.Unlock()
1254 require.NoError(t, d.closeTab(sess, tb, true))
1255 },
1256 },
1257 }
1258
1259 exercised := make([]string, 0, len(cases))
1260 for _, tc := range cases {
1261 exercised = append(exercised, tc.file)
1262 }
1263 require.ElementsMatch(t, producerFiles, exercised,
1264 "every current direct-paint producer file needs exactly one exercised transition")
1265
1266 for _, tc := range cases {
1267 t.Run(tc.file+"/"+tc.name, func(t *testing.T) {
1268 tabs := tc.tabs
1269 if tabs == 0 {
1270 tabs = 1
1271 }
1272 d, sess, ac, sends, releases := newManualTabSession(t, tabs)
1273 defer releaseAll(releases)
1274 invs := make(chan renderInvalidation, 8)
1275 sess.installRenderCoordinator(newRenderCoordinator(renderCoordinatorOptions{
1276 clock: d.clock,
1277 wake: func(renderWake) {},
1278 onInvalidate: func(inv renderInvalidation) { invs <- inv },
1279 }))
1280
1281 tc.run(t, d, sess, ac)
1282
1283 inv := awaitInvalidation(t, invs)
1284 if tc.producer != "" {
1285 require.Equal(t, tc.producer, inv.producer)
1286 }
1287 requireNoInvalidation(t, invs)
1288 requireNoCoordinatorOutputFrame(t, sends)
1289 })
1290 }
1291}
1292
1293// TestProducerInvalidationInventory is a supplementary inventory guard only:
1294// it keeps producerFiles aligned with the daemon source layout so the
1295// behavioral table above cannot drift silently. It is never acceptance
1296// evidence for coordinator behavior.
1297func TestProducerInvalidationInventory(t *testing.T) {
1298 for _, name := range producerFiles {
1299 _, err := os.Stat(filepath.Join(".", name))
1300 require.NoErrorf(t, err, "producer file %s is gone; update TestProducerInvalidations", name)
1301 }
1302}
1303
1304// --- coordinator resize dispatch ----------------------------------------------
1305
1306func TestConcurrentPaintInitializesOverlayUnderSendOwnership(t *testing.T) {
1307 p, release := newBlockingPTY(t)
1308 defer release()
1309 d, sess, ac, _ := newManualSessionWithPTYs(t, p)
1310 // Exercise the original seam: two fallback paints reach lazy initialization
1311 // together before either can compose.
1312 ac.overlays = nil
1313 ac.overlayOnce = sync.Once{}
1314 start := make(chan struct{})
1315 var wg sync.WaitGroup
1316 wg.Add(2)
1317 for range 2 {
1318 go func() {
1319 defer wg.Done()
1320 <-start
1321 d.paint(sess, ac, true, nil)
1322 }()
1323 }
1324 close(start)
1325 wg.Wait()
1326 require.NotNil(t, ac.overlays)
1327}
1328
1329func TestStartPaneGoroutinesAccountsForOneReader(t *testing.T) {
1330 p, release := newBlockingPTY(t)
1331 d, sess, _, _ := newManualSessionWithPTYs(t, p)
1332 tb := testAttachmentTab(sess)
1333 require.NotNil(t, tb)
1334 d.startPaneGoroutines(sess, tb, tb.focusedPane())
1335 release()
1336 select {
1337 case <-waitGroupDone(&d.sessWg):
1338 case <-time.After(time.Second):
1339 t.Fatal("one reader must balance exactly one WaitGroup count")
1340 }
1341}
1342
1343func requireWorkerExit(t *testing.T, done <-chan struct{}) {
1344 t.Helper()
1345 awaitTestCompletion(t, done, "cancelled coordinator timer worker did not exit")
1346}
1347
1348func TestRenderCoordinatorResizeLaneRejectsStaleToken(t *testing.T) {
1349 rc := newRenderCoordinator(renderCoordinatorOptions{})
1350 newer := &timerToken{generation: 2, timer: &portsmocks.MockTimer{}, cancel: make(chan struct{}), done: make(chan struct{})}
1351 stale := &timerToken{generation: 1}
1352
1353 rc.mu.Lock()
1354 rc.resizeLane.generation = 2
1355 rc.resizeLane.token = newer
1356 require.False(t, rc.resizeLane.clearLocked(stale))
1357 require.Same(t, newer, rc.resizeLane.token)
1358 require.True(t, rc.resizeLane.clearLocked(newer))
1359 require.Nil(t, rc.resizeLane.token)
1360 rc.mu.Unlock()
1361}
1362
1363func TestRenderCoordinatorNilClockRetryRunsSynchronously(t *testing.T) {
1364 rc := newRenderCoordinator(renderCoordinatorOptions{})
1365 owner := &attachedClient{}
1366 rc.attach(owner)
1367 lease := rc.attachmentLease(owner)
1368 epoch := rc.recordResizeRequestForLease(domain.Size{Cols: 120, Rows: 40}, owner, lease)
1369 require.NotZero(t, epoch)
1370 require.True(t, rc.resizeCurrentForLease(epoch, owner, lease, true))
1371
1372 called := false
1373 rc.scheduleResizeRetryForLease(epoch, owner, lease, func() { called = true })
1374 require.True(t, called, "a nil clock must run the retry without a timer")
1375 rc.mu.Lock()
1376 retryToken := rc.retryLane.token
1377 rc.mu.Unlock()
1378 require.Nil(t, retryToken, "a disabled retry lane must not retain stale completion ownership")
1379}
1380
1381func TestRenderCoordinatorInertTimerFiresSynchronouslyWithoutWorker(t *testing.T) {
1382 clock := portsmocks.NewMockClock(t)
1383 timer := portsmocks.NewMockTimer(t)
1384 clock.EXPECT().NewTimer(minOutputRenderDeadline).Return(timer).Once()
1385 timer.EXPECT().C().Return((<-chan time.Time)(nil)).Once()
1386 timer.EXPECT().Stop().Return(true).Once()
1387
1388 wakes := make(chan renderWake, 1)
1389 rc := newRenderCoordinator(renderCoordinatorOptions{
1390 clock: clock,
1391 ackReady: func() bool { return true },
1392 wake: func(w renderWake) { wakes <- w },
1393 })
1394 rc.invalidate(renderInvalidation{class: invalidateOutput})
1395
1396 require.Equal(t, renderWake{coalesced: 1}, <-wakes)
1397 require.Equal(t, renderCoordinatorBurstMetricsSnapshot{invalidations: 1, wakes: 1, coalesced: 1}, rc.burstMetricsSnapshot())
1398 rc.mu.Lock()
1399 require.Nil(t, rc.normalLane.token)
1400 rc.mu.Unlock()
1401}
1402
1403func TestRenderCoordinatorSyncBatchSurvivesAttachmentLifecycle(t *testing.T) {
1404 t.Run("detach retains a gated headless preview until complete", func(t *testing.T) {
1405 for _, transition := range []struct {
1406 name string
1407 run func(*renderCoordinator, *attachedClient)
1408 }{
1409 {"detach", func(rc *renderCoordinator, ac *attachedClient) { rc.noteDetach(ac) }},
1410 } {
1411 t.Run(transition.name, func(t *testing.T) {
1412 h := newCoordinatorHarness(t)
1413 target, viewer, p := &attachedClient{}, &attachedClient{}, &pane{}
1414 h.rc.attach(target)
1415 h.rc.subscribePreviewFor(viewer, 1, func(w renderWake) { h.previews <- w })
1416 h.rc.noteSyncBegin(p, 1)
1417 watchdog := h.armedTimers(t)[0]
1418 transition.run(h.rc, target)
1419
1420 // PTY-owned output remains visible to picker observers while the
1421 // target attachment is parked. Its deadline must retain the sync gate.
1422 h.rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "complete preview"})
1423 timers := h.armedTimers(t)
1424 require.Len(t, timers, 1)
1425 h.rc.mu.Lock()
1426 normalWorkerDone := h.rc.normalLane.token.done
1427 h.rc.mu.Unlock()
1428 require.NotNil(t, normalWorkerDone)
1429 timers[0].ch <- time.Time{}
1430 requireWorkerExit(t, normalWorkerDone)
1431 requireNoWake(t, h.previews)
1432
1433 h.rc.mu.Lock()
1434 require.Same(t, h.rc.syncBatches[p].lane.token.timer, watchdog.mock)
1435 h.rc.mu.Unlock()
1436 h.rc.noteSyncEnd(p, 1)
1437 preview := awaitWake(t, h.previews)
1438 require.Equal(t, renderWake{urgent: true, coalesced: 1}, preview)
1439 requireNoWake(t, h.previews)
1440 wake := awaitWake(t, h.wakes)
1441 require.Equal(t, renderWake{urgent: true, coalesced: 1}, wake)
1442 require.Nil(t, wake.lease, "shared wakes do not carry an attachment lease")
1443 })
1444 }
1445 })
1446
1447 t.Run("replacement first paint waits for active batch and rejects stale owner", func(t *testing.T) {
1448 h := newCoordinatorHarness(t)
1449 old, replacement, p := &attachedClient{}, &attachedClient{}, &pane{}
1450 h.rc.attach(old)
1451 h.rc.noteSyncBegin(p, 1)
1452 watchdog := h.armedTimers(t)[0]
1453 h.rc.noteDetach(old)
1454 h.rc.attachWithReadiness(replacement, false)
1455 require.False(t, h.rc.markAttachmentReady(h.rc.attachmentLease(old)), "old attachment Welcome is stale")
1456 require.True(t, h.rc.markAttachmentReady(h.rc.attachmentLease(replacement)))
1457
1458 h.rc.invalidateForAttachment(old, renderInvalidation{class: invalidateUrgent, producer: "stale old attachment"})
1459 require.Empty(t, h.armedTimers(t))
1460 h.rc.invalidateForAttachment(replacement, renderInvalidation{class: invalidateUrgent, reset: true, producer: "replacement first paint"})
1461 timers := h.armedTimers(t)
1462 require.Len(t, timers, 1)
1463 h.rc.mu.Lock()
1464 normalWorkerDone := h.rc.normalLane.token.done
1465 h.rc.mu.Unlock()
1466 require.NotNil(t, normalWorkerDone)
1467 timers[0].ch <- time.Time{}
1468 requireWorkerExit(t, normalWorkerDone)
1469 requireNoWake(t, h.wakes)
1470
1471 h.rc.mu.Lock()
1472 require.Same(t, h.rc.syncBatches[p].lane.token.timer, watchdog.mock)
1473 h.rc.mu.Unlock()
1474 h.rc.noteSyncEnd(p, 1)
1475 wake := awaitWake(t, h.wakes)
1476 require.True(t, wake.reset)
1477 require.True(t, wake.urgent)
1478 require.Nil(t, wake.lease, "shared wakes do not carry an attachment lease")
1479 requireNoWake(t, h.wakes)
1480 })
1481
1482 t.Run("pane removal and session teardown cancel sync workers", func(t *testing.T) {
1483 for _, lifecycle := range []struct {
1484 name string
1485 end func(*renderCoordinator, *pane)
1486 }{
1487 {"pane removal", func(rc *renderCoordinator, p *pane) { rc.noteSyncPaneRemoved(p) }},
1488 {"session teardown", func(rc *renderCoordinator, _ *pane) { rc.beginSessionTeardown().finish() }},
1489 } {
1490 t.Run(lifecycle.name, func(t *testing.T) {
1491 clk := newCoordinatorMockClock(t, 2)
1492 rc := newRenderCoordinator(renderCoordinatorOptions{clock: clk.clock})
1493 p := &pane{}
1494 rc.noteSyncBegin(p, 1)
1495 syncTimer := <-clk.timers
1496 rc.mu.Lock()
1497 syncDone := rc.syncBatches[p].lane.token.done
1498 rc.mu.Unlock()
1499 lifecycle.end(rc, p)
1500 syncTimer.mock.AssertNumberOfCalls(t, "Stop", 1)
1501 requireWorkerExit(t, syncDone)
1502 if lifecycle.name == "session teardown" {
1503 rc.waitForTimerWorkers()
1504 }
1505 })
1506 }
1507 })
1508}
1509
1510func TestRenderCoordinatorResizeEpochDispatch(t *testing.T) {
1511 newResizeFixture := func(t *testing.T) (*Daemon, *session, *attachedClient, chan ports.Frame, *coordinatorMockClock, chan renderInvalidation) {
1512 t.Helper()
1513 p, releasePTY := newBlockingPTY(t)
1514 t.Cleanup(releasePTY)
1515 d, sess, ac, sends := newManualSessionWithPTYs(t, p)
1516 clk := newCoordinatorMockClock(t, 4)
1517 d.clock = clk.clock
1518 invs := make(chan renderInvalidation, 4)
1519 rc := newRenderCoordinator(renderCoordinatorOptions{
1520 clock: clk.clock,
1521 wake: func(renderWake) {},
1522 onInvalidate: func(inv renderInvalidation) { invs <- inv },
1523 })
1524 rc.attach(ac)
1525 sess.installRenderCoordinator(rc)
1526 return d, sess, ac, sends, clk, invs
1527 }
1528
1529 t.Run("bounded timer records metadata and dispatches through the coordinator", func(t *testing.T) {
1530 d, sess, ac, sends, clk, invs := newResizeFixture(t)
1531
1532 d.resize(sess, ac, domain.Size{Cols: 120, Rows: 24})
1533 timer := awaitCoordinatorScheduledTimer(t, clk)
1534 require.GreaterOrEqual(t, timer.duration, minOutputRenderDeadline)
1535 require.LessOrEqual(t, timer.duration, maxOutputRenderDeadline)
1536 snap := sess.renderCoordinator().resizeSnapshot()
1537 require.Equal(t, domain.Size{Cols: 120, Rows: 24}, snap.size,
1538 "the coordinator must record the latest requested resize before delegating")
1539 require.Same(t, ac, snap.source)
1540 require.Equal(t, uint64(1), snap.epoch)
1541 requireNoCoordinatorOutputFrame(t, sends)
1542
1543 done := captureResizeCallbackDone(t, sess.renderCoordinator())
1544 timer.ch <- time.Time{}
1545 awaitTestCompletion(t, done, "resize callback did not complete")
1546 inv := awaitInvalidation(t, invs)
1547 require.True(t, inv.reset, "the resize dispatch must request a full-redraw invalidation")
1548 requireNoInvalidation(t, invs)
1549 requireNoCoordinatorOutputFrame(t, sends)
1550 })
1551
1552 t.Run("stale generations stay rejected", func(t *testing.T) {
1553 d, sess, ac, sends, clk, invs := newResizeFixture(t)
1554
1555 d.resize(sess, ac, domain.Size{Cols: 100, Rows: 24})
1556 first := awaitCoordinatorScheduledTimer(t, clk)
1557 d.resize(sess, ac, domain.Size{Cols: 120, Rows: 24})
1558 latest := awaitCoordinatorScheduledTimer(t, clk)
1559 require.Equal(t, uint64(2), sess.renderCoordinator().resizeSnapshot().epoch,
1560 "every resize request must advance the coordinator epoch")
1561
1562 done := captureResizeCallbackDone(t, sess.renderCoordinator())
1563 first.ch <- time.Time{}
1564 requireNoInvalidation(t, invs)
1565 requireNoCoordinatorOutputFrame(t, sends)
1566
1567 latest.ch <- time.Time{}
1568 awaitTestCompletion(t, done, "latest resize callback did not complete")
1569 inv := awaitInvalidation(t, invs)
1570 require.True(t, inv.reset)
1571 requireNoInvalidation(t, invs)
1572 requireNoCoordinatorOutputFrame(t, sends)
1573 require.Equal(t, domain.Size{Cols: 120, Rows: 24}, sess.renderCoordinator().resizeSnapshot().size)
1574 })
1575
1576 t.Run("cancellation still drops the pending dispatch", func(t *testing.T) {
1577 d, sess, ac, sends, clk, invs := newResizeFixture(t)
1578
1579 d.resize(sess, ac, domain.Size{Cols: 90, Rows: 30})
1580 timer := awaitCoordinatorScheduledTimer(t, clk)
1581 done := captureResizeCallbackDone(t, sess.renderCoordinator())
1582 sess.renderCoordinator().noteDetach(ac)
1583
1584 timer.ch <- time.Time{}
1585 awaitTestCompletion(t, done, "cancelled resize callback did not complete")
1586 requireNoInvalidation(t, invs)
1587 requireNoCoordinatorOutputFrame(t, sends)
1588 })
1589
1590 t.Run("request reports immediate completion and async schedule acceptance", func(t *testing.T) {
1591 t.Run("immediate reset invalidation", func(t *testing.T) {
1592 d, sess, ac, _, _, invs := newResizeFixture(t)
1593
1594 require.True(t, d.requestTransactionalResize(sess, ac, domain.Size{Cols: 120, Rows: 24}, true))
1595 inv := awaitInvalidation(t, invs)
1596 require.True(t, inv.reset)
1597 })
1598
1599 t.Run("superseded immediate transaction", func(t *testing.T) {
1600 resizeStarted := make(chan struct{})
1601 releaseResize := make(chan struct{})
1602 pty := &transactionalResizePTY{onResize: func() {
1603 close(resizeStarted)
1604 <-releaseResize
1605 }}
1606 d, sess, ac, _ := newManualSessionWithPTYs(t, pty)
1607 clk := newCoordinatorMockClock(t, 2)
1608 d.clock = clk.clock
1609 invs := make(chan renderInvalidation, 1)
1610 rc := newRenderCoordinator(renderCoordinatorOptions{
1611 clock: clk.clock,
1612 wake: func(renderWake) {},
1613 onInvalidate: func(inv renderInvalidation) { invs <- inv },
1614 })
1615 rc.attach(ac)
1616 sess.installRenderCoordinator(rc)
1617
1618 result := make(chan bool, 1)
1619 go func() {
1620 result <- d.requestTransactionalResize(sess, ac, domain.Size{Cols: 120, Rows: 24}, true)
1621 }()
1622 <-resizeStarted
1623 require.NotZero(t, rc.recordResizeRequest(domain.Size{Cols: 140, Rows: 30}, ac))
1624 close(releaseResize)
1625
1626 require.False(t, <-result)
1627 requireNoInvalidation(t, invs)
1628 })
1629
1630 t.Run("async schedule and teardown rejection", func(t *testing.T) {
1631 d, sess, ac, _, clk, _ := newResizeFixture(t)
1632
1633 require.True(t, d.requestTransactionalResize(sess, ac, domain.Size{Cols: 120, Rows: 24}, false))
1634 awaitCoordinatorScheduledTimer(t, clk)
1635
1636 // A torn-down coordinator cannot accept a stale attachment's schedule.
1637 sess.renderCoordinator().beginSessionTeardown().finish()
1638 require.False(t, d.requestTransactionalResize(sess, ac, domain.Size{Cols: 140, Rows: 30}, false))
1639 })
1640 })
1641}