1package daemon
2
3import (
4 "context"
5 "io"
6 "log/slog"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/mock"
12 "github.com/stretchr/testify/require"
13
14 vt "github.com/bnema/vev-vt"
15 renderer "github.com/bnema/vev-vt/ansi"
16 "github.com/bnema/vev/internal/domain"
17 "github.com/bnema/vev/internal/ports"
18 portsmocks "github.com/bnema/vev/internal/ports/mocks"
19 "github.com/bnema/vev/internal/usecase/keys"
20)
21
22// --- test doubles -----------------------------------------------------------
23
24// stubClock returns timers whose channel never fires, so a scheduler under it
25// blocks in its debounce loop until the session context is cancelled. Used by
26// every test that is not specifically exercising the debounce.
27type stubClock struct{}
28
29func (stubClock) Now() time.Time { return time.Time{} }
30func (stubClock) NewTimer(time.Duration) ports.Timer { return stubTimer{} }
31
32type stubTimer struct{}
33
34func (stubTimer) C() <-chan time.Time { return nil }
35func (stubTimer) Reset(time.Duration) bool { return false }
36func (stubTimer) Stop() bool { return true }
37
38type signalClock struct {
39 called chan struct{}
40 timers chan *signalTimer
41 once sync.Once
42}
43
44func (c *signalClock) Now() time.Time { return time.Time{} }
45func (c *signalClock) NewTimer(d time.Duration) ports.Timer {
46 if c.called != nil {
47 c.once.Do(func() { close(c.called) })
48 }
49 if c.timers == nil {
50 return stubTimer{}
51 }
52 t := &signalTimer{ch: make(chan time.Time, 1), duration: d}
53 c.timers <- t
54 return t
55}
56
57type signalTimer struct {
58 ch chan time.Time
59 duration time.Duration
60}
61
62func (t *signalTimer) C() <-chan time.Time { return t.ch }
63func (t *signalTimer) Reset(time.Duration) bool { return false }
64func (t *signalTimer) Stop() bool { return true }
65
66func newTestDaemon(t testing.TB, ptys ports.PTYFactory, clk ports.Clock) *Daemon {
67 t.Helper()
68 return newTestDaemonWithCleanup(t, ptys, clk, true)
69}
70
71func createSessionForTest(d *Daemon, name string, ephemeral bool, cwd string, sz domain.Size, term terminalEnv, env []string, restoredTabNames ...[]string) (*session, error) {
72 d.mu.Lock()
73 defer d.mu.Unlock()
74 return d.createSessionLocked(name, ephemeral, cwd, sz, term, env, restoredTabNames...)
75}
76
77func newTestDaemonWithCleanup(t testing.TB, ptys ports.PTYFactory, clk ports.Clock, registerCleanup bool) *Daemon {
78 t.Helper()
79 d := New(ptys, clk, slog.New(slog.NewTextHandler(io.Discard, nil)))
80 d.serveCtx, d.serveCancel = context.WithCancel(context.Background())
81 d.hardCtx, d.hardCancel = context.WithCancel(context.Background())
82 if registerCleanup {
83 t.Cleanup(d.serveCancel)
84 t.Cleanup(d.hardCancel)
85 }
86 return d
87}
88
89// newBlockingPTY returns a MockPTY whose Read blocks until release is called,
90// then reports io.EOF (the "child exited" signal).
91func newBlockingPTY(t *testing.T) (*portsmocks.MockPTY, func()) {
92 t.Helper()
93 return newBlockingPTYWithWrites(t, nil)
94}
95
96func newBlockingPTYWithWrites(t *testing.T, writes chan<- []byte) (*portsmocks.MockPTY, func()) {
97 t.Helper()
98 p := portsmocks.NewMockPTY(t)
99 release := make(chan struct{})
100 var once sync.Once
101 unblock := func() { once.Do(func() { close(release) }) }
102 p.EXPECT().Read(mock.Anything).RunAndReturn(func(b []byte) (int, error) {
103 <-release
104 return 0, io.EOF
105 }).Maybe()
106 p.EXPECT().Write(mock.Anything).RunAndReturn(func(b []byte) (int, error) {
107 if writes != nil {
108 cp := append([]byte(nil), b...)
109 writes <- cp
110 }
111 return len(b), nil
112 }).Maybe()
113 p.EXPECT().Resize(mock.Anything).Return(nil).Maybe()
114 // Close unblocks a parked Read, exactly as the real PTY does (its Close
115 // closes the master fd, failing the in-flight read).
116 p.EXPECT().Close().RunAndReturn(func() error { unblock(); return nil }).Maybe()
117 p.EXPECT().Pid().Return(4242).Maybe()
118 return p, unblock
119}
120
121// newBlockingPTYWithFailingWrite returns a MockPTY whose Write always fails
122// with writeErr, simulating a wedged pty (or a dead child) that cannot
123// accept forwarded keystrokes.
124func newBlockingPTYWithFailingWrite(t *testing.T, writeErr error) (*portsmocks.MockPTY, func()) {
125 t.Helper()
126 p := portsmocks.NewMockPTY(t)
127 release := make(chan struct{})
128 var once sync.Once
129 unblock := func() { once.Do(func() { close(release) }) }
130 p.EXPECT().Read(mock.Anything).RunAndReturn(func(b []byte) (int, error) {
131 <-release
132 return 0, io.EOF
133 }).Maybe()
134 p.EXPECT().Write(mock.Anything).Return(0, writeErr).Maybe()
135 p.EXPECT().Resize(mock.Anything).Return(nil).Maybe()
136 p.EXPECT().Close().RunAndReturn(func() error { unblock(); return nil }).Maybe()
137 p.EXPECT().Pid().Return(4242).Maybe()
138 return p, unblock
139}
140
141// quietPTY is an independently closable PTY for background floating prewarms.
142// It intentionally has no testify expectations: tests using the factories below
143// are asserting their regular panes, not incidental asynchronous prewarms.
144type quietPTY struct{ done chan struct{} }
145
146func newQuietPTY() *quietPTY { return &quietPTY{done: make(chan struct{})} }
147func (p *quietPTY) Read([]byte) (int, error) { <-p.done; return 0, io.EOF }
148func (*quietPTY) Write(b []byte) (int, error) { return len(b), nil }
149func (p *quietPTY) Close() error {
150 select {
151 case <-p.done:
152 default:
153 close(p.done)
154 }
155 return nil
156}
157func (*quietPTY) Resize(domain.Size) error { return nil }
158func (*quietPTY) Pid() int { return 0 }
159func (*quietPTY) ForegroundPgid() (int, error) { return 0, nil }
160
161// blockingOpenFactory keeps activation tests deterministically in Warming
162// without starting pane reader/scheduler goroutines.
163type blockingOpenFactory struct {
164 release chan struct{}
165 once sync.Once
166}
167
168func newBlockingOpenFactory(t *testing.T, d *Daemon) *blockingOpenFactory {
169 t.Helper()
170 f := &blockingOpenFactory{release: make(chan struct{})}
171 t.Cleanup(func() {
172 f.once.Do(func() { close(f.release) })
173 d.sessWg.Wait()
174 })
175 return f
176}
177
178func (f *blockingOpenFactory) Open(_ context.Context, _ string, _ []string, _ []string, _ string, _ domain.Size) (ports.PTY, error) {
179 <-f.release
180 return nil, io.ErrClosedPipe
181}
182
183// newFactory keeps the supplied fixture for normal panes. Floating prewarms
184// have a distinct (smaller) geometry and receive independent quiet PTYs, so
185// they cannot consume test output or add post-test mock calls.
186func newFactory(t *testing.T, p ports.PTY) *portsmocks.MockPTYFactory {
187 t.Helper()
188 f := portsmocks.NewMockPTYFactory(t)
189 var normal domain.Size
190 f.EXPECT().Open(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
191 func(_ context.Context, _ string, _ []string, _ []string, _ string, sz domain.Size) (ports.PTY, error) {
192 if !normal.Valid() {
193 normal = sz
194 }
195 if sz == normal {
196 return p, nil
197 }
198 return newQuietPTY(), nil
199 },
200 ).Maybe()
201 return f
202}
203
204func newFactorySeq(t *testing.T, ptys ...ports.PTY) *portsmocks.MockPTYFactory {
205 t.Helper()
206 f := portsmocks.NewMockPTYFactory(t)
207 var mu sync.Mutex
208 var normal domain.Size
209 next := 0
210 f.EXPECT().Open(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
211 func(_ context.Context, _ string, _ []string, _ []string, _ string, sz domain.Size) (ports.PTY, error) {
212 mu.Lock()
213 defer mu.Unlock()
214 if !normal.Valid() {
215 normal = sz
216 }
217 if sz != normal {
218 return newQuietPTY(), nil
219 }
220 if next >= len(ptys) {
221 return newQuietPTY(), nil
222 }
223 p := ptys[next]
224 next++
225 return p, nil
226 },
227 ).Maybe()
228 return f
229}
230
231// newConn scripts a MockTransport: Recv yields first then more (in order),
232// then blocks until the connection is released or Closed (returning io.EOF).
233// Every Send is captured on the returned channel.
234func newConn(t *testing.T, first ports.Frame, more ...ports.Frame) (*portsmocks.MockTransport, chan ports.Frame, func()) {
235 t.Helper()
236 tr := portsmocks.NewMockTransport(t)
237 sends := make(chan ports.Frame, 64)
238 recvCh := make(chan ports.Frame, 1+len(more))
239 recvCh <- first
240 for _, f := range more {
241 recvCh <- f
242 }
243 done := make(chan struct{})
244 var once sync.Once
245 closeDone := func() { once.Do(func() { close(done) }) }
246
247 tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
248 select {
249 case f := <-recvCh:
250 return f, nil
251 case <-done:
252 return ports.Frame{}, io.EOF
253 }
254 }).Maybe()
255 tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f ports.Frame) error {
256 sends <- f
257 return nil
258 }).Maybe()
259 tr.EXPECT().Close().RunAndReturn(func() error { closeDone(); return nil }).Maybe()
260 return tr, sends, closeDone
261}
262
263func mustHello(intent uint8, name string, sz domain.Size) ports.Frame {
264 return ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(ports.Hello{
265 Version: ports.ProtocolVersion, Intent: intent, Name: name, Size: sz, TermEnv: "xterm-256color",
266 })}
267}
268
269func frameInput(data []byte) ports.Frame {
270 return ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: data})}
271}
272
273func screenLineText(s *vt.Screen, y int) string {
274 out := make([]rune, s.Frame.Width)
275 for x := range s.Frame.Width {
276 out[x] = s.Frame.At(x, y).Rune
277 }
278 return string(out)
279}
280
281// awaitFrame waits for the next frame of type typ on ch, failing on timeout.
282func awaitFrame(t *testing.T, ch chan ports.Frame, typ ports.MsgType) ports.Frame {
283 t.Helper()
284 deadline := time.After(2 * time.Second)
285 for {
286 select {
287 case f := <-ch:
288 if f.Type == typ {
289 return f
290 }
291 case <-deadline:
292 t.Fatalf("timed out waiting for frame type %d", typ)
293 }
294 }
295}
296
297func firstSession(d *Daemon) *session {
298 d.mu.Lock()
299 defer d.mu.Unlock()
300 for _, s := range d.sessions {
301 if s != nil {
302 return s
303 }
304 }
305 return nil
306}
307
308func sessionCount(d *Daemon) int {
309 d.mu.Lock()
310 defer d.mu.Unlock()
311 return len(d.sessions)
312}
313
314func listSessions(t *testing.T, d *Daemon) ports.Sessions {
315 t.Helper()
316 tr, sends, _ := newConn(t, ports.Frame{Type: ports.MsgList, Payload: ports.MarshalList(ports.List{})})
317 d.handleList(tr)
318 f := awaitFrame(t, sends, ports.MsgSessions)
319 sessions, err := ports.UnmarshalSessions(f.Payload)
320 require.NoError(t, err)
321 return sessions
322}
323
324// welcomeBlockingTransport holds the route-level handshake precisely between
325// attachment publication and Welcome acceptance. It uses the generated
326// transport mock while channels make both sides of the wire deterministic.
327type welcomeBlockingTransport struct {
328 tr *portsmocks.MockTransport
329 sends chan ports.Frame
330 welcomeEntered chan struct{}
331 releaseWelcome chan struct{}
332 recvDone chan struct{}
333 releaseOnce sync.Once
334 closeOnce sync.Once
335}
336
337func newWelcomeBlockingTransport(t *testing.T) *welcomeBlockingTransport {
338 t.Helper()
339 b := &welcomeBlockingTransport{
340 sends: make(chan ports.Frame, 8),
341 welcomeEntered: make(chan struct{}),
342 releaseWelcome: make(chan struct{}),
343 recvDone: make(chan struct{}),
344 }
345 b.tr = portsmocks.NewMockTransport(t)
346 b.tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f ports.Frame) error {
347 b.sends <- f
348 if f.Type == ports.MsgWelcome {
349 close(b.welcomeEntered)
350 <-b.releaseWelcome
351 }
352 return nil
353 }).Maybe()
354 b.tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
355 <-b.recvDone
356 return ports.Frame{}, io.EOF
357 }).Maybe()
358 b.tr.EXPECT().Close().Return(nil).Maybe()
359 t.Cleanup(b.finish)
360 return b
361}
362
363func (b *welcomeBlockingTransport) release() {
364 b.releaseOnce.Do(func() { close(b.releaseWelcome) })
365}
366
367func (b *welcomeBlockingTransport) finish() {
368 b.release()
369 b.closeOnce.Do(func() { close(b.recvDone) })
370}
371
372func newCapturingTransport(t testing.TB) (*portsmocks.MockTransport, chan ports.Frame) {
373 t.Helper()
374 tr := portsmocks.NewMockTransport(t)
375 sends := make(chan ports.Frame, 64)
376 tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f ports.Frame) error {
377 sends <- f
378 return nil
379 }).Maybe()
380 tr.EXPECT().Close().Return(nil).Maybe()
381 return tr, sends
382}
383
384func newManualSessionWithPTYs(t testing.TB, ptys ...ports.PTY) (*Daemon, *session, *attachedClient, chan ports.Frame) {
385 t.Helper()
386 return newManualSessionWithPTYsClock(t, stubClock{}, ptys...)
387}
388
389func newManualSessionWithPTYsClock(t testing.TB, clock ports.Clock, ptys ...ports.PTY) (*Daemon, *session, *attachedClient, chan ports.Frame) {
390 t.Helper()
391 return newManualSessionWithPTYsClockCleanup(t, clock, true, ptys...)
392}
393
394func newManualSessionWithPTYsCleanup(t testing.TB, registerCleanup bool, ptys ...ports.PTY) (*Daemon, *session, *attachedClient, chan ports.Frame) {
395 t.Helper()
396 return newManualSessionWithPTYsClockCleanup(t, stubClock{}, registerCleanup, ptys...)
397}
398
399func newManualSessionWithPTYsClockCleanup(t testing.TB, clock ports.Clock, registerCleanup bool, ptys ...ports.PTY) (*Daemon, *session, *attachedClient, chan ports.Frame) {
400 t.Helper()
401 d := newTestDaemonWithCleanup(t, nil, clock, registerCleanup)
402 tr, sends := newCapturingTransport(t)
403 ac := &attachedClient{tr: tr, output: newOutputStateStream(), size: domain.Size{Cols: 80, Rows: 24}}
404 ac.initOverlays()
405 sctx, cancel := context.WithCancel(d.serveCtx)
406 tabs := make([]*tab, 0, len(ptys))
407 for _, p := range ptys {
408 wctx, wcancel := context.WithCancel(sctx)
409 tb := newTab(p, domain.Size{Cols: 80, Rows: 23})
410 tb.ctx, tb.cancel = wctx, wcancel
411 for _, pane := range tb.panes {
412 pane.ctx, pane.cancel = wctx, wcancel
413 }
414 tabs = append(tabs, tb)
415 }
416 sess := &session{sessionCore: sessionCore{id: "manual", name: "work", attachments: map[*attachedClient]struct{}{ac: {}}}, ctx: sctx, cancel: cancel, tabs: tabs}
417 for _, tb := range tabs {
418 publishTiledPaneOwners(sess, tb)
419 }
420 ac.setSession(sess)
421 ac.keys = keys.NewRouter(d.clock, daemonKeyHandler{d: d, ac: ac}, nil)
422 d.sessions[sess.id] = sess
423 if registerCleanup {
424 t.Cleanup(cancel)
425 }
426 return d, sess, ac, sends
427}
428
429func newManualTabSession(t *testing.T, n int) (*Daemon, *session, *attachedClient, chan ports.Frame, []func()) {
430 t.Helper()
431 ptys := make([]ports.PTY, 0, n)
432 releases := make([]func(), 0, n)
433 for range n {
434 p, release := newBlockingPTY(t)
435 ptys = append(ptys, p)
436 releases = append(releases, release)
437 }
438 d, sess, ac, sends := newManualSessionWithPTYs(t, ptys...)
439 return d, sess, ac, sends, releases
440}
441
442func testAttachment(sess *session) *attachedClient {
443 if sess == nil {
444 return nil
445 }
446 attachments := sess.snapshotAttachments()
447 if len(attachments) == 0 {
448 return nil
449 }
450 return attachments[0]
451}
452
453func testAttachmentTabLocked(sess *session) *tab {
454 if sess == nil {
455 return nil
456 }
457 var ac *attachedClient
458 for candidate := range sess.attachments {
459 ac = candidate
460 break
461 }
462 if ac == nil {
463 if len(sess.tabs) == 0 {
464 return nil
465 }
466 return sess.tabs[0]
467 }
468 view := ac.viewSnapshot()
469 for _, tb := range sess.tabs {
470 if domain.TabStableID(tb.stableID) == view.tabID {
471 return tb
472 }
473 }
474 return nil
475}
476
477func testAttachmentTab(sess *session) *tab {
478 if ac := testAttachment(sess); ac != nil {
479 return sess.tabForAttachment(ac)
480 }
481 return sess.firstTab()
482}
483
484func selectTestAttachmentTab(sess *session, idx int) bool {
485 ac := testAttachment(sess)
486 if ac == nil {
487 return idx == 0 && testAttachmentTab(sess) != nil
488 }
489 return sess.switchAttachmentTab(ac, idx)
490}
491
492// selectTestAttachmentTabLocked is used only by fixtures that already hold
493// sess.mu; production navigation always goes through selectAttachmentTab.
494func selectTestAttachmentTabLocked(sess *session, idx int) bool {
495 if sess == nil {
496 return false
497 }
498 var ac *attachedClient
499 for candidate := range sess.attachments {
500 ac = candidate
501 break
502 }
503 if ac == nil {
504 return idx == 0 && len(sess.tabs) > 0
505 }
506 return sess.activateAttachmentViewLocked(ac, idx)
507}
508
509func testAttachmentTabIndexLocked(sess *session) int {
510 if sess == nil {
511 return 0
512 }
513 var ac *attachedClient
514 for candidate := range sess.attachments {
515 ac = candidate
516 break
517 }
518 if ac == nil {
519 return 0
520 }
521 view := ac.viewSnapshot()
522 for i, tb := range sess.tabs {
523 if domain.TabStableID(tb.stableID) == view.tabID {
524 return i
525 }
526 }
527 return 0
528}
529
530func testAttachmentTabIndex(sess *session) int {
531 if sess == nil {
532 return 0
533 }
534 sess.mu.Lock()
535 defer sess.mu.Unlock()
536 return testAttachmentTabIndexLocked(sess)
537}
538
539func testRow(text string) []renderer.Cell {
540 cells := make([]renderer.Cell, 0, len(text))
541 for _, r := range text {
542 cells = append(cells, renderer.Cell{Rune: r, Style: renderer.DefaultStyle()})
543 }
544 return cells
545}
546
547// appendHistoryRow records a row whose logical extent covers it entirely and
548// that ends its logical line — the shape of a line scrolled off a grid it never
549// wrapped out of. Tests that care about wrapped extents build the bound instead.
550func appendHistoryRow(tb testing.TB, history *vt.History, row []renderer.Cell) {
551 tb.Helper()
552 id := max(vt.RowID(1_000_000), history.NextRowID())
553 require.NoError(tb, history.AppendWithID(row, vt.LineBound{End: len(row)}, id))
554}
555
556func TestAppendHistoryRowAllocatesUniqueIDsAfterEviction(t *testing.T) {
557 history := vt.NewHistory(vt.HistoryConfig{MaxRows: 2, ChunkRows: 1})
558 ids := make(map[vt.RowID]struct{})
559 for range 4 {
560 appendHistoryRow(t, history, testRow("row"))
561 view := history.View()
562 id := view.RowID(view.Len() - 1)
563 require.NotContains(t, ids, id)
564 ids[id] = struct{}{}
565 }
566 require.Len(t, ids, 4)
567}
568
569func tabCount(sess *session) int {
570 sess.mu.Lock()
571 defer sess.mu.Unlock()
572 return len(sess.tabs)
573}
574
575// --- handshake --------------------------------------------------------------
576
577func TestRouteRejectsInvalidTerminalSizeBeforeResumeMutation(t *testing.T) {
578 tests := []struct {
579 name string
580 size domain.Size
581 }{
582 {name: "zero columns", size: domain.Size{Cols: 0, Rows: 24}},
583 {name: "zero rows", size: domain.Size{Cols: 80, Rows: 0}},
584 }
585 for _, tt := range tests {
586 t.Run(tt.name, func(t *testing.T) {
587 pty, release := newBlockingPTY(t)
588 defer release()
589 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
590
591 transport := &closeTrackingTransport{}
592 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), transport)
593 require.NoError(t, err)
594 token := ac.resumeToken
595 d.clientGone(sess, ac, transport, false)
596
597 d.mu.Lock()
598 parked := d.parked[token]
599 d.mu.Unlock()
600 require.NotNil(t, parked)
601
602 invalid := helloResumeCapable(ports.IntentResume, "work", token)
603 invalid.Size = tt.size
604 _, _, err = d.route(invalid, &closeTrackingTransport{})
605 var protocolErr *protoErr
606 require.ErrorAs(t, err, &protocolErr)
607 require.Equal(t, ports.ErrInternal, protocolErr.code)
608
609 d.mu.Lock()
610 require.Same(t, parked, d.parked[token], "invalid resume must not consume the parked attachment")
611 d.mu.Unlock()
612 require.Same(t, transport, ac.transport(), "invalid resume must not bind a replacement transport")
613 })
614 }
615}
616
617func waitGroupDone(wg *sync.WaitGroup) <-chan struct{} {
618 done := make(chan struct{})
619 go func() {
620 wg.Wait()
621 close(done)
622 }()
623 return done
624}
625
626func rowText(row []renderer.Cell) string {
627 runes := make([]rune, len(row))
628 for i, c := range row {
629 if c.Continuation {
630 runes[i] = ' '
631 continue
632 }
633 runes[i] = c.Rune
634 }
635 return string(runes)
636}
637
638type notifyClock struct{}
639
640func (notifyClock) Now() time.Time { return time.Now() }
641func (notifyClock) NewTimer(d time.Duration) ports.Timer {
642 if d == debounceInterval {
643 return stubTimer{}
644 }
645 return realTimer{t: time.NewTimer(5 * time.Millisecond)}
646}
647
648type realTimer struct{ t *time.Timer }
649
650func (r realTimer) C() <-chan time.Time { return r.t.C }
651func (r realTimer) Reset(d time.Duration) bool { return r.t.Reset(d) }
652func (r realTimer) Stop() bool { return r.t.Stop() }
653
654func TestHandleHelloDefersFreshOutputUntilWelcome(t *testing.T) {
655 pty, releasePTY := newBlockingPTY(t)
656 defer releasePTY()
657 clock := newCoordinatorMockClock(t, 4)
658 d := newTestDaemon(t, newFactory(t, pty), clock.clock)
659 tr := newWelcomeBlockingTransport(t)
660 done := make(chan struct{})
661 go func() {
662 d.handleHello(tr.tr, mustHello(ports.IntentNew, "welcome-gate", domain.Size{Cols: 80, Rows: 24}))
663 close(done)
664 }()
665
666 awaitTestCompletion(t, tr.welcomeEntered, "timed out waiting for Welcome send")
667 handshakeTimer := awaitHandshakeTimer(t, clock)
668 welcome := awaitFrame(t, tr.sends, ports.MsgWelcome)
669 require.Equal(t, ports.MsgWelcome, welcome.Type)
670 sess := firstSession(d)
671 sess.mu.Lock()
672 ac := sess.snapshotAttachmentsLocked()[0]
673 sess.mu.Unlock()
674 require.NotNil(t, ac, "route must publish the attachment before Welcome returns")
675
676 // Exercise the real producer fan-in after publication. The deadline worker
677 // has completed before asserting the gate, so this is not a timing race.
678 d.invalidateRender(sess, ac, true, "welcome-gate-test")
679 // Attach-time theme setup may have armed an older bulk timer. The fresh
680 // urgent producer replaces it, so advance the newest published deadline.
681 timer := awaitLatestCoordinatorTimer(t, clock)
682 rc := sess.renderCoordinator()
683 rc.mu.Lock()
684 workerDone := rc.normalLane.token.done
685 rc.mu.Unlock()
686 require.NotNil(t, workerDone)
687 timer.ch <- time.Time{}
688 awaitTestCompletion(t, workerDone, "coordinator deadline worker did not complete")
689 requireNoCoordinatorOutputFrame(t, tr.sends)
690
691 tr.release()
692 output := awaitFrame(t, tr.sends, ports.MsgOutput)
693 first, err := ports.UnmarshalOutput(output.Payload)
694 require.NoError(t, err)
695 require.Zero(t, first.Base, "the first post-Welcome frame is full")
696 tr.finish()
697 awaitTestCompletion(t, done, "timed out waiting for Welcome handler completion")
698 awaitTestCompletion(t, handshakeTimer.stopped, "handshake timer was not canceled after publication")
699 requireNoCoordinatorOutputFrame(t, tr.sends)
700}
701
702// TestServeReturnsDespiteWedgedClientOnShutdown: killSession's teardown must
703// not be gated behind the Detached notice. The client transport wedges every
704// MsgDetached send until the transport is closed (mirroring a full kernel
705// send buffer: only Close fails the in-flight write). Serve must still tear
706
707func mustOutputData(t *testing.T, sends chan ports.Frame) []byte {
708 t.Helper()
709 f := awaitFrame(t, sends, ports.MsgOutput)
710 out, err := ports.UnmarshalOutput(f.Payload)
711 require.NoError(t, err)
712 return out.Data
713}