1package client
2
3import (
4 "context"
5 "io"
6 "strings"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/require"
12
13 "github.com/bnema/vev/internal/domain"
14 "github.com/bnema/vev/internal/ports"
15 "github.com/bnema/vev/internal/protocol"
16)
17
18// Resize presentation-invalidation tests (Plan 001, uncommitted slice).
19//
20// The mechanism under test is deliberately split across two owners:
21//
22// - the attachment host's geometry collector records the latest valid
23// Terminal.Geometry and emits two independent coalesced wakeups: the
24// attachment sequence wakeup (geometryUpdate) that a granted foreground
25// consumes through Resize, and the presentation-invalidation wakeup
26// (presentationUpdate) that only the supervisor's serialized control path
27// consumes;
28// - the supervisor consumes presentationUpdate in every wait it owns
29// (attempt, ready, termination, backoff) and repaints only a picker
30// presentation (PickerPresentation), sampling Terminal.Geometry at render
31// time. The attachment settlement wait never consumes it: Begin has already
32// admitted a foreground that can write session output through the same
33// terminal before MarkAttached and the initial publication, so the
34// invalidation stays buffered for the next picker state instead of being
35// painted over it.
36//
37// Every test below is deterministic: a fake clock, scripted fakes, bounded
38// waits, and channel signals instead of sleeps. The resize channel is pushed
39// by the test itself, so no ordering is hoped for.
40
41// resizeTestTerminal is the supervisor-owned terminal double with a
42// test-driven ResizeEvents channel and a Geometry that reports the latest
43// pushed size. It embeds the attachment-supervisor terminal double so it also
44// satisfies the UI publication seam, and only overrides geometry.
45type resizeTestTerminal struct {
46 *attachTestTerminal
47 resizes chan domain.Geometry
48 geometryMu sync.Mutex
49 geometry domain.Geometry
50}
51
52func newResizeTestTerminal(in io.Reader, geometry domain.Geometry) *resizeTestTerminal {
53 return &resizeTestTerminal{
54 attachTestTerminal: &attachTestTerminal{in: in},
55 resizes: make(chan domain.Geometry, 8),
56 geometry: geometry,
57 }
58}
59
60func (t *resizeTestTerminal) ResizeEvents() <-chan domain.Geometry { return t.resizes }
61
62func (t *resizeTestTerminal) Geometry() (domain.Geometry, error) {
63 t.geometryMu.Lock()
64 defer t.geometryMu.Unlock()
65 return t.geometry, nil
66}
67
68// resize records the new size as the terminal's current geometry and then
69// publishes the event, exactly like a physical terminal whose size query and
70// resize event agree.
71func (t *resizeTestTerminal) resize(geometry domain.Geometry) {
72 t.geometryMu.Lock()
73 t.geometry = geometry
74 t.geometryMu.Unlock()
75 t.resizes <- geometry
76}
77
78func (t *resizeTestTerminal) closeResizeEvents() { close(t.resizes) }
79
80// setWritten replaces the terminal's committed output, so a test can seed the
81// bytes an admitted foreground would have written and then prove no picker
82// repaint overpainted them.
83func (t *resizeTestTerminal) setWritten(data string) {
84 t.attachTestTerminal.mu.Lock()
85 defer t.attachTestTerminal.mu.Unlock()
86 t.attachTestTerminal.out.Reset()
87 t.attachTestTerminal.out.WriteString(data)
88}
89
90// resizeRenderSample is one observed Render call: the exact state handed to the
91// renderer plus the terminal geometry the renderer would have sampled, exactly
92// as the offline composition does when it paints the picker.
93type resizeRenderSample struct {
94 state State
95 geometry domain.Geometry
96}
97
98// resizeRenderRecorder records every Render call both in a channel (for
99// ordered waits) and in a slice (for counting), so a test can prove that a
100// repaint happened, in what state, and against which geometry.
101type resizeRenderRecorder struct {
102 terminal *resizeTestTerminal
103 renders chan resizeRenderSample
104
105 mu sync.Mutex
106 all []resizeRenderSample
107}
108
109func newResizeRenderRecorder(terminal *resizeTestTerminal) *resizeRenderRecorder {
110 return &resizeRenderRecorder{terminal: terminal, renders: make(chan resizeRenderSample, 256)}
111}
112
113// resizeRenderPaint is the picker frame marker the recording renderer writes.
114// It stands for the production composition's picker bytes: a test can prove the
115// renderer ran and that its output never landed over session output.
116const resizeRenderPaint = "<picker-frame>"
117
118func (r *resizeRenderRecorder) render(state State) {
119 geometry, _ := r.terminal.Geometry()
120 sample := resizeRenderSample{state: state, geometry: geometry}
121 r.mu.Lock()
122 r.all = append(r.all, sample)
123 r.mu.Unlock()
124 // Exactly like the production offline composition, the recording renderer
125 // writes its frame only while the shared fence admits the state. A mutation
126 // that widened the fence would therefore land bytes a test can observe.
127 if PickerPresentation(state) {
128 _, _ = r.terminal.Out().Write([]byte(resizeRenderPaint))
129 }
130 select {
131 case r.renders <- sample:
132 default:
133 }
134}
135
136// drain discards every render already delivered to the ordered channel; the
137// counted slice is retained so a caller can still take a baseline count.
138func (r *resizeRenderRecorder) drain() {
139 for {
140 select {
141 case <-r.renders:
142 default:
143 return
144 }
145 }
146}
147
148func (r *resizeRenderRecorder) count() int {
149 r.mu.Lock()
150 defer r.mu.Unlock()
151 return len(r.all)
152}
153
154// last returns the most recent render sample, or the zero sample when nothing
155// was rendered.
156func (r *resizeRenderRecorder) last() resizeRenderSample {
157 r.mu.Lock()
158 defer r.mu.Unlock()
159 if len(r.all) == 0 {
160 return resizeRenderSample{}
161 }
162 return r.all[len(r.all)-1]
163}
164
165// samples returns every recorded render sample in call order.
166func (r *resizeRenderRecorder) samples() []resizeRenderSample {
167 r.mu.Lock()
168 defer r.mu.Unlock()
169 return append([]resizeRenderSample(nil), r.all...)
170}
171
172// awaitResizeRender waits for the next render matching match, bounded.
173func awaitResizeRender(t *testing.T, recorder *resizeRenderRecorder, match func(resizeRenderSample) bool) resizeRenderSample {
174 t.Helper()
175 timeout := time.After(5 * time.Second)
176 for {
177 select {
178 case sample := <-recorder.renders:
179 if match(sample) {
180 return sample
181 }
182 case <-timeout:
183 t.Fatal("the supervisor never repainted the presentation for the resize")
184 return resizeRenderSample{}
185 }
186 }
187}
188
189// pendingPresentationSignals reports how many coalesced presentation
190// invalidations are buffered. Reading len of a channel is synchronized, so it
191// is safe from the test goroutine while the collector runs.
192func (h *attachmentHost) pendingPresentationSignals() int { return len(h.presentationUpdate) }
193
194// pendingGeometrySignals reports how many attachment geometry wakeups are
195// buffered, so a test can prove the two signals are independent.
196func (h *attachmentHost) pendingGeometrySignals() int { return len(h.geometryUpdate) }
197
198// geometryProgress reports how many valid resizes the collector has recorded.
199func (h *attachmentHost) geometryProgress() uint64 {
200 h.geometryMu.Lock()
201 defer h.geometryMu.Unlock()
202 return h.geometrySeq
203}
204
205// resizeSupervisorHarness is one running supervisor with its fakes: a
206// test-driven resize channel, a recording renderer, and a fake clock.
207type resizeSupervisorHarness struct {
208 sup *Supervisor
209 service *supervisorTestService
210 clock *supervisorTestClock
211 terminal *resizeTestTerminal
212 reader *attachTestReader
213 connector *supervisorTestConnector
214 recorder *resizeRenderRecorder
215
216 cancel context.CancelFunc
217 runDone chan struct{}
218
219 runMu sync.Mutex
220 runErr error
221}
222
223type resizeHarnessOptions struct {
224 // picker is the supervisor's picker seam; nil runs the picker-less driver.
225 picker pickerHost
226 // geometry is the terminal's initial reported geometry.
227 geometry domain.Geometry
228 // closeResizes hands the supervisor an already closed ResizeEvents channel.
229 closeResizes bool
230 // connect scripts the broker attempts; nil returns the harness service.
231 connect func(ctx context.Context, call int) (ports.BrokerNavigator, error)
232}
233
234// startResizeHarness builds and runs one supervisor over the resize-aware
235// fakes. It returns as soon as Run is started, so a test can park on any wait.
236func startResizeHarness(t *testing.T, options resizeHarnessOptions) *resizeSupervisorHarness {
237 t.Helper()
238 ctx, cancel := context.WithCancel(context.Background())
239 clock := newSupervisorTestClock()
240 reader := newAttachTestReader()
241 geometry := options.geometry
242 if geometry == (domain.Geometry{}) {
243 geometry = domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}}
244 }
245 terminal := newResizeTestTerminal(reader, geometry)
246 if options.closeResizes {
247 terminal.closeResizeEvents()
248 }
249 service := newSupervisorTestService(ports.BrokerConnectionID{1})
250 service.publish(3, 1)
251 connect := options.connect
252 if connect == nil {
253 connect = func(context.Context, int) (ports.BrokerNavigator, error) { return service, nil }
254 }
255 connector := newSupervisorTestConnector(connect)
256 recorder := newResizeRenderRecorder(terminal)
257 sup := mustSupervisor(t, SupervisorConfig{
258 Connector: connector,
259 Terminal: terminal,
260 Clock: clock,
261 Jitter: func() float64 { return 0 },
262 Picker: options.picker,
263 Render: recorder.render,
264 })
265
266 harness := &resizeSupervisorHarness{
267 sup: sup,
268 service: service,
269 clock: clock,
270 terminal: terminal,
271 reader: reader,
272 connector: connector,
273 recorder: recorder,
274 cancel: cancel,
275 runDone: make(chan struct{}),
276 }
277 go func() {
278 err := sup.Run(ctx)
279 harness.runMu.Lock()
280 harness.runErr = err
281 harness.runMu.Unlock()
282 close(harness.runDone)
283 }()
284 t.Cleanup(func() {
285 cancel()
286 select {
287 case <-harness.runDone:
288 case <-time.After(5 * time.Second):
289 t.Error("the supervisor did not stop")
290 }
291 // Release the single input lifetime's reader so the pump goroutine that
292 // outlives Run, exactly like a real stdin that never reaches EOF, ends.
293 reader.close()
294 })
295 return harness
296}
297
298func (h *resizeSupervisorHarness) waitRun(t *testing.T) error {
299 t.Helper()
300 select {
301 case <-h.runDone:
302 case <-time.After(5 * time.Second):
303 t.Fatal("the supervisor did not stop")
304 }
305 h.runMu.Lock()
306 defer h.runMu.Unlock()
307 return h.runErr
308}
309
310// resizeStreamRegistry admits one fresh scripted logical stream per open call
311// and lets a test reach the stream of a specific attachment attempt.
312type resizeStreamRegistry struct {
313 mu sync.Mutex
314 streams []*sessionTestStream
315}
316
317func (r *resizeStreamRegistry) open(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
318 stream := newSessionTestStream()
319 r.mu.Lock()
320 r.streams = append(r.streams, stream)
321 r.mu.Unlock()
322 return stream, nil
323}
324
325// stream waits until the index-th stream was admitted and returns it.
326func (r *resizeStreamRegistry) stream(t *testing.T, index int) *sessionTestStream {
327 t.Helper()
328 require.Eventually(t, func() bool {
329 r.mu.Lock()
330 defer r.mu.Unlock()
331 return len(r.streams) > index
332 }, 5*time.Second, time.Millisecond, "the supervisor never admitted stream %d", index)
333 r.mu.Lock()
334 defer r.mu.Unlock()
335 return r.streams[index]
336}
337
338// streamResizes returns the typed resize messages one attachment stream sent.
339func streamResizes(stream *sessionTestStream) []protocol.Resize {
340 var resizes []protocol.Resize
341 for _, message := range stream.messages() {
342 if resize, ok := message.(protocol.Resize); ok {
343 resizes = append(resizes, resize)
344 }
345 }
346 return resizes
347}
348
349// awaitStreamResizes waits until the stream sent at least want resize messages.
350func awaitStreamResizes(t *testing.T, stream *sessionTestStream, want int) []protocol.Resize {
351 t.Helper()
352 require.Eventually(t, func() bool { return len(streamResizes(stream)) >= want }, 5*time.Second, time.Millisecond,
353 "the attachment stream never sent %d resize messages", want)
354 return streamResizes(stream)
355}
356
357// TestSupervisorResizeBurstCoalescesAndRenderObservesLatestGeometry holds the
358// supervisor's state lock across a whole resize burst, so the collector records
359// every geometry while the repaint consumer cannot observe a partially applied
360// burst. The burst must coalesce into at most one pending invalidation and the
361// repaint that follows must sample the latest Terminal.Geometry: no render may
362// observe an intermediate size.
363func TestSupervisorResizeBurstCoalescesAndRenderObservesLatestGeometry(t *testing.T) {
364 picker := newAttachTestPicker()
365 harness := startResizeHarness(t, resizeHarnessOptions{picker: picker})
366 require.Eventually(t, func() bool { return harness.sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
367 harness.recorder.drain()
368 baseline := harness.recorder.count()
369
370 burst := []domain.Geometry{
371 {Size: domain.Size{Cols: 90, Rows: 30}},
372 {Size: domain.Size{Cols: 100, Rows: 35}},
373 {Size: domain.Size{Cols: 120, Rows: 40}},
374 }
375 var pending int
376 func() {
377 harness.sup.mu.Lock()
378 defer harness.sup.mu.Unlock()
379 for _, geometry := range burst {
380 harness.terminal.resize(geometry)
381 }
382 require.Eventually(t, func() bool {
383 return harness.sup.attachments.geometryProgress() == uint64(len(burst))
384 }, 5*time.Second, time.Millisecond, "the collector records every geometry of the burst")
385 pending = harness.sup.attachments.pendingPresentationSignals()
386 }()
387 require.Equal(t, 1, pending,
388 "a resize burst coalesces into exactly one pending presentation invalidation")
389
390 latest := burst[len(burst)-1]
391 require.Eventually(t, func() bool { return harness.recorder.count() > baseline }, 5*time.Second, time.Millisecond,
392 "the coalesced burst produces a repaint")
393 for _, sample := range harness.recorder.samples()[baseline:] {
394 require.Equal(t, PresentPicker, sample.state.Presentation)
395 require.Equal(t, latest.Size, sample.geometry.Size,
396 "every repaint after the burst observes the latest Terminal.Geometry")
397 }
398 require.Equal(t, latest.Size, harness.recorder.last().geometry.Size)
399}
400
401// TestSupervisorIdlePickerRepaintsOnResizeWithoutInputOrPublication proves the
402// core idle behavior: a resize wakes the established-connection wait and
403// repaints the picker even though no terminal input arrived and no broker
404// snapshot was published.
405func TestSupervisorIdlePickerRepaintsOnResizeWithoutInputOrPublication(t *testing.T) {
406 picker := newAttachTestPicker()
407 harness := startResizeHarness(t, resizeHarnessOptions{picker: picker})
408 require.Eventually(t, func() bool { return harness.sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
409
410 revision := harness.service.hub.current().Revision
411 harness.recorder.drain()
412
413 resized := domain.Geometry{Size: domain.Size{Cols: 132, Rows: 43}}
414 harness.terminal.resize(resized)
415
416 sample := awaitResizeRender(t, harness.recorder, func(sample resizeRenderSample) bool {
417 return sample.state.Presentation == PresentPicker && sample.state.Connectivity == ConnectivityReady
418 })
419 require.Equal(t, State{Presentation: PresentPicker, Connectivity: ConnectivityReady, Generation: 1}, sample.state,
420 "a resize repaint reports the untouched ready state")
421 require.Equal(t, resized.Size, sample.geometry.Size, "the render samples the latest terminal geometry")
422 require.Equal(t, revision, harness.service.hub.current().Revision, "a resize repaint must not need a broker publication")
423 require.Zero(t, picker.consumedCount(), "a resize repaint must not consume terminal input")
424 require.Zero(t, harness.sup.attachments.pendingPresentationSignals(), "the coalesced invalidation was consumed")
425}
426
427// TestSupervisorResizeRepaintsConnectingRetryAndNonRetryableWaiting proves the
428// resize invalidation reaches every pre-attachment waiting state the supervisor
429// parks in: an in-flight broker connect, the retry cadence, and the
430// non-retryable picker wait.
431func TestSupervisorResizeRepaintsConnectingRetryAndNonRetryableWaiting(t *testing.T) {
432 unavailable := ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker down"}
433 incompatible := ports.BrokerError{Code: ports.BrokerErrorIncompatible, Text: "protocol"}
434
435 tests := []struct {
436 name string
437 // start parks one supervisor on the wait under test and returns the state
438 // it must be presenting when the resize arrives.
439 start func(t *testing.T) (*resizeSupervisorHarness, State)
440 }{
441 {
442 name: "broker connect in flight",
443 start: func(t *testing.T) (*resizeSupervisorHarness, State) {
444 harness := startResizeHarness(t, resizeHarnessOptions{
445 connect: func(ctx context.Context, _ int) (ports.BrokerNavigator, error) {
446 <-ctx.Done()
447 return nil, ctx.Err()
448 },
449 })
450 require.Equal(t, 1, harness.connector.awaitStart(t))
451 return harness, State{Presentation: PresentPicker, Connectivity: ConnectivityConnectingBroker, Generation: 1}
452 },
453 },
454 {
455 name: "retry cadence wait",
456 start: func(t *testing.T) (*resizeSupervisorHarness, State) {
457 harness := startResizeHarness(t, resizeHarnessOptions{
458 connect: func(context.Context, int) (ports.BrokerNavigator, error) { return nil, unavailable },
459 })
460 harness.clock.awaitTimer(t)
461 // The cadence exists because the first attempt already failed, so its
462 // connector-start notification is still queued. Consume it now; the
463 // retry assertion below must observe the next attempt's notification,
464 // not this initial one.
465 require.Equal(t, 1, harness.connector.awaitStart(t))
466 return harness, State{Presentation: PresentPicker, Connectivity: ConnectivityRetryWait, Attempt: 1, Generation: 1, Err: unavailable}
467 },
468 },
469 {
470 name: "non-retryable picker wait",
471 start: func(t *testing.T) (*resizeSupervisorHarness, State) {
472 harness := startResizeHarness(t, resizeHarnessOptions{
473 connect: func(context.Context, int) (ports.BrokerNavigator, error) { return nil, incompatible },
474 })
475 require.Eventually(t, func() bool {
476 state := harness.sup.State()
477 return state.Connectivity == ConnectivityDisconnected && state.Err != nil
478 }, 5*time.Second, time.Millisecond)
479 return harness, State{Presentation: PresentPicker, Connectivity: ConnectivityDisconnected, Generation: 1, Err: incompatible}
480 },
481 },
482 }
483
484 for _, tt := range tests {
485 t.Run(tt.name, func(t *testing.T) {
486 harness, want := tt.start(t)
487 require.Equal(t, want, harness.sup.State(), "the supervisor is parked exactly on the wait under test")
488 harness.recorder.drain()
489
490 var retryTimer *supervisorTestTimer
491 if tt.name == "retry cadence wait" {
492 retryTimer = harness.clock.snapshotTimers()[0]
493 require.Equal(t, 1, harness.clock.timerCount())
494 require.Equal(t, 1, harness.clock.liveTimers())
495 }
496
497 resized := domain.Geometry{Size: domain.Size{Cols: 111, Rows: 41}}
498 harness.terminal.resize(resized)
499
500 sample := awaitResizeRender(t, harness.recorder, func(sample resizeRenderSample) bool {
501 return sample.state.Presentation == PresentPicker
502 })
503 require.Equal(t, want, sample.state, "a resize repaint reports the parked state unchanged")
504 require.Equal(t, resized.Size, sample.geometry.Size, "the render samples the latest terminal geometry")
505 require.Zero(t, harness.sup.attachments.pendingPresentationSignals(), "the coalesced invalidation was consumed")
506 require.Equal(t, want.Connectivity, harness.sup.State().Connectivity, "a repaint never advances connectivity")
507 if retryTimer != nil {
508 require.Equal(t, 1, harness.clock.timerCount(), "resize must not replace the retry timer")
509 require.Equal(t, 1, harness.clock.liveTimers(), "the original retry timer remains live")
510 retryTimer.fire()
511 require.Equal(t, 2, harness.connector.awaitStart(t), "the original timer advances the next attempt")
512 }
513 })
514 }
515}
516
517// TestAttachmentHostResizeBurstCoalescesAndKeepsGeometryWakeupsIndependent pins
518// the collector itself: a burst of valid resizes is coalesced into exactly one
519// buffered presentation invalidation while the attachment geometry sequence
520// still advances to the latest geometry, and draining the presentation signal
521// cannot steal the attachment's own geometry wakeup.
522func TestAttachmentHostResizeBurstCoalescesAndKeepsGeometryWakeupsIndependent(t *testing.T) {
523 resizes := make(chan domain.Geometry, 8)
524 host := newAttachmentHost(attachmentHostConfig{})
525 host.resizes = resizes
526 host.ownerBoundary = true
527 ctx, cancel := context.WithCancel(context.Background())
528 defer cancel()
529 host.startGeometry(ctx)
530 defer host.stopGeometry()
531
532 fg := host.newForeground(AttachmentToken{Generation: 1, Attempt: 1}, newSessionTestStream())
533 require.True(t, host.authority.grant(fg))
534
535 burst := []domain.Geometry{
536 {Size: domain.Size{Cols: 90, Rows: 30}},
537 {Size: domain.Size{Cols: 100, Rows: 35}},
538 {Size: domain.Size{Cols: 120, Rows: 40}},
539 }
540 for _, geometry := range burst {
541 resizes <- geometry
542 }
543 require.Eventually(t, func() bool {
544 return host.geometryProgress() == uint64(len(burst)) && host.pendingPresentationSignals() == 1
545 }, 5*time.Second, time.Millisecond, "a resize burst must coalesce into one presentation invalidation")
546 require.Equal(t, 1, host.pendingGeometrySignals(), "the attachment geometry wakeup coalesces independently")
547
548 latest, ok := host.latestGeometry()
549 require.True(t, ok)
550 require.Equal(t, burst[len(burst)-1].Size, latest.Size, "the latest geometry wins the coalesced burst")
551
552 // Draining the presentation invalidation must not touch the attachment's
553 // geometry wakeup: the foreground still receives the latest geometry.
554 <-host.presentationUpdate
555 geometry, ok := fg.Resize(context.Background())
556 require.True(t, ok)
557 require.Equal(t, burst[len(burst)-1].Size, geometry.Size)
558
559 next := domain.Geometry{Size: domain.Size{Cols: 132, Rows: 43}}
560 resizes <- next
561 geometry, ok = fg.Resize(context.Background())
562 require.True(t, ok)
563 require.Equal(t, next.Size, geometry.Size)
564 require.Equal(t, 1, host.pendingPresentationSignals(), "every resize still raises a coalesced invalidation")
565 fg.finish()
566}
567
568// TestSupervisorResizeDuringAttachmentSettlementPreservesSessionOutput pins the
569// safe boundary this slice was narrowed to. Once Begin has admitted the
570// foreground it owns the shared terminal writer before MarkAttached and the
571// initial publication, so a resize invalidation that lands in the settlement
572// wait must not be consumed there: the composition's renderer would write picker
573// bytes over the session output the foreground is writing. The invalidation
574// stays coalesced and buffered through connecting and attached, and is consumed
575// with a repaint at the latest geometry only once the attachment returns to the
576// picker.
577func TestSupervisorResizeDuringAttachmentSettlementPreservesSessionOutput(t *testing.T) {
578 picker := newAttachTestPicker()
579 harness := startResizeHarness(t, resizeHarnessOptions{picker: picker})
580 registry := &resizeStreamRegistry{}
581 harness.service.setOpenStream(registry.open)
582
583 picker.commit(sessionTestRequest(true))
584 stream := registry.stream(t, 0)
585 require.Eventually(t, func() bool {
586 return harness.sup.State().Presentation == PresentConnecting
587 }, 5*time.Second, time.Millisecond, "attachment settlement exposes connecting presentation")
588 harness.recorder.drain()
589
590 resized := domain.Geometry{Size: domain.Size{Cols: 109, Rows: 39}}
591 harness.terminal.resize(resized)
592 require.Eventually(t, func() bool { return harness.sup.attachments.pendingPresentationSignals() == 1 },
593 5*time.Second, time.Millisecond, "the collector still records the resize while connecting")
594
595 // The admitted foreground commits its initial publication through the same
596 // terminal. The supervisor must not have painted over it, and the resize must
597 // not have driven a connecting repaint at the new geometry.
598 deliverReadyStream(t, stream)
599 awaitAttachedState(t, harness.sup)
600 // The attached worker's palette query draws nothing; strip it.
601 written := strings.ReplaceAll(harness.terminal.written(), paletteColorBatch, "")
602 at := strings.Index(written, "\x1b[Hready")
603 require.GreaterOrEqual(t, at, 0, "the admitted foreground committed its initial publication")
604 require.Equal(t, "\x1b[Hready", written[at:],
605 "a resize during settlement must not append picker bytes over session output")
606 for _, sample := range harness.recorder.samples() {
607 require.False(t, sample.state.Presentation == PresentConnecting && sample.geometry.Size == resized.Size,
608 "a resize during settlement must not invoke the picker renderer for the connecting presentation")
609 }
610 require.Equal(t, 1, harness.sup.attachments.pendingPresentationSignals(),
611 "the invalidation stays buffered through connecting and attached")
612
613 // Back at the picker the preserved invalidation is consumed and repaints at
614 // the latest geometry. The transition back to the picker renders once; the
615 // consumed invalidation renders once more, so exactly two picker renders at
616 // the resized geometry follow the attachment. A run that never emitted the
617 // invalidation cannot reach this point, and a run that consumed it without a
618 // repaint leaves only one.
619 stream.deliver(protocol.Detached{Reason: protocol.ReasonSessionKilled})
620 awaitPickerState(t, harness.sup)
621 pickerAtResize := func(sample resizeRenderSample) bool {
622 return sample.state.Presentation == PresentPicker && sample.geometry.Size == resized.Size
623 }
624 awaitResizeRender(t, harness.recorder, pickerAtResize)
625 awaitResizeRender(t, harness.recorder, pickerAtResize)
626 require.Eventually(t, func() bool { return harness.sup.attachments.pendingPresentationSignals() == 0 },
627 5*time.Second, time.Millisecond, "the picker wait consumes the preserved invalidation")
628}
629
630// TestSupervisorResizeConnectingBurstStaysCoalescedAndBuffered proves the
631// invalidation a settlement wait must not consume is still coalesced: a burst of
632// resizes while the connecting presentation holds an admitted foreground leaves
633// exactly one pending signal, and the whole burst is then consumed by the picker
634// wait after the attachment ends without any intermediate repaint.
635func TestSupervisorResizeConnectingBurstStaysCoalescedAndBuffered(t *testing.T) {
636 picker := newAttachTestPicker()
637 harness := startResizeHarness(t, resizeHarnessOptions{picker: picker})
638 registry := &resizeStreamRegistry{}
639 harness.service.setOpenStream(registry.open)
640
641 picker.commit(sessionTestRequest(true))
642 stream := registry.stream(t, 0)
643 require.Eventually(t, func() bool {
644 return harness.sup.State().Presentation == PresentConnecting
645 }, 5*time.Second, time.Millisecond)
646
647 burst := []domain.Geometry{
648 {Size: domain.Size{Cols: 90, Rows: 30}},
649 {Size: domain.Size{Cols: 100, Rows: 35}},
650 {Size: domain.Size{Cols: 120, Rows: 40}},
651 }
652 for _, geometry := range burst {
653 harness.terminal.resize(geometry)
654 }
655 require.Eventually(t, func() bool {
656 return harness.sup.attachments.geometryProgress() == uint64(len(burst))
657 }, 5*time.Second, time.Millisecond, "the collector records every geometry of the burst")
658 require.Eventually(t, func() bool { return harness.sup.attachments.pendingPresentationSignals() == 1 },
659 5*time.Second, time.Millisecond, "a connecting resize burst coalesces into one buffered invalidation")
660
661 deliverReadyStream(t, stream)
662 awaitAttachedState(t, harness.sup)
663 require.Equal(t, 1, harness.sup.attachments.pendingPresentationSignals(),
664 "the coalesced burst is preserved through connecting and attached")
665
666 stream.deliver(protocol.Detached{Reason: protocol.ReasonSessionKilled})
667 awaitPickerState(t, harness.sup)
668 pickerAtLatest := func(sample resizeRenderSample) bool {
669 return sample.state.Presentation == PresentPicker && sample.geometry.Size == burst[len(burst)-1].Size
670 }
671 awaitResizeRender(t, harness.recorder, pickerAtLatest)
672 awaitResizeRender(t, harness.recorder, pickerAtLatest)
673 require.Eventually(t, func() bool { return harness.sup.attachments.pendingPresentationSignals() == 0 },
674 5*time.Second, time.Millisecond, "the coalesced burst is consumed by the picker wait, once")
675}
676
677// awaitAttachedSettlement waits until the supervisor's live attachment has
678// claimed the attached transition and the settlement wait has drained every
679// earlier invalidation. It is the unambiguous boundary the attached-resize
680// assertion needs: once the current foreground reports attached, the
681// settlement wait has disarmed its presentation arm, so a resize pushed after
682// this point must stay buffered for the picker instead of being consumed.
683func awaitAttachedSettlement(t *testing.T, harness *resizeSupervisorHarness) {
684 t.Helper()
685 require.Eventually(t, func() bool {
686 harness.sup.mu.Lock()
687 attachedPresentation := harness.sup.state.Presentation == PresentAttached
688 harness.sup.mu.Unlock()
689 foreground := harness.sup.attachments.authority.foreground()
690 return attachedPresentation && foreground != nil && foreground.Attached()
691 }, 5*time.Second, time.Millisecond, "the attachment never claimed the attached boundary")
692 require.Eventually(t, func() bool { return harness.sup.attachments.pendingPresentationSignals() == 0 },
693 5*time.Second, time.Millisecond, "a pre-resize invalidation was not drained before the boundary")
694}
695
696// TestSupervisorAttachedResizeKeepsGeometryWakeupAndPreservesRepaint proves an
697// attached foreground is not painted over and its geometry wakeup is not
698// stolen: the live attachment receives the typed resize while the coalesced
699// presentation invalidation stays buffered for the picker, and is then consumed
700// by the supervisor's next picker wait.
701func TestSupervisorAttachedResizeKeepsGeometryWakeupAndPreservesRepaint(t *testing.T) {
702 picker := newAttachTestPicker()
703 harness := startResizeHarness(t, resizeHarnessOptions{picker: picker})
704 registry := &resizeStreamRegistry{}
705 harness.service.setOpenStream(registry.open)
706
707 picker.commit(sessionTestRequest(true))
708 stream := registry.stream(t, 0)
709 deliverReadyStream(t, stream)
710 // Settle the attached boundary before the resize: the presentation is
711 // attached and the live foreground has claimed attached, and every earlier
712 // invalidation is drained, so the resize below lands unambiguously on an
713 // attached settlement wait.
714 awaitAttachedSettlement(t, harness)
715 harness.recorder.drain()
716 baseline := harness.recorder.count()
717
718 resized := domain.Geometry{Size: domain.Size{Cols: 100, Rows: 40}}
719 harness.terminal.resize(resized)
720
721 resizes := awaitStreamResizes(t, stream, 1)
722 require.Equal(t, resized.Size, resizes[0].Size, "the live attachment still receives its geometry wakeup")
723 // The attachment geometry wakeup and the presentation invalidation are
724 // pushed by the collector independently, and the worker can observe the new
725 // geometry through the recorded sequence before the invalidation is pushed.
726 // Wait for the coalesced token to be buffered; a supervisor that wrongly
727 // consumed it while attached could never make this count return to one, so
728 // this stays a behavioral check rather than a sleep.
729 require.Eventually(t, func() bool { return harness.sup.attachments.pendingPresentationSignals() == 1 },
730 5*time.Second, time.Millisecond, "the supervisor does not consume the invalidation while attached")
731 require.Equal(t, baseline, harness.recorder.count(), "an attached presentation is never painted over")
732 require.Equal(t, 1, harness.sup.attachments.pendingPresentationSignals(), "the invalidation stays buffered until the picker wait")
733
734 stream.deliver(protocol.Detached{Reason: protocol.ReasonSessionKilled})
735 awaitPickerState(t, harness.sup)
736 require.Eventually(t, func() bool { return harness.sup.attachments.pendingPresentationSignals() == 0 }, 5*time.Second, time.Millisecond,
737 "the preserved invalidation is consumed by the picker wait")
738}
739
740// TestSupervisorResizeAfterDetachReachesOnlyReplacementAttachment proves a
741// resize is delivered to the valid replacement attachment token only. A resize
742// that lands in the detach/reattach gap reaches no stream at all: it repaints the
743// picker, the retired token's stream keeps exactly its own earlier resize, and
744// the replacement inherits the gap geometry and receives only its own later
745// resize under its own (generation, attempt) identity.
746func TestSupervisorResizeAfterDetachReachesOnlyReplacementAttachment(t *testing.T) {
747 picker := newAttachTestPicker()
748 harness := startResizeHarness(t, resizeHarnessOptions{picker: picker})
749 registry := &resizeStreamRegistry{}
750 harness.service.setOpenStream(registry.open)
751
752 picker.commit(sessionTestRequest(true))
753 first := registry.stream(t, 0)
754 deliverReadyStream(t, first)
755 awaitAttachedState(t, harness.sup)
756 require.Equal(t, AttachmentToken{Generation: 1, Attempt: 1}, authorityToken(&harness.sup.attachments.authority))
757
758 firstResize := domain.Geometry{Size: domain.Size{Cols: 101, Rows: 41}}
759 harness.terminal.resize(firstResize)
760 firstResizes := awaitStreamResizes(t, first, 1)
761 require.Equal(t, firstResize.Size, firstResizes[0].Size)
762
763 // The attachment detaches: its generation is retired before any replacement
764 // exists.
765 first.deliver(protocol.Detached{Reason: protocol.ReasonSessionKilled})
766 awaitPickerState(t, harness.sup)
767 require.True(t, authorityToken(&harness.sup.attachments.authority).IsZero(), "the retired token no longer owns the foreground")
768
769 // A resize in the detach/reattach gap: no attachment owns the foreground, so
770 // the retired stream must not receive it, only the picker repaints, and the
771 // signal is consumed by the picker wait.
772 harness.recorder.drain()
773 gapResize := domain.Geometry{Size: domain.Size{Cols: 110, Rows: 45}}
774 harness.terminal.resize(gapResize)
775 gapSample := awaitResizeRender(t, harness.recorder, func(sample resizeRenderSample) bool {
776 return sample.state.Presentation == PresentPicker && sample.geometry.Size == gapResize.Size
777 })
778 require.Equal(t, gapResize.Size, gapSample.geometry.Size, "a gap resize repaints the picker")
779 require.Eventually(t, func() bool { return harness.sup.attachments.pendingPresentationSignals() == 0 }, 5*time.Second, time.Millisecond,
780 "the gap invalidation is consumed by the picker wait")
781 require.Len(t, streamResizes(first), 1, "the retired attachment token must not receive a gap resize")
782
783 // A replacement attachment claims the next attempt identity. It carries the
784 // latest geometry in its own session state, which is the gap resize, so the
785 // only stream that ever observes it is the valid replacement.
786 picker.commit(sessionTestRequest(true))
787 second := registry.stream(t, 1)
788 deliverReadyStream(t, second)
789 awaitAttachedState(t, harness.sup)
790 require.Equal(t, AttachmentToken{Generation: 1, Attempt: 2}, authorityToken(&harness.sup.attachments.authority))
791 inherited := awaitStreamResizes(t, second, 1)
792 require.Equal(t, gapResize.Size, inherited[0].Size, "the replacement inherits the latest geometry")
793
794 // The replacement's own resize reaches it and nothing else.
795 secondResize := domain.Geometry{Size: domain.Size{Cols: 120, Rows: 50}}
796 harness.terminal.resize(secondResize)
797 replacement := awaitStreamResizes(t, second, 2)
798 require.Equal(t, secondResize.Size, replacement[1].Size, "only the replacement token receives its resize")
799 require.Len(t, streamResizes(first), 1, "a retired attachment token must never receive a replacement's resize")
800}
801
802// TestResizeCollectorAndSupervisorJoinOnClosedEventsAndCancellation proves the
803// collector and the supervisor's owned waits end cleanly when ResizeEvents is
804// already closed and when the process context is cancelled: no path blocks on a
805// closed or abandoned channel.
806func TestResizeCollectorAndSupervisorJoinOnClosedEventsAndCancellation(t *testing.T) {
807 t.Run("collector joins a closed ResizeEvents channel", func(t *testing.T) {
808 closed := make(chan domain.Geometry)
809 close(closed)
810 host := newAttachmentHost(attachmentHostConfig{})
811 host.resizes = closed
812 ctx, cancel := context.WithCancel(context.Background())
813 defer cancel()
814 host.startGeometry(ctx)
815 select {
816 case <-host.geometryDone:
817 case <-time.After(5 * time.Second):
818 t.Fatal("the collector did not end on closed resize events")
819 }
820 host.stopGeometry()
821 require.Nil(t, host.geometryDone)
822 require.Nil(t, host.geometryCancel)
823
824 // The host can be started again with a live channel and joined by
825 // cancellation, even with a coalesced invalidation still pending.
826 resizes := make(chan domain.Geometry, 2)
827 host.resizes = resizes
828 host.startGeometry(ctx)
829 resizes <- domain.Geometry{Size: domain.Size{Cols: 90, Rows: 30}}
830 resizes <- domain.Geometry{Size: domain.Size{Cols: 91, Rows: 31}}
831 require.Eventually(t, func() bool { return host.pendingPresentationSignals() == 1 }, 5*time.Second, time.Millisecond)
832 cancel()
833 select {
834 case <-host.geometryDone:
835 case <-time.After(5 * time.Second):
836 t.Fatal("the collector did not end on cancellation")
837 }
838 host.stopGeometry()
839 require.Nil(t, host.geometryDone)
840 })
841
842 t.Run("supervisor run joins on closed ResizeEvents and cancellation", func(t *testing.T) {
843 harness := startResizeHarness(t, resizeHarnessOptions{
844 closeResizes: true,
845 connect: func(context.Context, int) (ports.BrokerNavigator, error) {
846 return nil, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "down"}
847 },
848 })
849 require.Equal(t, 1, harness.connector.awaitStart(t))
850 harness.cancel()
851 require.ErrorIs(t, harness.waitRun(t), context.Canceled)
852 _, _, restores := harness.terminal.counts()
853 require.Equal(t, 1, restores, "raw mode is restored exactly once")
854 require.Nil(t, harness.sup.attachments.geometryDone, "Run joined the geometry collector")
855 require.Nil(t, harness.sup.attachments.geometryCancel)
856 })
857}
858
859// TestSupervisorResizeInvalidationRendersOnlyPickerPresentations pins the
860// presentation fence directly: only PresentPicker may be repainted by a resize,
861// so the pre-attachment connecting transition, an attached foreground, and a
862// terminating process are never painted over. It also proves the fence is
863// behavioral, not just counted: with the session byte already in the terminal,
864// a connecting repaint would change the renderer count and append picker bytes
865// over the session output. A supervisor without a renderer is a no-op.
866func TestSupervisorResizeInvalidationRendersOnlyPickerPresentations(t *testing.T) {
867 terminal := newResizeTestTerminal(newAttachTestReader(), domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}})
868 recorder := newResizeRenderRecorder(terminal)
869 sup := mustSupervisor(t, SupervisorConfig{
870 Connector: newSupervisorTestConnector(nil),
871 Terminal: terminal,
872 Clock: newSupervisorTestClock(),
873 Render: recorder.render,
874 })
875
876 tests := []struct {
877 name string
878 presentation Presentation
879 wantRender bool
880 }{
881 {name: "picker is repainted", presentation: PresentPicker, wantRender: true},
882 {name: "connecting is never painted over", presentation: PresentConnecting},
883 {name: "attached is never painted over", presentation: PresentAttached},
884 {name: "terminating is never repainted", presentation: PresentTerminating},
885 }
886 for _, tt := range tests {
887 t.Run(tt.name, func(t *testing.T) {
888 // A session frame as an admitted foreground would have written it, so
889 // any picker repaint for a refused presentation is observable.
890 const session = "\x1b[Hsession output"
891 terminal.setWritten(session)
892 sup.mu.Lock()
893 sup.state = State{Presentation: tt.presentation, Connectivity: ConnectivityReady}
894 sup.mu.Unlock()
895 recorder.drain()
896 baseline := recorder.count()
897
898 sup.renderResizeInvalidation()
899
900 if tt.wantRender {
901 sample := awaitResizeRender(t, recorder, func(resizeRenderSample) bool { return true })
902 require.Equal(t, tt.presentation, sample.state.Presentation)
903 return
904 }
905 require.Equal(t, baseline, recorder.count(), "presentation %s must not be repainted by a resize", tt.presentation)
906 require.Equal(t, session, terminal.written(),
907 "presentation %s must never overpaint session output", tt.presentation)
908 })
909 }
910
911 // A supervisor with no renderer configured through the internal zero value
912 // never panics, including on a nil receiver.
913 (&Supervisor{}).renderResizeInvalidation()
914 var nilSupervisor *Supervisor
915 nilSupervisor.renderResizeInvalidation()
916}
917
918// TestSupervisorPresentationInvalidationNilHostIsSafe pins the nil-host
919// contract of the presentation invalidation helper and of a wait that selects
920// on it. A supervisor whose attachment host is absent (the zero value, or a run
921// whose host was detached) must expose a nil signal and still park on its
922// remaining signals instead of dereferencing a nil host.
923func TestSupervisorPresentationInvalidationNilHostIsSafe(t *testing.T) {
924 t.Run("helper returns nil without a host", func(t *testing.T) {
925 var nilSupervisor *Supervisor
926 require.Nil(t, nilSupervisor.presentationInvalidation(), "a nil receiver exposes no invalidation")
927
928 zero := &Supervisor{}
929 require.Nil(t, zero.presentationInvalidation(), "a zero-value supervisor exposes no invalidation")
930
931 // A fully built supervisor exposes its host's signal; once the host is
932 // gone the helper must fall back to nil rather than panic.
933 terminal := newResizeTestTerminal(newAttachTestReader(), domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}})
934 sup := mustSupervisor(t, SupervisorConfig{
935 Connector: newSupervisorTestConnector(nil),
936 Terminal: terminal,
937 Clock: newSupervisorTestClock(),
938 })
939 require.NotNil(t, sup.presentationInvalidation(), "a live host exposes its coalesced invalidation")
940 sup.attachments = nil
941 require.Nil(t, sup.presentationInvalidation(), "a detached host exposes no invalidation")
942 })
943
944 t.Run("a wait with a nil host still settles", func(t *testing.T) {
945 terminal := newResizeTestTerminal(newAttachTestReader(), domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}})
946 clock := newSupervisorTestClock()
947 sup := mustSupervisor(t, SupervisorConfig{
948 Connector: newSupervisorTestConnector(nil),
949 Terminal: terminal,
950 Clock: clock,
951 Jitter: func() float64 { return 0 },
952 })
953 sup.attachments = nil
954 require.Nil(t, sup.presentationInvalidation())
955 // One transient failure makes the cadence non-zero, so waitBackoff
956 // arms a real timer.
957 sup.transition(supervisorEvent{kind: supervisorTransientFailure})
958
959 // The reader never yields, so the EOF arm stays closed and the timer is
960 // the only signal that can settle this wait.
961 reader := newAttachTestReader()
962 lifetime := startTerminalInputLifetime(reader, nil)
963 t.Cleanup(func() {
964 reader.close()
965 lifetime.stop()
966 })
967
968 settled := make(chan struct{})
969 var terminated bool
970 var waitErr error
971 go func() {
972 defer close(settled)
973 terminated, waitErr = sup.waitBackoff(context.Background(), lifetime)
974 }()
975 timer := clock.awaitTimer(t)
976 timer.fire()
977 select {
978 case <-settled:
979 case <-time.After(5 * time.Second):
980 t.Fatal("a nil-host backoff wait never settled on its timer")
981 }
982 require.False(t, terminated)
983 require.NoError(t, waitErr)
984 })
985
986 t.Run("a wait with a nil host settles on cancellation", func(t *testing.T) {
987 terminal := newResizeTestTerminal(newAttachTestReader(), domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}})
988 clock := newSupervisorTestClock()
989 sup := mustSupervisor(t, SupervisorConfig{
990 Connector: newSupervisorTestConnector(nil),
991 Terminal: terminal,
992 Clock: clock,
993 Jitter: func() float64 { return 0 },
994 })
995 sup.attachments = nil
996 sup.transition(supervisorEvent{kind: supervisorTransientFailure})
997
998 reader := newAttachTestReader()
999 lifetime := startTerminalInputLifetime(reader, nil)
1000 t.Cleanup(func() {
1001 reader.close()
1002 lifetime.stop()
1003 })
1004
1005 ctx, cancel := context.WithCancel(context.Background())
1006 settled := make(chan struct{})
1007 var terminated bool
1008 var waitErr error
1009 go func() {
1010 defer close(settled)
1011 terminated, waitErr = sup.waitBackoff(ctx, lifetime)
1012 }()
1013 clock.awaitTimer(t)
1014 cancel()
1015 select {
1016 case <-settled:
1017 case <-time.After(5 * time.Second):
1018 t.Fatal("a nil-host backoff wait never settled on cancellation")
1019 }
1020 require.True(t, terminated)
1021 require.ErrorIs(t, waitErr, context.Canceled)
1022 })
1023}