1//go:build linux
2
3package app
4
5import (
6 "context"
7 "errors"
8 "io"
9 "os"
10 "slices"
11 "sync"
12 "testing"
13 "time"
14
15 "github.com/stretchr/testify/mock"
16 "github.com/stretchr/testify/require"
17
18 "github.com/bnema/vev/internal/adapters/brokeripc"
19 "github.com/bnema/vev/internal/adapters/clock"
20 "github.com/bnema/vev/internal/domain"
21 "github.com/bnema/vev/internal/ports"
22 portsmocks "github.com/bnema/vev/internal/ports/mocks"
23 "github.com/bnema/vev/internal/protocol"
24 "github.com/bnema/vev/internal/protocol/catalogue"
25 "github.com/bnema/vev/internal/usecase/client"
26)
27
28// Terminal composition tests (Plan 001 P7.4a).
29//
30// runAttach is the ordinary terminal entry point. These tests drive it through
31// the composition-root seams only: connectProductionClientBroker supplies one
32// scripted broker connection and terminalForAttach supplies a recording
33// terminal, so the whole client supervisor runs without a production broker
34// process, a daemon dialer, or a controlling terminal. Every assertion is about
35// the composition contract: which broker stream the parsed CLI target resolves
36// to, that a refused target opens nothing, and that an admitted attachment
37// commits a real session publication to the terminal.
38
39const (
40 terminalCompositionEpoch = ports.BrokerEpoch(41)
41 terminalCompositionGeneration = domain.RemoteGeneration(3)
42)
43
44func terminalCompositionPolicy() ports.BrokerPolicy {
45 return ports.BrokerPolicy{
46 ProtocolVersion: protocol.Version,
47 CatalogSchemaVersion: catalogue.RemoteCatalogSchemaVersion,
48 EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned,
49 Transport: "unix-mux",
50 Trust: "same-user",
51 Launch: "explicit",
52 Isolation: "per-user",
53 }
54}
55
56func terminalCompositionRegistration(endpoint string) domain.RemoteRegistration {
57 var incarnation [16]byte
58 for i := range incarnation {
59 incarnation[i] = byte(i + 1)
60 }
61 return domain.RemoteRegistration{Endpoint: endpoint, Incarnation: incarnation, Generation: terminalCompositionGeneration}
62}
63
64// terminalCompositionSnapshot publishes the local daemon plus the given remote
65// endpoints, each carrying exactly the sessions named.
66func terminalCompositionSnapshot(localSessions []string, remotes map[string][]string) ports.BrokerSnapshot {
67 policy := terminalCompositionPolicy()
68 observations := []ports.BrokerDaemonObservation{{
69 Local: true,
70 DisplayOrigin: "local",
71 Policy: policy,
72 Identity: "local-daemon",
73 Incarnation: ports.BrokerDaemonIncarnation{9},
74 ProtocolVersion: protocol.Version,
75 Availability: domain.RemoteAvailabilityReachable,
76 LastSuccess: time.Now(),
77 InventoryKnown: true,
78 Sessions: terminalCompositionSessions(localSessions),
79 }}
80 endpoints := make([]string, 0, len(remotes))
81 for endpoint := range remotes {
82 endpoints = append(endpoints, endpoint)
83 }
84 sortStrings(endpoints)
85 for _, endpoint := range endpoints {
86 observations = append(observations, ports.BrokerDaemonObservation{
87 Endpoint: endpoint,
88 DisplayOrigin: endpoint,
89 Policy: policy,
90 Registration: terminalCompositionRegistration(endpoint),
91 Identity: ports.BrokerDaemonIdentity("identity-" + endpoint),
92 Incarnation: ports.BrokerDaemonIncarnation{7},
93 ProtocolVersion: protocol.Version,
94 Availability: domain.RemoteAvailabilityReachable,
95 LastSuccess: time.Now(),
96 InventoryKnown: true,
97 Sessions: terminalCompositionSessions(remotes[endpoint]),
98 })
99 }
100 return ports.BrokerSnapshot{Epoch: terminalCompositionEpoch, Revision: 1, Daemons: observations}
101}
102
103func terminalCompositionSessions(names []string) []catalogue.RemoteCatalogSession {
104 sessions := make([]catalogue.RemoteCatalogSession, 0, len(names))
105 for i, name := range names {
106 var lifecycle domain.SessionLifecycleID
107 lifecycle[0] = byte(i + 3)
108 lifecycle[1] = 0x5a
109 sessions = append(sessions, catalogue.RemoteCatalogSession{
110 LifecycleID: lifecycle, Name: name, State: catalogue.RemoteCatalogSessionUp,
111 })
112 }
113 return sessions
114}
115
116func sortStrings(values []string) {
117 for i := 1; i < len(values); i++ {
118 for j := i; j > 0 && values[j] < values[j-1]; j-- {
119 values[j], values[j-1] = values[j-1], values[j]
120 }
121 }
122}
123
124// terminalCompositionService is the scripted broker service one terminal client
125// adopts. It records every opened stream request, serves one committed
126// publication, and answers each admitted attachment with a Welcome and one
127// initial full session frame so the supervisor reaches its attached
128// presentation.
129type terminalCompositionService struct {
130 mu sync.Mutex
131 snapshot ports.BrokerSnapshot
132 // daemonSessions, when set, is what the local daemon owns regardless of
133 // what the broker catalogue has observed.
134 daemonSessions []string
135 closed bool
136 next ports.BrokerStreamID
137 opened []ports.BrokerOpenStreamRequest
138 streams []*terminalCompositionStream
139 done chan struct{}
140 changed chan struct{}
141}
142
143func newTerminalCompositionService(snapshot ports.BrokerSnapshot) *terminalCompositionService {
144 return &terminalCompositionService{
145 snapshot: snapshot,
146 done: make(chan struct{}),
147 changed: make(chan struct{}, 1),
148 }
149}
150
151func (s *terminalCompositionService) ConnectionID() ports.BrokerConnectionID {
152 return ports.BrokerConnectionID{4}
153}
154
155func (s *terminalCompositionService) NextStreamID() (ports.BrokerStreamID, error) {
156 s.mu.Lock()
157 defer s.mu.Unlock()
158 if s.closed {
159 return 0, ports.BrokerAdmissionClosed
160 }
161 s.next++
162 return s.next, nil
163}
164
165func (s *terminalCompositionService) Done() <-chan struct{} { return s.done }
166func (s *terminalCompositionService) Err() error { return nil }
167
168func (s *terminalCompositionService) Snapshot() ports.BrokerSnapshot { return s.snapshot.Clone() }
169
170func (s *terminalCompositionService) Subscribe() (ports.BrokerSubscription, error) {
171 return terminalCompositionSubscription{changed: s.changed}, nil
172}
173
174func (s *terminalCompositionService) SubscribePreview(ports.BrokerPreviewRequest) (ports.BrokerPreviewSubscription, error) {
175 return nil, errors.New("terminal composition service does not support preview")
176}
177
178func (s *terminalCompositionService) OpenStream(_ context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
179 s.mu.Lock()
180 if s.closed {
181 s.mu.Unlock()
182 return nil, ports.BrokerAdmissionClosed
183 }
184 s.opened = append(s.opened, request)
185 sessions := terminalCompositionDaemonSessions(s.snapshot, request)
186 if request.Local && s.daemonSessions != nil {
187 sessions = terminalCompositionSessions(s.daemonSessions)
188 }
189 stream := newTerminalCompositionStream(request, sessions)
190 s.streams = append(s.streams, stream)
191 s.mu.Unlock()
192 return stream, nil
193}
194
195func (s *terminalCompositionService) CloseStream(ports.BrokerConnectionID, ports.BrokerStreamID) error {
196 return nil
197}
198
199func (s *terminalCompositionService) AddHost(context.Context, string, ports.BrokerPolicy) (domain.RemoteRegistration, error) {
200 return domain.RemoteRegistration{}, nil
201}
202
203func (s *terminalCompositionService) RemoveHost(context.Context, domain.RemoteRegistration) (bool, error) {
204 return false, nil
205}
206
207func (s *terminalCompositionService) UpdateHostPolicy(context.Context, domain.RemoteRegistration, ports.BrokerPolicy) (domain.RemoteRegistration, error) {
208 return domain.RemoteRegistration{}, nil
209}
210
211func (s *terminalCompositionService) RequestReconcile(string) {}
212
213func (s *terminalCompositionService) Close() error {
214 s.mu.Lock()
215 defer s.mu.Unlock()
216 if !s.closed {
217 s.closed = true
218 close(s.done)
219 }
220 return nil
221}
222
223// openedRequests returns the attachment streams opened; the preflight's
224// control listings are reported by openedControls.
225func (s *terminalCompositionService) openedRequests() []ports.BrokerOpenStreamRequest {
226 return s.openedWithPurpose(ports.BrokerStreamAttachment)
227}
228
229func (s *terminalCompositionService) openedControls() []ports.BrokerOpenStreamRequest {
230 return s.openedWithPurpose(ports.BrokerStreamControl)
231}
232
233func (s *terminalCompositionService) openedWithPurpose(purpose ports.BrokerStreamPurpose) []ports.BrokerOpenStreamRequest {
234 s.mu.Lock()
235 defer s.mu.Unlock()
236 var out []ports.BrokerOpenStreamRequest
237 for _, request := range s.opened {
238 if request.Purpose == purpose {
239 out = append(out, request)
240 }
241 }
242 return out
243}
244
245// streamHello returns the Hello the supervisor sent on the single admitted
246// stream, proving the composition carried the CLI intent through to the typed
247// session handshake.
248func (s *terminalCompositionService) streamHello(t *testing.T) protocol.Hello {
249 t.Helper()
250 s.mu.Lock()
251 var streams []*terminalCompositionStream
252 for _, stream := range s.streams {
253 if stream.request.Purpose == ports.BrokerStreamAttachment {
254 streams = append(streams, stream)
255 }
256 }
257 s.mu.Unlock()
258 require.Len(t, streams, 1)
259 for _, message := range streams[0].sentMessages() {
260 if hello, ok := message.(protocol.Hello); ok {
261 return hello
262 }
263 }
264 t.Fatal("the composed client never sent a session Hello")
265 return protocol.Hello{}
266}
267
268type terminalCompositionSubscription struct{ changed chan struct{} }
269
270func (s terminalCompositionSubscription) Changed() <-chan struct{} { return s.changed }
271func (s terminalCompositionSubscription) Close() {}
272
273// terminalCompositionStream answers one admitted attachment with a Welcome and
274// one initial full session frame carrying real terminal bytes, so a test can
275// prove the committed publication reached the terminal.
276type terminalCompositionStream struct {
277 request ports.BrokerOpenStreamRequest
278
279 mu sync.Mutex
280 sent []protocol.ClientMessage
281 queue []protocol.ServerMessage
282 closed bool
283 done chan struct{}
284}
285
286// terminalCompositionDaemonSessions is the session list the scripted
287// destination daemon owns: the sessions its observation publishes.
288func terminalCompositionDaemonSessions(snapshot ports.BrokerSnapshot, request ports.BrokerOpenStreamRequest) []catalogue.RemoteCatalogSession {
289 for _, observation := range snapshot.Daemons {
290 if observation.Local == request.Local && (request.Local || observation.Endpoint == request.Endpoint) {
291 return observation.Sessions
292 }
293 }
294 return nil
295}
296
297func newTerminalCompositionStream(request ports.BrokerOpenStreamRequest, sessions []catalogue.RemoteCatalogSession) *terminalCompositionStream {
298 stream := &terminalCompositionStream{request: request, done: make(chan struct{})}
299 if request.Purpose == ports.BrokerStreamControl {
300 // The scripted daemon answers a List with its own sessions.
301 infos := make([]protocol.SessionInfo, 0, len(sessions))
302 for _, session := range sessions {
303 infos = append(infos, protocol.SessionInfo{Name: session.Name})
304 }
305 stream.queue = append(stream.queue, protocol.Sessions{Sessions: infos})
306 return stream
307 }
308 if request.Admission == ports.BrokerAdmissionAttachNamed && !slices.ContainsFunc(sessions, func(session catalogue.RemoteCatalogSession) bool { return session.Name == request.Name }) {
309 // Like the real daemon, an unknown name is refused, never created.
310 stream.queue = append(stream.queue, protocol.ErrorMsg{Code: protocol.ErrNoSuchSession, Text: "no such resumable session: " + request.Name})
311 return stream
312 }
313 target := terminalCompositionCommittedTarget(request)
314 stream.queue = append(stream.queue, protocol.Welcome{
315 SessionID: "sess", SessionName: request.Name,
316 CommittedIdentity: &protocol.CommittedRouteIdentity{Target: target},
317 })
318 stream.queue = append(stream.queue, protocol.Output{
319 Epoch: 1, Full: true, New: 1,
320 Size: domain.Size{Cols: 80, Rows: 24},
321 Context: &protocol.ViewContext{
322 Publication: 1,
323 Route: protocol.CommittedRouteIdentity{Target: target},
324 TabID: domain.TabStableID("tab-1"),
325 FocusedPaneID: domain.PaneStableID("pane-1"),
326 },
327 Data: []byte("terminal-composition-ready\r\n"),
328 })
329 return stream
330}
331
332// terminalCompositionCommittedTarget is the exact identity the scripted daemon
333// commits for one admitted request. An ephemeral creation has no requested name,
334// so the scripted daemon allocates one exactly as the real daemon does.
335func terminalCompositionCommittedTarget(request ports.BrokerOpenStreamRequest) protocol.ExactSessionTarget {
336 if request.Admission == ports.BrokerAdmissionExact {
337 return request.Target
338 }
339 name := request.Name
340 if name == "" {
341 name = "1"
342 }
343 var lifecycle domain.SessionLifecycleID
344 lifecycle[0] = 0x11
345 return protocol.ExactSessionTarget{LifecycleID: lifecycle, SessionName: name}
346}
347
348func (s *terminalCompositionStream) SendClient(message protocol.ClientMessage) error {
349 s.mu.Lock()
350 defer s.mu.Unlock()
351 s.sent = append(s.sent, message)
352 return nil
353}
354
355func (s *terminalCompositionStream) ReceiveServer() (protocol.ServerMessage, error) {
356 s.mu.Lock()
357 if len(s.queue) != 0 {
358 message := s.queue[0]
359 s.queue = s.queue[1:]
360 s.mu.Unlock()
361 return message, nil
362 }
363 s.mu.Unlock()
364 <-s.done
365 return nil, io.EOF
366}
367
368func (s *terminalCompositionStream) Capabilities() protocol.ConnectionCapabilities {
369 return protocol.ConnectionCapabilities{}
370}
371
372func (s *terminalCompositionStream) LinkState() ports.LinkState { return ports.LinkStateConnected }
373func (s *terminalCompositionStream) LinkEvents() <-chan ports.LinkEvent {
374 return nil
375}
376
377func (s *terminalCompositionStream) Done() <-chan struct{} { return s.done }
378func (s *terminalCompositionStream) Err() error { return nil }
379
380func (s *terminalCompositionStream) Close() error {
381 s.mu.Lock()
382 defer s.mu.Unlock()
383 if !s.closed {
384 s.closed = true
385 close(s.done)
386 }
387 return nil
388}
389
390func (s *terminalCompositionStream) sentMessages() []protocol.ClientMessage {
391 s.mu.Lock()
392 defer s.mu.Unlock()
393 return append([]protocol.ClientMessage(nil), s.sent...)
394}
395
396// terminalCompositionTerminal is the terminal double: it records written bytes,
397// flushes, raw-mode entries/restores, and every UI publication so a test can
398// prove a committed attachment published a real frame.
399type terminalCompositionTerminal struct {
400 mu sync.Mutex
401 out []byte
402 flushes int
403 restores int
404 statuses []ports.UIPresentationStatus
405 changes chan struct{}
406 input *terminalCompositionInput
407}
408
409func newTerminalCompositionTerminal() *terminalCompositionTerminal {
410 return &terminalCompositionTerminal{changes: make(chan struct{}), input: newTerminalCompositionInput()}
411}
412
413// terminalCompositionInput is the supervisor's single terminal reader. It
414// yields bytes only when a test pushes them and reaches EOF when it closes, so
415// a client with no pushed input stays live until the test ends the run.
416type terminalCompositionInput struct {
417 chunks chan []byte
418}
419
420func newTerminalCompositionInput() *terminalCompositionInput {
421 return &terminalCompositionInput{chunks: make(chan []byte)}
422}
423
424func (i *terminalCompositionInput) Read(buffer []byte) (int, error) {
425 chunk, ok := <-i.chunks
426 if !ok {
427 return 0, io.EOF
428 }
429 return copy(buffer, chunk), nil
430}
431
432func (t *terminalCompositionTerminal) EnableKittyKeyboard(int) error { return nil }
433
434func (t *terminalCompositionTerminal) EnterRaw() (func() error, error) {
435 return func() error {
436 t.mu.Lock()
437 t.restores++
438 t.mu.Unlock()
439 return nil
440 }, nil
441}
442
443func (t *terminalCompositionTerminal) Geometry() (domain.Geometry, error) {
444 return domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}}, nil
445}
446
447func (t *terminalCompositionTerminal) ResizeEvents() <-chan domain.Geometry { return nil }
448
449// In blocks until the run ends: the supervisor owns exactly one terminal read
450// lifetime, and an immediately-empty reader would look like a terminal EOF and
451// end an otherwise live client.
452func (t *terminalCompositionTerminal) In() io.Reader { return t.input }
453
454func (t *terminalCompositionTerminal) Out() io.Writer { return t }
455
456func (t *terminalCompositionTerminal) Write(data []byte) (int, error) {
457 t.mu.Lock()
458 defer t.mu.Unlock()
459 t.out = append(t.out, data...)
460 return len(data), nil
461}
462
463func (t *terminalCompositionTerminal) Flush() error {
464 t.mu.Lock()
465 defer t.mu.Unlock()
466 t.flushes++
467 return nil
468}
469
470func (t *terminalCompositionTerminal) BeginOutput(ports.UIContext) {}
471func (t *terminalCompositionTerminal) EndOutput(bool) {}
472
473func (t *terminalCompositionTerminal) PublishContext(context ports.UIContext) error {
474 t.mu.Lock()
475 defer t.mu.Unlock()
476 t.statuses = append(t.statuses, context.Status)
477 return nil
478}
479
480func (t *terminalCompositionTerminal) written() string {
481 t.mu.Lock()
482 defer t.mu.Unlock()
483 return string(t.out)
484}
485
486func (t *terminalCompositionTerminal) publishes() []ports.UIPresentationStatus {
487 t.mu.Lock()
488 defer t.mu.Unlock()
489 return append([]ports.UIPresentationStatus(nil), t.statuses...)
490}
491
492// startTerminalComposition drives the real runAttach entry point over one
493// scripted broker service and recording terminal. It substitutes only the two
494// composition seams (connect-or-spawn and the terminal) plus the callback seam,
495// so every assertion holds for the production terminal path itself and not for
496// a parallel test harness.
497type terminalCompositionRun struct {
498 service *terminalCompositionService
499 terminal *terminalCompositionTerminal
500 states chan client.State
501 notices chan client.LifecycleNotice
502 failures chan error
503 finished chan struct{}
504 runMu sync.Mutex
505 runErr error
506 cancel context.CancelFunc
507}
508
509// wait returns the run's terminal error. It is idempotent: every caller waits on
510// the same closed channel, so a failure observed by one helper never consumes the
511// result another helper needs.
512func (r *terminalCompositionRun) wait(t *testing.T) error {
513 t.Helper()
514 select {
515 case <-r.finished:
516 case <-time.After(brokerTestWait):
517 t.Fatal("terminal client did not stop")
518 }
519 r.runMu.Lock()
520 defer r.runMu.Unlock()
521 return r.runErr
522}
523
524func startTerminalComposition(t *testing.T, intent uint8, name, remoteTarget string, snapshot ports.BrokerSnapshot) *terminalCompositionRun {
525 t.Helper()
526 return startTerminalCompositionWithDaemon(t, intent, name, remoteTarget, snapshot, nil)
527}
528
529// startTerminalCompositionWithDaemon is startTerminalComposition whose local
530// daemon owns daemonSessions even when the catalogue has not observed them.
531func startTerminalCompositionWithDaemon(t *testing.T, intent uint8, name, remoteTarget string, snapshot ports.BrokerSnapshot, daemonSessions []string) *terminalCompositionRun {
532 t.Helper()
533 // The ordinary terminal path reads the nested-session environment; a test
534 // drives the broker composition with it unset.
535 t.Setenv("VEV", "")
536 service := newTerminalCompositionService(snapshot)
537 service.daemonSessions = daemonSessions
538 terminal := newTerminalCompositionTerminal()
539 previousConnect := connectProductionClientBroker
540 previousTerminal := terminalForAttach
541 previousCallbacks := terminalBrokerCallbacks
542 connectProductionClientBroker = func(context.Context) (ports.BrokerService, error) {
543 return service, nil
544 }
545 terminalForAttach = func() ports.Terminal { return terminal }
546
547 run := &terminalCompositionRun{
548 service: service,
549 terminal: terminal,
550 states: make(chan client.State, 64),
551 notices: make(chan client.LifecycleNotice, 16),
552 failures: make(chan error, 16),
553 finished: make(chan struct{}),
554 }
555 terminalBrokerCallbacks = func() terminalBrokerSeams {
556 return terminalBrokerSeams{
557 OnState: func(state client.State) {
558 select {
559 case run.states <- state:
560 default:
561 }
562 },
563 OnLifecycle: func(notice client.LifecycleNotice) {
564 select {
565 case run.notices <- notice:
566 default:
567 }
568 },
569 OnFailure: func(err error) {
570 select {
571 case run.failures <- err:
572 default:
573 }
574 },
575 }
576 }
577 t.Cleanup(func() {
578 connectProductionClientBroker = previousConnect
579 terminalForAttach = previousTerminal
580 terminalBrokerCallbacks = previousCallbacks
581 })
582
583 ctx, cancel := context.WithCancel(context.Background())
584 run.cancel = cancel
585 go func() {
586 err := runAttach(ctx, intent, name, remoteTarget)
587 run.runMu.Lock()
588 run.runErr = err
589 run.runMu.Unlock()
590 close(run.finished)
591 }()
592 t.Cleanup(func() {
593 cancel()
594 select {
595 case <-run.finished:
596 case <-time.After(brokerTestWait):
597 t.Error("terminal client did not stop")
598 }
599 })
600 return run
601}
602
603// awaitTerminalCompositionAttachment waits for the supervisor to present an
604// attached state and returns it.
605func awaitTerminalCompositionAttachment(t *testing.T, run *terminalCompositionRun) client.State {
606 t.Helper()
607 deadline := time.NewTimer(brokerTestWait)
608 defer deadline.Stop()
609 for {
610 select {
611 case state := <-run.states:
612 if state.Presentation == client.PresentAttached {
613 return state
614 }
615 case <-run.finished:
616 t.Fatalf("terminal client ended before attaching: %v", run.wait(t))
617 case <-deadline.C:
618 t.Fatal("terminal client never attached")
619 }
620 }
621}
622
623// awaitTerminalCompositionNotice waits for one bounded lifecycle notice.
624func awaitTerminalCompositionNotice(t *testing.T, run *terminalCompositionRun) client.LifecycleNotice {
625 t.Helper()
626 select {
627 case notice := <-run.notices:
628 return notice
629 case <-time.After(brokerTestWait):
630 t.Fatal("terminal client reported no lifecycle notice")
631 return client.LifecycleNotice{}
632 }
633}
634
635func TestTerminalNavigationTranslationClosesTheIntent(t *testing.T) {
636 tests := []struct {
637 name string
638 intent uint8
639 session string
640 remote string
641 wantKind client.InitialNavigationKind
642 wantLocal bool
643 wantName string
644 wantResolver bool
645 wantError bool
646 }{
647 {name: "no arguments creates a local ephemeral session", intent: protocol.IntentEphemeral, wantKind: client.InitialNavigationCreateEphemeral, wantLocal: true},
648 {name: "new creates a local named session", intent: protocol.IntentNew, session: "scratch", wantKind: client.InitialNavigationCreateNamed, wantLocal: true, wantName: "scratch"},
649 {name: "attach names a local session for the daemon to resolve", intent: protocol.IntentAttach, session: "work", wantKind: client.InitialNavigationAttachNamed, wantLocal: true, wantName: "work"},
650 {name: "remote attach resolves the remote registration", intent: protocol.IntentAttach, session: "work", remote: "user@example.test", wantResolver: true},
651 {name: "remote new resolves a named remote creation", intent: protocol.IntentNew, session: "work", remote: "user@example.test", wantResolver: true},
652 {name: "remote ephemeral attach resolves a remote creation", intent: protocol.IntentEphemeral, remote: "user@example.test", wantResolver: true},
653 {name: "invalid local name is refused", intent: protocol.IntentNew, session: "bad name", wantError: true},
654 {name: "invalid remote target is refused", intent: protocol.IntentAttach, session: "work", remote: "bad target", wantError: true},
655 }
656 for _, tc := range tests {
657 t.Run(tc.name, func(t *testing.T) {
658 navigation, resolver, err := terminalBrokerNavigation(tc.intent, tc.session, tc.remote)
659 if tc.wantError {
660 require.Error(t, err)
661 require.Nil(t, resolver)
662 require.Equal(t, client.InitialNavigation{}, navigation)
663 return
664 }
665 require.NoError(t, err)
666 if tc.wantResolver {
667 require.NotNil(t, resolver)
668 require.Equal(t, client.InitialNavigation{}, navigation)
669 return
670 }
671 require.Nil(t, resolver)
672 require.Equal(t, tc.wantKind, navigation.Kind)
673 require.Equal(t, tc.wantLocal, navigation.Destination.Local)
674 require.Equal(t, tc.wantName, navigation.Name)
675 require.NoError(t, navigation.Validate())
676 })
677 }
678}
679
680// TestTerminalCompositionOpensTheExactBrokerStream pins the ordinary terminal
681// entry point to the broker: each parsed CLI target resolves to exactly one
682// admission on one broker logical stream, and the composition dials no daemon.
683func TestTerminalCompositionOpensTheExactBrokerStream(t *testing.T) {
684 localSnapshot := func() ports.BrokerSnapshot {
685 return terminalCompositionSnapshot([]string{"work"}, nil)
686 }
687 remoteSnapshot := func() ports.BrokerSnapshot {
688 return terminalCompositionSnapshot(nil, map[string][]string{"user@example.test": {"work"}})
689 }
690 tests := []struct {
691 name string
692 intent uint8
693 session string
694 remote string
695 snapshot ports.BrokerSnapshot
696 daemonSessions []string
697 wantAdmission ports.BrokerStreamAdmission
698 wantLocal bool
699 wantName string
700 wantHello uint8
701 }{
702 {name: "no arguments creates a local ephemeral session", intent: protocol.IntentEphemeral, snapshot: localSnapshot(), wantAdmission: ports.BrokerAdmissionCreateEphemeral, wantLocal: true, wantHello: protocol.IntentEphemeral},
703 {name: "new creates a local named session", intent: protocol.IntentNew, session: "scratch", snapshot: localSnapshot(), wantAdmission: ports.BrokerAdmissionCreateNamed, wantLocal: true, wantName: "scratch", wantHello: protocol.IntentNew},
704 {name: "attach names the local session", intent: protocol.IntentAttach, session: "work", snapshot: localSnapshot(), wantAdmission: ports.BrokerAdmissionAttachNamed, wantLocal: true, wantName: "work", wantHello: protocol.IntentAttach},
705 {name: "remote attach names the remote session", intent: protocol.IntentAttach, session: "work", remote: "user@example.test", snapshot: remoteSnapshot(), wantAdmission: ports.BrokerAdmissionAttachNamed, wantName: "work", wantHello: protocol.IntentAttach},
706 {name: "attach reaches a session the catalogue has not observed yet", intent: protocol.IntentAttach, session: "work", snapshot: terminalCompositionUnobservedLocal(), daemonSessions: []string{"work"}, wantAdmission: ports.BrokerAdmissionAttachNamed, wantLocal: true, wantName: "work", wantHello: protocol.IntentAttach},
707 }
708 for _, tc := range tests {
709 t.Run(tc.name, func(t *testing.T) {
710 run := startTerminalCompositionWithDaemon(t, tc.intent, tc.session, tc.remote, tc.snapshot, tc.daemonSessions)
711 awaitTerminalCompositionAttachment(t, run)
712 opened := run.service.openedRequests()
713 require.Len(t, opened, 1, "one CLI target opens exactly one broker stream")
714 request := opened[0]
715 require.NoError(t, request.Validate())
716 require.Equal(t, terminalCompositionEpoch, request.Epoch)
717 require.Equal(t, ports.BrokerStreamAttachment, request.Purpose)
718 require.Equal(t, tc.wantAdmission, request.Admission)
719 require.Equal(t, tc.wantLocal, request.Local)
720 require.Equal(t, tc.wantName, request.Name)
721 require.Equal(t, ports.BrokerConnectionID{4}, request.Connection)
722 require.NotZero(t, request.Stream)
723 require.NotEmpty(t, request.Policy.Transport)
724 require.Equal(t, ports.BrokerDaemonStartIfNeeded, request.StartMode)
725 if !tc.wantLocal {
726 require.Equal(t, "user@example.test", request.Endpoint)
727 require.Equal(t, terminalCompositionRegistration("user@example.test"), request.Registration)
728 }
729 if tc.wantAdmission == ports.BrokerAdmissionExact {
730 require.NoError(t, request.Target.Validate())
731 require.Equal(t, tc.session, request.Target.SessionName)
732 }
733 // The same intent reaches the typed session handshake, so the
734 // broker admission and the daemon Hello can never disagree.
735 hello := run.service.streamHello(t)
736 require.Equal(t, tc.wantHello, hello.Intent)
737 require.Equal(t, tc.session, hello.Name)
738 require.Equal(t, tc.remote != "", hello.Remote)
739 if tc.wantAdmission == ports.BrokerAdmissionExact {
740 require.NotNil(t, hello.ExactTarget)
741 require.Equal(t, request.Target, *hello.ExactTarget)
742 }
743 if tc.wantAdmission == ports.BrokerAdmissionAttachNamed {
744 require.Nil(t, hello.ExactTarget, "a named attach leaves identity to the daemon")
745 controls := run.service.openedControls()
746 require.Len(t, controls, 1, "the preflight asks the daemon once")
747 require.Equal(t, ports.BrokerDaemonStartIfNeeded, controls[0].StartMode, "the preflight may start a stopped daemon")
748 }
749 })
750 }
751}
752
753// terminalCompositionUnobservedLocal is the publication a freshly started
754// broker commits right after `kill --all`: the local daemon is known but its
755// sessions are not observed yet, while the daemon itself restores them.
756func terminalCompositionUnobservedLocal() ports.BrokerSnapshot {
757 snapshot := terminalCompositionSnapshot(nil, nil)
758 snapshot.Daemons[0].InventoryKnown = false
759 return snapshot
760}
761
762// TestTerminalCompositionPublishesRealSessionOutput proves an admitted
763// attachment is a committed publication: the daemon's initial full frame reached
764// the terminal through a committed UI output transaction, and the supervisor
765// presented attached only after that commit.
766func TestTerminalCompositionPublishesRealSessionOutput(t *testing.T) {
767 run := startTerminalComposition(t, protocol.IntentEphemeral, "", "", terminalCompositionSnapshot([]string{"work"}, nil))
768 awaitTerminalCompositionAttachment(t, run)
769 require.Contains(t, run.terminal.written(), "terminal-composition-ready")
770 // The initial publication is committed before the attached marker, so its
771 // recorded boundary status is the pre-attachment Connecting presentation.
772 require.Contains(t, run.terminal.publishes(), ports.UIStatusConnecting)
773 // The attached presentation is published just after the attached state, on
774 // the attachment goroutine, so wait for it instead of racing it.
775 require.Eventually(t, func() bool {
776 return slices.Contains(run.terminal.publishes(), ports.UIStatusAttached)
777 }, brokerTestWait, 5*time.Millisecond)
778 _, _, restores := run.terminalCounts()
779 require.Zero(t, restores)
780}
781
782func (r *terminalCompositionRun) terminalCounts() (int, int, int) {
783 r.terminal.mu.Lock()
784 defer r.terminal.mu.Unlock()
785 return len(r.terminal.out), r.terminal.flushes, r.terminal.restores
786}
787
788// TestTerminalCompositionRefusedTargetOpensNothingAndKeepsThePicker pins the
789// failure contract: a target the committed catalogue does not carry is a local
790// refusal. No stream is opened, the supervisor reports a bounded
791// selection-unavailable notice, and the process stays in the picker instead of
792// attaching a fallback or creating implicitly.
793func TestTerminalCompositionRefusedTargetOpensNothingAndKeepsThePicker(t *testing.T) {
794 tests := []struct {
795 name string
796 intent uint8
797 session string
798 remote string
799 snapshot ports.BrokerSnapshot
800 }{
801 {name: "local daemon absent", intent: protocol.IntentEphemeral, snapshot: ports.BrokerSnapshot{Epoch: terminalCompositionEpoch, Revision: 1}},
802 {name: "unconfigured remote host", intent: protocol.IntentAttach, session: "work", remote: "user@elsewhere.test", snapshot: terminalCompositionSnapshot([]string{"work"}, nil)},
803 }
804 for _, tc := range tests {
805 t.Run(tc.name, func(t *testing.T) {
806 run := startTerminalComposition(t, tc.intent, tc.session, tc.remote, tc.snapshot)
807 require.Equal(t, client.LifecycleNoticeSelectionUnavailable, awaitTerminalCompositionNotice(t, run).Kind)
808 // A local refusal is surfaced once, with the bounded text the supervisor
809 // classifies; the resolver's own sentinel is deliberately not propagated
810 // past that taxonomy, so only the presence and the classification are
811 // asserted.
812 select {
813 case failure := <-run.failures:
814 require.Error(t, failure)
815 case <-time.After(brokerTestWait):
816 t.Fatal("the refused target reported no failure")
817 }
818 require.Empty(t, run.service.openedRequests(), "a refused target never opens a stream")
819 // The pending navigation presents Connecting, never the picker
820 // first; the refusal then returns to the picker and never attaches.
821 last := client.PresentConnecting
822 for {
823 select {
824 case state := <-run.states:
825 require.NotEqual(t, client.PresentAttached, state.Presentation, "a refusal never attaches")
826 last = state.Presentation
827 case <-time.After(100 * time.Millisecond):
828 require.Equal(t, client.PresentPicker, last, "a refusal ends on the picker")
829 return
830 }
831 }
832 })
833 }
834}
835
836// TestTerminalCompositionDaemonRefusesMissingName pins that existence is the
837// daemon's decision: a declined (non-interactive) create prompt keeps the
838// named attach, the daemon answers ErrNoSuchSession, and the client reports
839// that refusal without attaching or creating.
840func TestTerminalCompositionDaemonRefusesMissingName(t *testing.T) {
841 tests := []struct {
842 name string
843 remote string
844 snapshot ports.BrokerSnapshot
845 }{
846 {name: "local", snapshot: terminalCompositionSnapshot([]string{"work"}, nil)},
847 {name: "remote", remote: "user@example.test", snapshot: terminalCompositionSnapshot(nil, map[string][]string{"user@example.test": {"work"}})},
848 }
849 for _, tc := range tests {
850 t.Run(tc.name, func(t *testing.T) {
851 withAttachInteractiveConsole(t, false)
852 run := startTerminalComposition(t, protocol.IntentAttach, "gone", tc.remote, tc.snapshot)
853 require.Equal(t, client.LifecycleNoticeDestinationFailed, awaitTerminalCompositionNotice(t, run).Kind)
854 select {
855 case failure := <-run.failures:
856 var protocolErr *client.ProtocolError
857 require.ErrorAs(t, failure, &protocolErr)
858 require.Equal(t, protocol.ErrNoSuchSession, protocolErr.Code)
859 case <-time.After(brokerTestWait):
860 t.Fatal("the daemon refusal was not reported")
861 }
862 opened := run.service.openedRequests()
863 require.Len(t, opened, 1)
864 require.Equal(t, ports.BrokerAdmissionAttachNamed, opened[0].Admission, "a missing name is never created implicitly")
865 })
866 }
867}
868
869// TestTerminalCompositionClosesRawModeOnTerminalEnd pins that the shared
870// composition restores the terminal exactly once when the run ends, however the
871// attachment settled.
872func TestTerminalCompositionClosesRawModeOnTerminalEnd(t *testing.T) {
873 run := startTerminalComposition(t, protocol.IntentEphemeral, "", "", terminalCompositionSnapshot([]string{"work"}, nil))
874 awaitTerminalCompositionAttachment(t, run)
875 run.cancel()
876 require.NoError(t, ignoreContextCancellation(run.wait(t)))
877 _, _, restores := run.terminalCounts()
878 require.Equal(t, 1, restores, "raw mode is restored exactly once")
879}
880
881// TestTerminalCompositionCapturesSessionEnvironmentDefensively proves the test
882// terminal composition keeps its process snapshot defensive: it captures
883// env/Cwd, mutates the process seam after construction, and the supervisor
884// still creates without failure.
885func TestTerminalCompositionCapturesSessionEnvironmentDefensively(t *testing.T) {
886 t.Setenv("VEV_TEST_SESSION_ENV", "composition-value")
887 snapshot := terminalSessionEnvironment()
888 require.Equal(t, client.SessionEnvironmentLocalCLI, snapshot.Provenance)
889 require.NoError(t, snapshot.Validate())
890 require.Contains(t, snapshot.Env, "VEV_TEST_SESSION_ENV=composition-value")
891 // Mutate the process seam after capture: the snapshot must keep the
892 // captured value rather than aliasing live process state.
893 require.NoError(t, os.Setenv("VEV_TEST_SESSION_ENV", "mutated-after-capture"))
894 require.Contains(t, snapshot.Env, "VEV_TEST_SESSION_ENV=composition-value")
895 require.NotContains(t, snapshot.Env, "VEV_TEST_SESSION_ENV=mutated-after-capture")
896 clone := snapshot.Clone()
897 // Mutate the clone's slice after capture: the supervisor creation must
898 // not fail and the source slice stays intact, so the snapshot is a copy
899 // rather than a live alias of the process environment.
900 for i := range clone.Env {
901 clone.Env[i] = "MUTATED=x"
902 }
903 require.NotEqual(t, snapshot.Env, clone.Env)
904 servicesnapshot := terminalCompositionSnapshot([]string{"work"}, nil)
905 service := newTerminalCompositionService(servicesnapshot)
906 terminal := newTerminalCompositionTerminal()
907 navigation := localEphemeralNavigation()
908 connector := portsmocks.NewMockBrokerConnector(t)
909 connector.EXPECT().Connect(mock.Anything).Return(service, nil).Maybe()
910 _, err := client.NewSupervisor(client.SupervisorConfig{
911 Connector: connector,
912 Terminal: terminal,
913 Clock: clock.New(),
914 // The picker is omitted: creation alone proves the snapshot does not
915 // fail supervisor validation.
916 InitialNavigation: navigation,
917 SessionEnvironment: snapshot,
918 })
919 require.NoError(t, err)
920}
921
922// TestTerminalCompositionReportsBrokerFailureInThePicker pins that a broker the
923// composition cannot connect to (or start) is no longer a fatal startup error:
924// the run enters raw mode, stays in the picker, and reports the typed failure
925// without opening any stream. It remains live until the terminal ends.
926func TestTerminalCompositionReportsBrokerFailureInThePicker(t *testing.T) {
927 t.Setenv("VEV", "")
928 startupErr := errors.New("broker not reachable")
929 previousConnect := connectProductionClientBroker
930 terminal := newTerminalCompositionTerminal()
931 previousTerminal := terminalForAttach
932 connectProductionClientBroker = func(context.Context) (ports.BrokerService, error) { return nil, startupErr }
933 terminalForAttach = func() ports.Terminal { return terminal }
934 t.Cleanup(func() {
935 connectProductionClientBroker = previousConnect
936 terminalForAttach = previousTerminal
937 })
938
939 ctx, cancel := context.WithCancel(context.Background())
940 done := make(chan error, 1)
941 go func() { done <- runAttach(ctx, protocol.IntentEphemeral, "", "") }()
942
943 // A broker failure is surfaced, not fatal, and no stream is opened.
944 require.Eventually(t, func() bool {
945 for _, status := range terminal.publishes() {
946 if status == ports.UIStatusConnecting || status == ports.UIStatusAttached {
947 return false
948 }
949 }
950 return true
951 }, 200*time.Millisecond, 10*time.Millisecond)
952 select {
953 case err := <-done:
954 t.Fatalf("the composition ended on a broker failure: %v", err)
955 case <-time.After(200 * time.Millisecond):
956 }
957 _, _, restores := terminalCounts(terminal)
958 require.Zero(t, restores, "the run is still live, so raw mode is not yet restored")
959
960 cancel()
961 require.NoError(t, ignoreContextCancellation(<-done))
962 _, _, restores = terminalCounts(terminal)
963 require.Equal(t, 1, restores, "the cancelled run restores raw mode exactly once")
964}
965
966// TestRunAttachNestedSessionBehavior pins the nested-session product contract of
967// the terminal path: an explicit `new <name>` inside an existing session still
968// creates the session without attaching, and every other intent is refused
969// without reaching the broker.
970func TestRunAttachNestedSessionBehavior(t *testing.T) {
971 t.Setenv("VEV", "outer")
972 connectCalls := 0
973 previousConnect := connectProductionClientBroker
974 previousCreate := createDetachedTerminalSession
975 var createdName string
976 connectProductionClientBroker = func(context.Context) (ports.BrokerService, error) {
977 connectCalls++
978 return nil, errors.New("the broker must not be reached")
979 }
980 createDetachedTerminalSession = func(_ context.Context, name string) error {
981 createdName = name
982 return nil
983 }
984 t.Cleanup(func() {
985 connectProductionClientBroker = previousConnect
986 createDetachedTerminalSession = previousCreate
987 })
988
989 require.NoError(t, runAttach(context.Background(), protocol.IntentNew, "scratch", ""))
990 require.Equal(t, "scratch", createdName)
991
992 for _, intent := range []uint8{protocol.IntentEphemeral, protocol.IntentAttach} {
993 err := runAttach(context.Background(), intent, "scratch", "")
994 require.ErrorContains(t, err, "sessions should be nested with care")
995 }
996 require.Zero(t, connectCalls, "a nested invocation never connects to the broker")
997}
998
999func terminalCounts(terminal *terminalCompositionTerminal) (int, int, int) {
1000 terminal.mu.Lock()
1001 defer terminal.mu.Unlock()
1002 return len(terminal.out), terminal.flushes, terminal.restores
1003}
1004
1005// TestTerminalCompositionUsesTheBrokerConnector pins the production terminal
1006// path source: runAttach composes the shared broker client over the process'
1007// production connector and keeps no direct-dialer attach composition, attach
1008// preflight, or client host registry. The old pre-cutover symbols are removed
1009// outright, so this fails if any is reintroduced as a bypass.
1010func TestTerminalCompositionUsesTheBrokerConnector(t *testing.T) {
1011 sources := loadAppSources(t, false)
1012 body := sources["run.go"]
1013 require.Contains(t, body, "runBrokerClient(ctx, brokerClientConfig{")
1014 require.Contains(t, body, "newProductionBrokerConnector()")
1015 for _, forbidden := range []string{
1016 "runAttachWithDeps", "runAttachDeps", "localDaemonDialer", "dialOnlyLocalDialer",
1017 "defaultLocalDialer", "resolveMissingSessionAttach", "newClientHostRegistry",
1018 "listSessionsWithDialer", "clientEndpointFactory",
1019 } {
1020 require.NotContains(t, body, forbidden, "the terminal composition must not keep %q", forbidden)
1021 }
1022}
1023
1024// TestTerminalExactAttachAcceptedByRealDaemon is the end-to-end regression for
1025// the exact-attach wire shape: against the real broker and a real daemon, the
1026// Hello the client supervisor builds for an exact admission (Intent=attach,
1027// Name equal to the exact target's session name, ExactTarget set) is accepted
1028// and answered with a committed Welcome. It fails if the strict Hello contract
1029// and the worker disagree about the name.
1030func TestTerminalExactAttachAcceptedByRealDaemon(t *testing.T) {
1031 fixture := startOfflineClientFixture(t)
1032 ctx, cancel := context.WithTimeout(context.Background(), brokerTestWait)
1033 defer cancel()
1034 service, err := brokeripc.NewConnector(fixture.socket, brokeripc.Config{}).Connect(ctx)
1035 require.NoError(t, err)
1036 defer service.Close()
1037 require.NotZero(t, awaitTerminalCompositionEpoch(t, service))
1038
1039 createID, err := service.NextStreamID()
1040 require.NoError(t, err)
1041 create, err := service.OpenStream(ctx, ports.BrokerOpenStreamRequest{
1042 Purpose: ports.BrokerStreamAttachment, Stream: createID, Local: true, Policy: fixture.policy,
1043 Admission: ports.BrokerAdmissionCreateNamed, Name: "attach-exact", StartMode: ports.BrokerDaemonStartIfNeeded,
1044 })
1045 require.NoError(t, err)
1046 require.NoError(t, create.SendClient(protocol.Hello{Version: protocol.Version, Intent: protocol.IntentNew, Name: "attach-exact", Size: domain.Size{Cols: 80, Rows: 24}}))
1047 target := receiveOfflineWelcome(t, create).CommittedIdentity.Target
1048 require.NoError(t, create.Close())
1049
1050 exactID, err := service.NextStreamID()
1051 require.NoError(t, err)
1052 exact, err := service.OpenStream(ctx, ports.BrokerOpenStreamRequest{
1053 Purpose: ports.BrokerStreamAttachment, Stream: exactID, Local: true, Policy: fixture.policy,
1054 Admission: ports.BrokerAdmissionExact, Target: target, StartMode: ports.BrokerDaemonStartIfNeeded,
1055 })
1056 require.NoError(t, err)
1057 defer exact.Close()
1058
1059 // The exact Hello shape the client supervisor's worker builds for this
1060 // request; it must satisfy the protocol and be accepted by the daemon.
1061 hello := protocol.Hello{
1062 Version: protocol.Version, Intent: protocol.IntentAttach, Name: target.SessionName,
1063 ExactTarget: &target, Size: domain.Size{Cols: 80, Rows: 24},
1064 }
1065 require.NoError(t, protocol.ValidateHello(hello))
1066 require.NoError(t, exact.SendClient(hello))
1067 require.Equal(t, "attach-exact", receiveOfflineWelcome(t, exact).SessionName)
1068}
1069
1070// TestTerminalCompositionExactAttachReportsMissingLocalInventory documents the
1071// composition's behavior when the committed broker publication carries no local
1072// session inventory: an exact attach is refused locally as a bounded selection
1073// notice, opens no stream, and never invents a creation or a same-name
1074// fallback. It pins the honest refusal so a future broker-side local catalogue
1075// is the only thing that can make this navigation succeed.
1076func TestTerminalCompositionExactAttachReportsMissingLocalInventory(t *testing.T) {
1077 fixture := startOfflineClientFixture(t)
1078 ctx, cancel := context.WithTimeout(context.Background(), brokerTestWait)
1079 defer cancel()
1080
1081 terminal := newTerminalCompositionTerminal()
1082 notices := make(chan client.LifecycleNotice, 16)
1083 failures := make(chan error, 16)
1084 var lifecycleID domain.SessionLifecycleID
1085 lifecycleID[0], lifecycleID[1] = 3, 0x5a
1086 navigation := client.InitialNavigation{
1087 Kind: client.InitialNavigationAttachExact,
1088 Epoch: terminalCompositionEpoch,
1089 Destination: ports.BrokerEndpointFence{Local: true},
1090 Target: protocol.ExactSessionTarget{LifecycleID: lifecycleID, SessionName: "work"},
1091 }
1092 require.NoError(t, navigation.Validate())
1093
1094 done := make(chan error, 1)
1095 go func() {
1096 done <- runBrokerClient(ctx, brokerClientConfig{
1097 Connector: brokeripc.NewConnector(fixture.socket, brokeripc.Config{}).Navigator(),
1098 Terminal: terminal,
1099 InitialNavigation: navigation,
1100 SessionEnvironment: client.SessionEnvironment{Provenance: client.SessionEnvironmentLocalCLI},
1101 OnFailure: func(err error) {
1102 select {
1103 case failures <- err:
1104 default:
1105 }
1106 },
1107 OnLifecycle: func(notice client.LifecycleNotice) {
1108 select {
1109 case notices <- notice:
1110 default:
1111 }
1112 },
1113 })
1114 }()
1115
1116 select {
1117 case notice := <-notices:
1118 require.Equal(t, client.LifecycleNoticeSelectionUnavailable, notice.Kind)
1119 case <-time.After(brokerTestWait):
1120 t.Fatal("the exact attach reported no lifecycle notice")
1121 }
1122 select {
1123 case failure := <-failures:
1124 require.Error(t, failure)
1125 case <-time.After(brokerTestWait):
1126 t.Fatal("the exact attach reported no failure")
1127 }
1128 // The refused navigation never writes a session frame and never attaches:
1129 // it presents Connecting while pending and ends on the picker.
1130 publishes := terminal.publishes()
1131 for _, status := range publishes {
1132 require.NotEqual(t, ports.UIStatusAttached, status, "a refused navigation never attaches")
1133 }
1134 require.NotEmpty(t, publishes)
1135 require.Equal(t, ports.UIStatusPicker, publishes[len(publishes)-1], "a refused navigation ends on the picker")
1136 cancel()
1137 require.NoError(t, ignoreContextCancellation(<-done))
1138}
1139
1140// awaitTerminalCompositionEpoch waits for one connection's first committed
1141// publication and returns its epoch.
1142func awaitTerminalCompositionEpoch(t *testing.T, service ports.BrokerService) ports.BrokerEpoch {
1143 t.Helper()
1144 sub, err := service.Subscribe()
1145 require.NoError(t, err)
1146 defer sub.Close()
1147 deadline := time.NewTimer(brokerTestWait)
1148 defer deadline.Stop()
1149 for {
1150 if snapshot := service.Snapshot(); snapshot.Epoch != 0 {
1151 return snapshot.Epoch
1152 }
1153 select {
1154 case <-sub.Changed():
1155 case <-service.Done():
1156 t.Fatalf("broker connection ended before publishing a snapshot: %v", service.Err())
1157 case <-deadline.C:
1158 t.Fatal("broker never published a snapshot")
1159 }
1160 }
1161}