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 and remote-transport
534 // environment; a test drives the broker composition with both unset.
535 t.Setenv("VEV", "")
536 t.Setenv(envRemoteTransport, "")
537 service := newTerminalCompositionService(snapshot)
538 service.daemonSessions = daemonSessions
539 terminal := newTerminalCompositionTerminal()
540 previousConnect := connectProductionClientBroker
541 previousTerminal := terminalForAttach
542 previousCallbacks := terminalBrokerCallbacks
543 connectProductionClientBroker = func(context.Context) (ports.BrokerService, error) {
544 return service, nil
545 }
546 terminalForAttach = func() ports.Terminal { return terminal }
547
548 run := &terminalCompositionRun{
549 service: service,
550 terminal: terminal,
551 states: make(chan client.State, 64),
552 notices: make(chan client.LifecycleNotice, 16),
553 failures: make(chan error, 16),
554 finished: make(chan struct{}),
555 }
556 terminalBrokerCallbacks = func() terminalBrokerSeams {
557 return terminalBrokerSeams{
558 OnState: func(state client.State) {
559 select {
560 case run.states <- state:
561 default:
562 }
563 },
564 OnLifecycle: func(notice client.LifecycleNotice) {
565 select {
566 case run.notices <- notice:
567 default:
568 }
569 },
570 OnFailure: func(err error) {
571 select {
572 case run.failures <- err:
573 default:
574 }
575 },
576 }
577 }
578 t.Cleanup(func() {
579 connectProductionClientBroker = previousConnect
580 terminalForAttach = previousTerminal
581 terminalBrokerCallbacks = previousCallbacks
582 })
583
584 ctx, cancel := context.WithCancel(context.Background())
585 run.cancel = cancel
586 go func() {
587 err := runAttach(ctx, intent, name, remoteTarget)
588 run.runMu.Lock()
589 run.runErr = err
590 run.runMu.Unlock()
591 close(run.finished)
592 }()
593 t.Cleanup(func() {
594 cancel()
595 select {
596 case <-run.finished:
597 case <-time.After(brokerTestWait):
598 t.Error("terminal client did not stop")
599 }
600 })
601 return run
602}
603
604// awaitTerminalCompositionAttachment waits for the supervisor to present an
605// attached state and returns it.
606func awaitTerminalCompositionAttachment(t *testing.T, run *terminalCompositionRun) client.State {
607 t.Helper()
608 deadline := time.NewTimer(brokerTestWait)
609 defer deadline.Stop()
610 for {
611 select {
612 case state := <-run.states:
613 if state.Presentation == client.PresentAttached {
614 return state
615 }
616 case <-run.finished:
617 t.Fatalf("terminal client ended before attaching: %v", run.wait(t))
618 case <-deadline.C:
619 t.Fatal("terminal client never attached")
620 }
621 }
622}
623
624// awaitTerminalCompositionNotice waits for one bounded lifecycle notice.
625func awaitTerminalCompositionNotice(t *testing.T, run *terminalCompositionRun) client.LifecycleNotice {
626 t.Helper()
627 select {
628 case notice := <-run.notices:
629 return notice
630 case <-time.After(brokerTestWait):
631 t.Fatal("terminal client reported no lifecycle notice")
632 return client.LifecycleNotice{}
633 }
634}
635
636func TestTerminalNavigationTranslationClosesTheIntent(t *testing.T) {
637 tests := []struct {
638 name string
639 intent uint8
640 session string
641 remote string
642 wantKind client.InitialNavigationKind
643 wantLocal bool
644 wantName string
645 wantResolver bool
646 wantError bool
647 }{
648 {name: "no arguments creates a local ephemeral session", intent: protocol.IntentEphemeral, wantKind: client.InitialNavigationCreateEphemeral, wantLocal: true},
649 {name: "new creates a local named session", intent: protocol.IntentNew, session: "scratch", wantKind: client.InitialNavigationCreateNamed, wantLocal: true, wantName: "scratch"},
650 {name: "attach names a local session for the daemon to resolve", intent: protocol.IntentAttach, session: "work", wantKind: client.InitialNavigationAttachNamed, wantLocal: true, wantName: "work"},
651 {name: "remote attach resolves the remote registration", intent: protocol.IntentAttach, session: "work", remote: "user@example.test", wantResolver: true},
652 {name: "remote new resolves a named remote creation", intent: protocol.IntentNew, session: "work", remote: "user@example.test", wantResolver: true},
653 {name: "remote ephemeral attach resolves a remote creation", intent: protocol.IntentEphemeral, remote: "user@example.test", wantResolver: true},
654 {name: "invalid local name is refused", intent: protocol.IntentNew, session: "bad name", wantError: true},
655 {name: "invalid remote target is refused", intent: protocol.IntentAttach, session: "work", remote: "bad target", wantError: true},
656 }
657 for _, tc := range tests {
658 t.Run(tc.name, func(t *testing.T) {
659 navigation, resolver, err := terminalBrokerNavigation(tc.intent, tc.session, tc.remote)
660 if tc.wantError {
661 require.Error(t, err)
662 require.Nil(t, resolver)
663 require.Equal(t, client.InitialNavigation{}, navigation)
664 return
665 }
666 require.NoError(t, err)
667 if tc.wantResolver {
668 require.NotNil(t, resolver)
669 require.Equal(t, client.InitialNavigation{}, navigation)
670 return
671 }
672 require.Nil(t, resolver)
673 require.Equal(t, tc.wantKind, navigation.Kind)
674 require.Equal(t, tc.wantLocal, navigation.Destination.Local)
675 require.Equal(t, tc.wantName, navigation.Name)
676 require.NoError(t, navigation.Validate())
677 })
678 }
679}
680
681// TestTerminalCompositionOpensTheExactBrokerStream pins the ordinary terminal
682// entry point to the broker: each parsed CLI target resolves to exactly one
683// admission on one broker logical stream, and the composition dials no daemon.
684func TestTerminalCompositionOpensTheExactBrokerStream(t *testing.T) {
685 localSnapshot := func() ports.BrokerSnapshot {
686 return terminalCompositionSnapshot([]string{"work"}, nil)
687 }
688 remoteSnapshot := func() ports.BrokerSnapshot {
689 return terminalCompositionSnapshot(nil, map[string][]string{"user@example.test": {"work"}})
690 }
691 tests := []struct {
692 name string
693 intent uint8
694 session string
695 remote string
696 snapshot ports.BrokerSnapshot
697 daemonSessions []string
698 wantAdmission ports.BrokerStreamAdmission
699 wantLocal bool
700 wantName string
701 wantHello uint8
702 }{
703 {name: "no arguments creates a local ephemeral session", intent: protocol.IntentEphemeral, snapshot: localSnapshot(), wantAdmission: ports.BrokerAdmissionCreateEphemeral, wantLocal: true, wantHello: protocol.IntentEphemeral},
704 {name: "new creates a local named session", intent: protocol.IntentNew, session: "scratch", snapshot: localSnapshot(), wantAdmission: ports.BrokerAdmissionCreateNamed, wantLocal: true, wantName: "scratch", wantHello: protocol.IntentNew},
705 {name: "attach names the local session", intent: protocol.IntentAttach, session: "work", snapshot: localSnapshot(), wantAdmission: ports.BrokerAdmissionAttachNamed, wantLocal: true, wantName: "work", wantHello: protocol.IntentAttach},
706 {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},
707 {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},
708 }
709 for _, tc := range tests {
710 t.Run(tc.name, func(t *testing.T) {
711 run := startTerminalCompositionWithDaemon(t, tc.intent, tc.session, tc.remote, tc.snapshot, tc.daemonSessions)
712 awaitTerminalCompositionAttachment(t, run)
713 opened := run.service.openedRequests()
714 require.Len(t, opened, 1, "one CLI target opens exactly one broker stream")
715 request := opened[0]
716 require.NoError(t, request.Validate())
717 require.Equal(t, terminalCompositionEpoch, request.Epoch)
718 require.Equal(t, ports.BrokerStreamAttachment, request.Purpose)
719 require.Equal(t, tc.wantAdmission, request.Admission)
720 require.Equal(t, tc.wantLocal, request.Local)
721 require.Equal(t, tc.wantName, request.Name)
722 require.Equal(t, ports.BrokerConnectionID{4}, request.Connection)
723 require.NotZero(t, request.Stream)
724 require.NotEmpty(t, request.Policy.Transport)
725 require.Equal(t, ports.BrokerDaemonStartIfNeeded, request.StartMode)
726 if !tc.wantLocal {
727 require.Equal(t, "user@example.test", request.Endpoint)
728 require.Equal(t, terminalCompositionRegistration("user@example.test"), request.Registration)
729 }
730 if tc.wantAdmission == ports.BrokerAdmissionExact {
731 require.NoError(t, request.Target.Validate())
732 require.Equal(t, tc.session, request.Target.SessionName)
733 }
734 // The same intent reaches the typed session handshake, so the
735 // broker admission and the daemon Hello can never disagree.
736 hello := run.service.streamHello(t)
737 require.Equal(t, tc.wantHello, hello.Intent)
738 require.Equal(t, tc.session, hello.Name)
739 require.Equal(t, tc.remote != "", hello.Remote)
740 if tc.wantAdmission == ports.BrokerAdmissionExact {
741 require.NotNil(t, hello.ExactTarget)
742 require.Equal(t, request.Target, *hello.ExactTarget)
743 }
744 if tc.wantAdmission == ports.BrokerAdmissionAttachNamed {
745 require.Nil(t, hello.ExactTarget, "a named attach leaves identity to the daemon")
746 controls := run.service.openedControls()
747 require.Len(t, controls, 1, "the preflight asks the daemon once")
748 require.Equal(t, ports.BrokerDaemonStartIfNeeded, controls[0].StartMode, "the preflight may start a stopped daemon")
749 }
750 })
751 }
752}
753
754// terminalCompositionUnobservedLocal is the publication a freshly started
755// broker commits right after `kill --all`: the local daemon is known but its
756// sessions are not observed yet, while the daemon itself restores them.
757func terminalCompositionUnobservedLocal() ports.BrokerSnapshot {
758 snapshot := terminalCompositionSnapshot(nil, nil)
759 snapshot.Daemons[0].InventoryKnown = false
760 return snapshot
761}
762
763// TestTerminalCompositionPublishesRealSessionOutput proves an admitted
764// attachment is a committed publication: the daemon's initial full frame reached
765// the terminal through a committed UI output transaction, and the supervisor
766// presented attached only after that commit.
767func TestTerminalCompositionPublishesRealSessionOutput(t *testing.T) {
768 run := startTerminalComposition(t, protocol.IntentEphemeral, "", "", terminalCompositionSnapshot([]string{"work"}, nil))
769 awaitTerminalCompositionAttachment(t, run)
770 require.Contains(t, run.terminal.written(), "terminal-composition-ready")
771 // The initial publication is committed before the attached marker, so its
772 // recorded boundary status is the pre-attachment Connecting presentation.
773 require.Contains(t, run.terminal.publishes(), ports.UIStatusConnecting)
774 // The attached presentation is published just after the attached state, on
775 // the attachment goroutine, so wait for it instead of racing it.
776 require.Eventually(t, func() bool {
777 return slices.Contains(run.terminal.publishes(), ports.UIStatusAttached)
778 }, brokerTestWait, 5*time.Millisecond)
779 _, _, restores := run.terminalCounts()
780 require.Zero(t, restores)
781}
782
783func (r *terminalCompositionRun) terminalCounts() (int, int, int) {
784 r.terminal.mu.Lock()
785 defer r.terminal.mu.Unlock()
786 return len(r.terminal.out), r.terminal.flushes, r.terminal.restores
787}
788
789// TestTerminalCompositionRefusedTargetOpensNothingAndKeepsThePicker pins the
790// failure contract: a target the committed catalogue does not carry is a local
791// refusal. No stream is opened, the supervisor reports a bounded
792// selection-unavailable notice, and the process stays in the picker instead of
793// attaching a fallback or creating implicitly.
794func TestTerminalCompositionRefusedTargetOpensNothingAndKeepsThePicker(t *testing.T) {
795 tests := []struct {
796 name string
797 intent uint8
798 session string
799 remote string
800 snapshot ports.BrokerSnapshot
801 }{
802 {name: "local daemon absent", intent: protocol.IntentEphemeral, snapshot: ports.BrokerSnapshot{Epoch: terminalCompositionEpoch, Revision: 1}},
803 {name: "unconfigured remote host", intent: protocol.IntentAttach, session: "work", remote: "user@elsewhere.test", snapshot: terminalCompositionSnapshot([]string{"work"}, nil)},
804 }
805 for _, tc := range tests {
806 t.Run(tc.name, func(t *testing.T) {
807 run := startTerminalComposition(t, tc.intent, tc.session, tc.remote, tc.snapshot)
808 require.Equal(t, client.LifecycleNoticeSelectionUnavailable, awaitTerminalCompositionNotice(t, run).Kind)
809 // A local refusal is surfaced once, with the bounded text the supervisor
810 // classifies; the resolver's own sentinel is deliberately not propagated
811 // past that taxonomy, so only the presence and the classification are
812 // asserted.
813 select {
814 case failure := <-run.failures:
815 require.Error(t, failure)
816 case <-time.After(brokerTestWait):
817 t.Fatal("the refused target reported no failure")
818 }
819 require.Empty(t, run.service.openedRequests(), "a refused target never opens a stream")
820 // The pending navigation presents Connecting, never the picker
821 // first; the refusal then returns to the picker and never attaches.
822 last := client.PresentConnecting
823 for {
824 select {
825 case state := <-run.states:
826 require.NotEqual(t, client.PresentAttached, state.Presentation, "a refusal never attaches")
827 last = state.Presentation
828 case <-time.After(100 * time.Millisecond):
829 require.Equal(t, client.PresentPicker, last, "a refusal ends on the picker")
830 return
831 }
832 }
833 })
834 }
835}
836
837// TestTerminalCompositionDaemonRefusesMissingName pins that existence is the
838// daemon's decision: a declined (non-interactive) create prompt keeps the
839// named attach, the daemon answers ErrNoSuchSession, and the client reports
840// that refusal without attaching or creating.
841func TestTerminalCompositionDaemonRefusesMissingName(t *testing.T) {
842 tests := []struct {
843 name string
844 remote string
845 snapshot ports.BrokerSnapshot
846 }{
847 {name: "local", snapshot: terminalCompositionSnapshot([]string{"work"}, nil)},
848 {name: "remote", remote: "user@example.test", snapshot: terminalCompositionSnapshot(nil, map[string][]string{"user@example.test": {"work"}})},
849 }
850 for _, tc := range tests {
851 t.Run(tc.name, func(t *testing.T) {
852 withAttachInteractiveConsole(t, false)
853 run := startTerminalComposition(t, protocol.IntentAttach, "gone", tc.remote, tc.snapshot)
854 require.Equal(t, client.LifecycleNoticeDestinationFailed, awaitTerminalCompositionNotice(t, run).Kind)
855 select {
856 case failure := <-run.failures:
857 var protocolErr *client.ProtocolError
858 require.ErrorAs(t, failure, &protocolErr)
859 require.Equal(t, protocol.ErrNoSuchSession, protocolErr.Code)
860 case <-time.After(brokerTestWait):
861 t.Fatal("the daemon refusal was not reported")
862 }
863 opened := run.service.openedRequests()
864 require.Len(t, opened, 1)
865 require.Equal(t, ports.BrokerAdmissionAttachNamed, opened[0].Admission, "a missing name is never created implicitly")
866 })
867 }
868}
869
870// TestTerminalCompositionClosesRawModeOnTerminalEnd pins that the shared
871// composition restores the terminal exactly once when the run ends, however the
872// attachment settled.
873func TestTerminalCompositionClosesRawModeOnTerminalEnd(t *testing.T) {
874 run := startTerminalComposition(t, protocol.IntentEphemeral, "", "", terminalCompositionSnapshot([]string{"work"}, nil))
875 awaitTerminalCompositionAttachment(t, run)
876 run.cancel()
877 require.NoError(t, ignoreContextCancellation(run.wait(t)))
878 _, _, restores := run.terminalCounts()
879 require.Equal(t, 1, restores, "raw mode is restored exactly once")
880}
881
882// TestTerminalCompositionCapturesSessionEnvironmentDefensively proves the test
883// terminal composition keeps its process snapshot defensive: it captures
884// env/Cwd, mutates the process seam after construction, and the supervisor
885// still creates without failure.
886func TestTerminalCompositionCapturesSessionEnvironmentDefensively(t *testing.T) {
887 t.Setenv("VEV_TEST_SESSION_ENV", "composition-value")
888 snapshot := terminalSessionEnvironment()
889 require.Equal(t, client.SessionEnvironmentLocalCLI, snapshot.Provenance)
890 require.NoError(t, snapshot.Validate())
891 require.Contains(t, snapshot.Env, "VEV_TEST_SESSION_ENV=composition-value")
892 // Mutate the process seam after capture: the snapshot must keep the
893 // captured value rather than aliasing live process state.
894 require.NoError(t, os.Setenv("VEV_TEST_SESSION_ENV", "mutated-after-capture"))
895 require.Contains(t, snapshot.Env, "VEV_TEST_SESSION_ENV=composition-value")
896 require.NotContains(t, snapshot.Env, "VEV_TEST_SESSION_ENV=mutated-after-capture")
897 clone := snapshot.Clone()
898 // Mutate the clone's slice after capture: the supervisor creation must
899 // not fail and the source slice stays intact, so the snapshot is a copy
900 // rather than a live alias of the process environment.
901 for i := range clone.Env {
902 clone.Env[i] = "MUTATED=x"
903 }
904 require.NotEqual(t, snapshot.Env, clone.Env)
905 servicesnapshot := terminalCompositionSnapshot([]string{"work"}, nil)
906 service := newTerminalCompositionService(servicesnapshot)
907 terminal := newTerminalCompositionTerminal()
908 navigation := localEphemeralNavigation()
909 connector := portsmocks.NewMockBrokerConnector(t)
910 connector.EXPECT().Connect(mock.Anything).Return(service, nil).Maybe()
911 _, err := client.NewSupervisor(client.SupervisorConfig{
912 Connector: connector,
913 Terminal: terminal,
914 Clock: clock.New(),
915 // The picker is omitted: creation alone proves the snapshot does not
916 // fail supervisor validation.
917 InitialNavigation: navigation,
918 SessionEnvironment: snapshot,
919 })
920 require.NoError(t, err)
921}
922
923// TestTerminalCompositionReportsBrokerFailureInThePicker pins that a broker the
924// composition cannot connect to (or start) is no longer a fatal startup error:
925// the run enters raw mode, stays in the picker, and reports the typed failure
926// without opening any stream. It remains live until the terminal ends.
927func TestTerminalCompositionReportsBrokerFailureInThePicker(t *testing.T) {
928 t.Setenv("VEV", "")
929 t.Setenv(envRemoteTransport, "")
930 startupErr := errors.New("broker not reachable")
931 previousConnect := connectProductionClientBroker
932 terminal := newTerminalCompositionTerminal()
933 previousTerminal := terminalForAttach
934 connectProductionClientBroker = func(context.Context) (ports.BrokerService, error) { return nil, startupErr }
935 terminalForAttach = func() ports.Terminal { return terminal }
936 t.Cleanup(func() {
937 connectProductionClientBroker = previousConnect
938 terminalForAttach = previousTerminal
939 })
940
941 ctx, cancel := context.WithCancel(context.Background())
942 done := make(chan error, 1)
943 go func() { done <- runAttach(ctx, protocol.IntentEphemeral, "", "") }()
944
945 // A broker failure is surfaced, not fatal, and no stream is opened.
946 require.Eventually(t, func() bool {
947 for _, status := range terminal.publishes() {
948 if status == ports.UIStatusConnecting || status == ports.UIStatusAttached {
949 return false
950 }
951 }
952 return true
953 }, 200*time.Millisecond, 10*time.Millisecond)
954 select {
955 case err := <-done:
956 t.Fatalf("the composition ended on a broker failure: %v", err)
957 case <-time.After(200 * time.Millisecond):
958 }
959 _, _, restores := terminalCounts(terminal)
960 require.Zero(t, restores, "the run is still live, so raw mode is not yet restored")
961
962 cancel()
963 require.NoError(t, ignoreContextCancellation(<-done))
964 _, _, restores = terminalCounts(terminal)
965 require.Equal(t, 1, restores, "the cancelled run restores raw mode exactly once")
966}
967
968// TestRunAttachNestedSessionBehavior pins the nested-session product contract of
969// the terminal path: an explicit `new <name>` inside an existing session still
970// creates the session without attaching, and every other intent is refused
971// without reaching the broker.
972func TestRunAttachNestedSessionBehavior(t *testing.T) {
973 t.Setenv("VEV", "outer")
974 t.Setenv(envRemoteTransport, "")
975 connectCalls := 0
976 previousConnect := connectProductionClientBroker
977 previousCreate := createDetachedTerminalSession
978 var createdName string
979 connectProductionClientBroker = func(context.Context) (ports.BrokerService, error) {
980 connectCalls++
981 return nil, errors.New("the broker must not be reached")
982 }
983 createDetachedTerminalSession = func(_ context.Context, name string) error {
984 createdName = name
985 return nil
986 }
987 t.Cleanup(func() {
988 connectProductionClientBroker = previousConnect
989 createDetachedTerminalSession = previousCreate
990 })
991
992 require.NoError(t, runAttach(context.Background(), protocol.IntentNew, "scratch", ""))
993 require.Equal(t, "scratch", createdName)
994
995 for _, intent := range []uint8{protocol.IntentEphemeral, protocol.IntentAttach} {
996 err := runAttach(context.Background(), intent, "scratch", "")
997 require.ErrorContains(t, err, "sessions should be nested with care")
998 }
999 require.Zero(t, connectCalls, "a nested invocation never connects to the broker")
1000}
1001
1002func terminalCounts(terminal *terminalCompositionTerminal) (int, int, int) {
1003 terminal.mu.Lock()
1004 defer terminal.mu.Unlock()
1005 return len(terminal.out), terminal.flushes, terminal.restores
1006}
1007
1008// TestTerminalCompositionUsesTheBrokerConnector pins the production terminal
1009// path source: runAttach composes the shared broker client over the process'
1010// production connector and keeps no direct-dialer attach composition, attach
1011// preflight, or client host registry. The old pre-cutover symbols are removed
1012// outright, so this fails if any is reintroduced as a bypass.
1013func TestTerminalCompositionUsesTheBrokerConnector(t *testing.T) {
1014 sources := loadAppSources(t, false)
1015 body := sources["run.go"]
1016 require.Contains(t, body, "runBrokerClient(ctx, brokerClientConfig{")
1017 require.Contains(t, body, "newProductionBrokerConnector()")
1018 for _, forbidden := range []string{
1019 "runAttachWithDeps", "runAttachDeps", "localDaemonDialer", "dialOnlyLocalDialer",
1020 "defaultLocalDialer", "resolveMissingSessionAttach", "newClientHostRegistry",
1021 "listSessionsWithDialer", "clientEndpointFactory",
1022 } {
1023 require.NotContains(t, body, forbidden, "the terminal composition must not keep %q", forbidden)
1024 }
1025}
1026
1027// TestTerminalExactAttachAcceptedByRealDaemon is the end-to-end regression for
1028// the exact-attach wire shape: against the real broker and a real daemon, the
1029// Hello the client supervisor builds for an exact admission (Intent=attach,
1030// Name equal to the exact target's session name, ExactTarget set) is accepted
1031// and answered with a committed Welcome. It fails if the strict Hello contract
1032// and the worker disagree about the name.
1033func TestTerminalExactAttachAcceptedByRealDaemon(t *testing.T) {
1034 fixture := startOfflineClientFixture(t)
1035 ctx, cancel := context.WithTimeout(context.Background(), brokerTestWait)
1036 defer cancel()
1037 service, err := brokeripc.NewConnector(fixture.socket, brokeripc.Config{}).Connect(ctx)
1038 require.NoError(t, err)
1039 defer service.Close()
1040 require.NotZero(t, awaitTerminalCompositionEpoch(t, service))
1041
1042 createID, err := service.NextStreamID()
1043 require.NoError(t, err)
1044 create, err := service.OpenStream(ctx, ports.BrokerOpenStreamRequest{
1045 Purpose: ports.BrokerStreamAttachment, Stream: createID, Local: true, Policy: fixture.policy,
1046 Admission: ports.BrokerAdmissionCreateNamed, Name: "attach-exact", StartMode: ports.BrokerDaemonStartIfNeeded,
1047 })
1048 require.NoError(t, err)
1049 require.NoError(t, create.SendClient(protocol.Hello{Version: protocol.Version, Intent: protocol.IntentNew, Name: "attach-exact", Size: domain.Size{Cols: 80, Rows: 24}}))
1050 target := receiveOfflineWelcome(t, create).CommittedIdentity.Target
1051 require.NoError(t, create.Close())
1052
1053 exactID, err := service.NextStreamID()
1054 require.NoError(t, err)
1055 exact, err := service.OpenStream(ctx, ports.BrokerOpenStreamRequest{
1056 Purpose: ports.BrokerStreamAttachment, Stream: exactID, Local: true, Policy: fixture.policy,
1057 Admission: ports.BrokerAdmissionExact, Target: target, StartMode: ports.BrokerDaemonStartIfNeeded,
1058 })
1059 require.NoError(t, err)
1060 defer exact.Close()
1061
1062 // The exact Hello shape the client supervisor's worker builds for this
1063 // request; it must satisfy the protocol and be accepted by the daemon.
1064 hello := protocol.Hello{
1065 Version: protocol.Version, Intent: protocol.IntentAttach, Name: target.SessionName,
1066 ExactTarget: &target, Size: domain.Size{Cols: 80, Rows: 24},
1067 }
1068 require.NoError(t, protocol.ValidateHello(hello))
1069 require.NoError(t, exact.SendClient(hello))
1070 require.Equal(t, "attach-exact", receiveOfflineWelcome(t, exact).SessionName)
1071}
1072
1073// TestTerminalCompositionExactAttachReportsMissingLocalInventory documents the
1074// composition's behavior when the committed broker publication carries no local
1075// session inventory: an exact attach is refused locally as a bounded selection
1076// notice, opens no stream, and never invents a creation or a same-name
1077// fallback. It pins the honest refusal so a future broker-side local catalogue
1078// is the only thing that can make this navigation succeed.
1079func TestTerminalCompositionExactAttachReportsMissingLocalInventory(t *testing.T) {
1080 fixture := startOfflineClientFixture(t)
1081 ctx, cancel := context.WithTimeout(context.Background(), brokerTestWait)
1082 defer cancel()
1083
1084 terminal := newTerminalCompositionTerminal()
1085 notices := make(chan client.LifecycleNotice, 16)
1086 failures := make(chan error, 16)
1087 var lifecycleID domain.SessionLifecycleID
1088 lifecycleID[0], lifecycleID[1] = 3, 0x5a
1089 navigation := client.InitialNavigation{
1090 Kind: client.InitialNavigationAttachExact,
1091 Epoch: terminalCompositionEpoch,
1092 Destination: ports.BrokerEndpointFence{Local: true},
1093 Target: protocol.ExactSessionTarget{LifecycleID: lifecycleID, SessionName: "work"},
1094 }
1095 require.NoError(t, navigation.Validate())
1096
1097 done := make(chan error, 1)
1098 go func() {
1099 done <- runBrokerClient(ctx, brokerClientConfig{
1100 Connector: brokeripc.NewConnector(fixture.socket, brokeripc.Config{}).Navigator(),
1101 Terminal: terminal,
1102 InitialNavigation: navigation,
1103 SessionEnvironment: client.SessionEnvironment{Provenance: client.SessionEnvironmentLocalCLI},
1104 OnFailure: func(err error) {
1105 select {
1106 case failures <- err:
1107 default:
1108 }
1109 },
1110 OnLifecycle: func(notice client.LifecycleNotice) {
1111 select {
1112 case notices <- notice:
1113 default:
1114 }
1115 },
1116 })
1117 }()
1118
1119 select {
1120 case notice := <-notices:
1121 require.Equal(t, client.LifecycleNoticeSelectionUnavailable, notice.Kind)
1122 case <-time.After(brokerTestWait):
1123 t.Fatal("the exact attach reported no lifecycle notice")
1124 }
1125 select {
1126 case failure := <-failures:
1127 require.Error(t, failure)
1128 case <-time.After(brokerTestWait):
1129 t.Fatal("the exact attach reported no failure")
1130 }
1131 // The refused navigation never writes a session frame and never attaches:
1132 // it presents Connecting while pending and ends on the picker.
1133 publishes := terminal.publishes()
1134 for _, status := range publishes {
1135 require.NotEqual(t, ports.UIStatusAttached, status, "a refused navigation never attaches")
1136 }
1137 require.NotEmpty(t, publishes)
1138 require.Equal(t, ports.UIStatusPicker, publishes[len(publishes)-1], "a refused navigation ends on the picker")
1139 cancel()
1140 require.NoError(t, ignoreContextCancellation(<-done))
1141}
1142
1143// awaitTerminalCompositionEpoch waits for one connection's first committed
1144// publication and returns its epoch.
1145func awaitTerminalCompositionEpoch(t *testing.T, service ports.BrokerService) ports.BrokerEpoch {
1146 t.Helper()
1147 sub, err := service.Subscribe()
1148 require.NoError(t, err)
1149 defer sub.Close()
1150 deadline := time.NewTimer(brokerTestWait)
1151 defer deadline.Stop()
1152 for {
1153 if snapshot := service.Snapshot(); snapshot.Epoch != 0 {
1154 return snapshot.Epoch
1155 }
1156 select {
1157 case <-sub.Changed():
1158 case <-service.Done():
1159 t.Fatalf("broker connection ended before publishing a snapshot: %v", service.Err())
1160 case <-deadline.C:
1161 t.Fatal("broker never published a snapshot")
1162 }
1163 }
1164}