1package client
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "io"
8 "strings"
9 "sync"
10 "sync/atomic"
11 "testing"
12 "time"
13
14 "github.com/stretchr/testify/require"
15
16 "github.com/bnema/vev/internal/domain"
17 "github.com/bnema/vev/internal/domain/terminalcap"
18 "github.com/bnema/vev/internal/ports"
19 "github.com/bnema/vev/internal/protocol"
20)
21
22// Committed attachment supervisor tests (Plan 001 P5.3b). They drive the real
23// Run loop over the in-package broker fakes, with a fake clock, no sleeps, and
24// no production wiring.
25
26// attachTestTerminal is the supervisor-owned terminal double: it records output,
27// flushes, raw entrances/restores, and UI publication boundaries so a test can
28// prove exactly what a committed attachment wrote.
29type attachTestTerminal struct {
30 in io.Reader
31
32 mu sync.Mutex
33 out bytes.Buffer
34 flushes int
35 restores int
36 publications []ports.UIContext
37 pending ports.UIContext
38 revision uint64
39 txDepth int
40 txFailed bool
41 available bool
42 successes []bool
43 publishEntered chan struct{}
44 publishRelease <-chan struct{}
45 // publishMatch narrows the publication block to one matching transaction.
46 // A nil predicate blocks every publication, which is what the existing
47 // initial-publication tests rely on.
48 publishMatch func(ports.UIContext) bool
49 // detachedSignals receives one signal per committed Picker publication, so
50 // a test observes attachment release without polling.
51 detachedSignals chan struct{}
52 // events records the transaction order: each begin, each publication, and
53 // each successful drain. A test proves release ordering against it instead
54 // of inferring it from separate slices.
55 events []attachTestEvent
56 changes chan struct{}
57}
58
59// attachTestEvent is one ordered terminal transaction event.
60type attachTestEvent struct {
61 kind string // begin | publish | drain
62 status ports.UIPresentationStatus
63 state uint64
64}
65
66func (t *attachTestTerminal) EnableKittyKeyboard(int) error { return nil }
67
68func (t *attachTestTerminal) EnterRaw() (func() error, error) {
69 return func() error {
70 t.mu.Lock()
71 t.restores++
72 t.mu.Unlock()
73 return nil
74 }, nil
75}
76
77func (t *attachTestTerminal) Geometry() (domain.Geometry, error) {
78 return domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}}, nil
79}
80
81func (t *attachTestTerminal) ResizeEvents() <-chan domain.Geometry { return nil }
82
83func (t *attachTestTerminal) In() io.Reader { return t.in }
84
85func (t *attachTestTerminal) Out() io.Writer { return &attachTestWriter{term: t} }
86
87func (t *attachTestTerminal) Flush() error {
88 t.mu.Lock()
89 defer t.mu.Unlock()
90 t.flushes++
91 return nil
92}
93
94func (t *attachTestTerminal) BeginOutput(ctx ports.UIContext) {
95 t.mu.Lock()
96 if ctx.AttachmentHandle == "" && len(t.publications) != 0 && ctx.Status == ports.UIStatusAttached {
97 ctx.AttachmentHandle = t.publications[len(t.publications)-1].AttachmentHandle
98 }
99 if ctx.Generation == 0 && len(t.publications) != 0 && ctx.Status == ports.UIStatusAttached {
100 ctx.Generation = t.publications[len(t.publications)-1].Generation
101 }
102 if t.txDepth == 0 {
103 t.pending = ctx
104 t.txFailed = false
105 t.events = append(t.events, attachTestEvent{kind: "begin", status: ctx.Status, state: ctx.OutputState})
106 }
107 t.txDepth++
108 t.mu.Unlock()
109}
110
111func (t *attachTestTerminal) EndOutput(success bool) {
112 t.mu.Lock()
113 defer t.mu.Unlock()
114 t.successes = append(t.successes, success)
115 if t.txDepth == 0 {
116 return
117 }
118 if !success {
119 t.txFailed = true
120 }
121 t.txDepth--
122 if t.txDepth != 0 {
123 return
124 }
125 if t.txFailed {
126 t.available = false
127 t.pending = ports.UIContext{}
128 return
129 }
130 t.publications = append(t.publications, t.pending)
131 t.revision++
132 t.events = append(t.events, attachTestEvent{kind: "drain", status: t.pending.Status, state: t.pending.OutputState})
133 t.available = true
134 t.pending = ports.UIContext{}
135}
136
137func (t *attachTestTerminal) PublishContext(uiContext ports.UIContext) error {
138 if t.publishMatch == nil || t.publishMatch(uiContext) {
139 if t.publishEntered != nil {
140 select {
141 case t.publishEntered <- struct{}{}:
142 default:
143 }
144 }
145 if t.publishRelease != nil {
146 <-t.publishRelease
147 }
148 }
149 t.mu.Lock()
150 defer t.mu.Unlock()
151 if !t.available {
152 return ports.ErrUIUnavailable
153 }
154 // The composition terminal applies the published-context shape rule the real
155 // uiterm terminal applies: only an Attached presentation inherits a missing
156 // handle or generation; Picker and Connecting are recorded exactly as stated.
157 if uiContext.Status == ports.UIStatusAttached {
158 if uiContext.AttachmentHandle == "" && len(t.publications) != 0 {
159 uiContext.AttachmentHandle = t.publications[len(t.publications)-1].AttachmentHandle
160 }
161 if uiContext.Generation == 0 && len(t.publications) != 0 {
162 uiContext.Generation = t.publications[len(t.publications)-1].Generation
163 }
164 }
165 t.publications = append(t.publications, uiContext)
166 t.revision++
167 t.events = append(t.events, attachTestEvent{kind: "publish", status: uiContext.Status, state: uiContext.OutputState})
168 if uiContext.Status == ports.UIStatusPicker && t.detachedSignals != nil {
169 select {
170 case t.detachedSignals <- struct{}{}:
171 default:
172 }
173 }
174 return nil
175}
176
177func (t *attachTestTerminal) Snapshot() (ports.UISnapshot, error) {
178 t.mu.Lock()
179 defer t.mu.Unlock()
180 var ctx ports.UIContext
181 if len(t.publications) != 0 {
182 ctx = t.publications[len(t.publications)-1]
183 }
184 if !t.available || t.revision == 0 {
185 return ports.UISnapshot{}, ports.ErrUIUnavailable
186 }
187 return ports.UISnapshot{Revision: t.revision, Context: ctx}, nil
188}
189
190func (t *attachTestTerminal) Changes() <-chan struct{} {
191 t.mu.Lock()
192 defer t.mu.Unlock()
193 if t.changes == nil {
194 t.changes = make(chan struct{})
195 }
196 return t.changes
197}
198
199func (t *attachTestTerminal) publicationList() []ports.UIContext {
200 t.mu.Lock()
201 defer t.mu.Unlock()
202 return append([]ports.UIContext(nil), t.publications...)
203}
204
205// eventIndex reports the position of the first matching transaction event, or
206// -1 when it never happened.
207func (t *attachTestTerminal) eventIndex(kind string, status ports.UIPresentationStatus, state uint64) int {
208 t.mu.Lock()
209 defer t.mu.Unlock()
210 for i, event := range t.events {
211 if event.kind == kind && event.status == status && event.state == state {
212 return i
213 }
214 }
215 return -1
216}
217
218func (t *attachTestTerminal) written() string {
219 t.mu.Lock()
220 defer t.mu.Unlock()
221 return t.out.String()
222}
223
224func (t *attachTestTerminal) counts() (writes int, flushes int, restores int) {
225 t.mu.Lock()
226 defer t.mu.Unlock()
227 return len(t.out.Bytes()), t.flushes, t.restores
228}
229
230type attachTestWriter struct{ term *attachTestTerminal }
231
232func (w *attachTestWriter) Write(data []byte) (int, error) {
233 w.term.mu.Lock()
234 defer w.term.mu.Unlock()
235 return w.term.out.Write(data)
236}
237
238// attachTestReader blocks until a chunk is pushed, then returns EOF once
239// closed. It is the supervisor's single terminal reader.
240type attachTestReader struct {
241 chunks chan []byte
242}
243
244func newAttachTestReader() *attachTestReader {
245 return &attachTestReader{chunks: make(chan []byte, 8)}
246}
247
248func (r *attachTestReader) Read(buffer []byte) (int, error) {
249 chunk, ok := <-r.chunks
250 if !ok {
251 return 0, io.EOF
252 }
253 return copy(buffer, chunk), nil
254}
255
256func (r *attachTestReader) push(data []byte) { r.chunks <- data }
257
258func (r *attachTestReader) close() { close(r.chunks) }
259
260// attachTestPicker is the supervisor's picker seam double. It records input
261// ownership transitions, resolves exactly the committed key to one canned
262// request, and coalesces commit wakeups exactly like the real controller.
263type attachTestPicker struct {
264 mu sync.Mutex
265 opsReady chan struct{}
266 ownsInput bool
267 consumed int
268 // consumedBytes records every byte the picker consumed, in order.
269 consumedBytes []byte
270 op pickerOp
271 key string
272 // resolveErr makes ResolveCommit refuse locally, which is how a test reaches
273 // the local-refusal classification without dialing anything.
274 resolveErr error
275 request ports.BrokerOpenStreamRequest
276 resolved int
277 // consumeOp, when set, is the decision the next consumed read records,
278 // exactly like the controller deciding a commit from an Enter keypress.
279 consumeOp pickerOp
280 consumeRequest ports.BrokerOpenStreamRequest
281 // resolvedStream is the stream identity the last resolution reserved.
282 resolvedStream ports.BrokerStreamID
283 // tab is the tab every resolution names.
284 tab attachmentTab
285}
286
287func newAttachTestPicker() *attachTestPicker {
288 return &attachTestPicker{opsReady: make(chan struct{}, 1), ownsInput: true}
289}
290
291func (p *attachTestPicker) ApplySnapshot(ports.BrokerSnapshot) {}
292
293func (p *attachTestPicker) ConsumeTerminalRead(data []byte) bool {
294 p.mu.Lock()
295 defer p.mu.Unlock()
296 if !p.ownsInput {
297 return false
298 }
299 p.consumed++
300 p.consumedBytes = append(p.consumedBytes, data...)
301 if p.consumeOp != (pickerOp{}) {
302 p.op, p.key, p.request = p.consumeOp, "row", p.consumeRequest
303 p.consumeOp = pickerOp{}
304 select {
305 case p.opsReady <- struct{}{}:
306 default:
307 }
308 }
309 return true
310}
311
312// decideOnConsume scripts the decision the next consumed read records.
313func (p *attachTestPicker) decideOnConsume(op pickerOp, request ports.BrokerOpenStreamRequest) {
314 p.mu.Lock()
315 defer p.mu.Unlock()
316 p.consumeOp, p.consumeRequest = op, request
317}
318
319func (p *attachTestPicker) TakeOp() (pickerOp, string) {
320 p.mu.Lock()
321 defer p.mu.Unlock()
322 op, key := p.op, p.key
323 p.op, p.key = pickerOp{}, ""
324 return op, key
325}
326
327func (p *attachTestPicker) ResolveCommit(key string, base pickerResolveBase) (ports.BrokerOpenStreamRequest, attachmentTab, error) {
328 p.mu.Lock()
329 defer p.mu.Unlock()
330 p.resolved++
331 if p.resolveErr != nil {
332 return ports.BrokerOpenStreamRequest{}, attachmentTab{}, p.resolveErr
333 }
334 request := p.request
335 request.Connection = base.Connection
336 request.Stream = base.Stream
337 p.resolvedStream = base.Stream
338 return request, p.tab, nil
339}
340
341func (p *attachTestPicker) SetOwnsInput(owns bool) {
342 p.mu.Lock()
343 defer p.mu.Unlock()
344 p.ownsInput = owns
345}
346
347func (p *attachTestPicker) OpsReady() <-chan struct{} { return p.opsReady }
348
349func (p *attachTestPicker) commit(request ports.BrokerOpenStreamRequest) {
350 p.mu.Lock()
351 p.op.commit = true
352 p.key = "row"
353 p.request = request
354 p.mu.Unlock()
355 select {
356 case p.opsReady <- struct{}{}:
357 default:
358 }
359}
360
361func (p *attachTestPicker) owns() bool {
362 p.mu.Lock()
363 defer p.mu.Unlock()
364 return p.ownsInput
365}
366
367func (p *attachTestPicker) lastResolvedStream() ports.BrokerStreamID {
368 p.mu.Lock()
369 defer p.mu.Unlock()
370 return p.resolvedStream
371}
372
373func (p *attachTestPicker) consumedCount() int {
374 p.mu.Lock()
375 defer p.mu.Unlock()
376 return p.consumed
377}
378
379func (p *attachTestPicker) resolveCount() int {
380 p.mu.Lock()
381 defer p.mu.Unlock()
382 return p.resolved
383}
384
385// attachTestHarness is one running supervisor with its scripted broker.
386type attachTestHarness struct {
387 sup *Supervisor
388 service *supervisorTestService
389 clock *supervisorTestClock
390 terminal *attachTestTerminal
391 reader *attachTestReader
392 cancel context.CancelFunc
393 runDone chan struct{}
394
395 runMu sync.Mutex
396 runErr error
397}
398
399// startAttachHarness runs one supervisor over a scripted picker and service.
400// The caller installs service.setOpenStream before committing a selection.
401func startAttachHarness(t *testing.T, picker pickerHost) *attachTestHarness {
402 return startAttachHarnessConfig(t, picker, nil)
403}
404
405func startAttachHarnessConfig(t *testing.T, picker pickerHost, configure func(*SupervisorConfig)) *attachTestHarness {
406 t.Helper()
407 ctx, cancel := context.WithCancel(context.Background())
408 clock := newSupervisorTestClock()
409 reader := newAttachTestReader()
410 // available starts true, exactly like a real composition terminal: a
411 // presentation-only publication is observable before any frame drains.
412 terminal := &attachTestTerminal{in: reader, available: true}
413 service := newSupervisorTestService(ports.BrokerConnectionID{1})
414 service.publish(3, 1)
415
416 connector := newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) {
417 return service, nil
418 })
419 cfg := SupervisorConfig{
420 Connector: connector,
421 Terminal: terminal,
422 Clock: clock,
423 Jitter: func() float64 { return 0 },
424 Picker: picker,
425 }
426 if configure != nil {
427 configure(&cfg)
428 }
429 sup := mustSupervisor(t, cfg)
430
431 harness := &attachTestHarness{sup: sup, service: service, clock: clock, terminal: terminal, reader: reader, cancel: cancel, runDone: make(chan struct{})}
432 go func() {
433 err := sup.Run(ctx)
434 harness.runMu.Lock()
435 harness.runErr = err
436 harness.runMu.Unlock()
437 close(harness.runDone)
438 }()
439 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
440
441 t.Cleanup(func() {
442 cancel()
443 select {
444 case <-harness.runDone:
445 case <-time.After(5 * time.Second):
446 t.Error("the supervisor did not stop")
447 }
448 })
449 return harness
450}
451
452// waitRun joins the supervisor and returns its terminal error.
453func (h *attachTestHarness) waitRun(t *testing.T) error {
454 t.Helper()
455 select {
456 case <-h.runDone:
457 case <-time.After(5 * time.Second):
458 t.Fatal("the supervisor did not stop")
459 }
460 h.runMu.Lock()
461 defer h.runMu.Unlock()
462 return h.runErr
463}
464
465// snapshotTimers returns the armed timers for assertions without racing the
466// clock's own bookkeeping.
467func (c *supervisorTestClock) snapshotTimers() []*supervisorTestTimer {
468 c.mu.Lock()
469 defer c.mu.Unlock()
470 return append([]*supervisorTestTimer(nil), c.timers...)
471}
472
473// awaitAttachedState waits until the supervisor presents the attached state.
474func awaitAttachedState(t *testing.T, sup *Supervisor) {
475 t.Helper()
476 require.Eventually(t, func() bool { return sup.State().Presentation == PresentAttached }, 5*time.Second, time.Millisecond)
477}
478
479// authorityState reports the current foreground slot state without racing the
480// supervisor's own transitions.
481func (a *attachmentAuthority) authorityState() attachmentAuthorityState {
482 a.mu.Lock()
483 defer a.mu.Unlock()
484 return a.state
485}
486
487// awaitPickerState waits until the supervisor is back in the picker.
488func awaitPickerState(t *testing.T, sup *Supervisor) {
489 t.Helper()
490 require.Eventually(t, func() bool { return sup.State().Presentation == PresentPicker }, 5*time.Second, time.Millisecond)
491}
492
493// TestSupervisorDetachedReasonDeterminesExitOrPicker pins the client-visible
494// consequence of every protocol.Detached reason the daemon can send while
495// attached: only the explicit user detach (ReasonDetach) ends the run and
496// exits the client cleanly. Every other daemon-driven reason returns to the
497// picker with a lifecycle notice instead of leaving the user stranded with no
498// way out short of Ctrl+C.
499func TestSupervisorDetachedReasonDeterminesExitOrPicker(t *testing.T) {
500 tests := []struct {
501 name string
502 reason uint8
503 wantExit bool
504 wantNotice LifecycleNoticeKind
505 }{
506 {
507 name: "explicit detach exits the client",
508 reason: protocol.ReasonDetach,
509 wantExit: true,
510 wantNotice: LifecycleNoticeDetachAndExit,
511 },
512 {
513 name: "detach to picker stays in the client",
514 reason: protocol.ReasonDetachToPicker,
515 wantExit: false,
516 wantNotice: LifecycleNoticeDetachToPicker,
517 },
518 {
519 name: "session killed returns to the picker",
520 reason: protocol.ReasonSessionKilled,
521 wantExit: false,
522 wantNotice: LifecycleNoticeSessionEnded,
523 },
524 {
525 name: "server shutdown returns to the picker",
526 reason: protocol.ReasonServerShutdown,
527 wantExit: false,
528 wantNotice: LifecycleNoticeSessionEnded,
529 },
530 {
531 name: "replaced returns to the picker",
532 reason: protocol.ReasonReplaced,
533 wantExit: false,
534 wantNotice: LifecycleNoticeSessionEnded,
535 },
536 }
537 for _, tt := range tests {
538 t.Run(tt.name, func(t *testing.T) {
539 picker := newAttachTestPicker()
540 notices := make(chan LifecycleNotice, 8)
541 harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) {
542 cfg.NotifyLifecycle = func(notice LifecycleNotice) { notices <- notice }
543 })
544 stream := attachLiveSession(t, harness, picker)
545 stream.deliver(protocol.Detached{Reason: tt.reason})
546
547 if tt.wantExit {
548 require.NoError(t, harness.waitRun(t))
549 } else {
550 awaitPickerState(t, harness.sup)
551 }
552 select {
553 case notice := <-notices:
554 require.Equal(t, tt.wantNotice, notice.Kind)
555 case <-time.After(5 * time.Second):
556 t.Fatal("the expected lifecycle notice was never delivered")
557 }
558 })
559 }
560}
561
562// TestSupervisorPendingSwapContinuesRatherThanExiting guards the overlay
563// commit-elsewhere path (same-peer switch) against being folded into the
564// explicit-detach-and-exit outcome above: the outgoing attachment ends
565// through a local lifecycle action, never a protocol.Detached message, and a
566// pendingSwap in flight must land the supervisor on the new target instead of
567// exiting or surfacing any lifecycle notice.
568func TestSupervisorPendingSwapContinuesRatherThanExiting(t *testing.T) {
569 picker := newAttachTestPicker()
570 notices := make(chan LifecycleNotice, 8)
571 harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) {
572 cfg.NotifyLifecycle = func(notice LifecycleNotice) { notices <- notice }
573 })
574 first := attachLiveSession(t, harness, picker)
575 first.deliver(navigationOffer(1))
576 awaitPresentation(t, harness.sup, PresentAttachedPicker)
577
578 var mu sync.Mutex
579 second := newSessionTestStream()
580 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
581 return second, nil
582 })
583 other := sessionTestRequest(true)
584 other.Target = protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{2}, SessionName: "beta"}
585 picker.recordOp(pickerOp{commit: true}, "row", other)
586
587 awaitSent(t, first, "Detach", isSent[protocol.Detach])
588 awaitStreamHello(t, &mu, &second)
589 second.deliver(protocol.Welcome{SessionName: "beta"})
590 output := sessionTestOutput(1, "\x1b[Hbeta")
591 output.Context.Route.Target = other.Target
592 second.deliver(output)
593 awaitAttachedState(t, harness.sup)
594
595 select {
596 case <-harness.runDone:
597 t.Fatal("a same-peer swap must not end the supervisor run")
598 default:
599 }
600 select {
601 case notice := <-notices:
602 t.Fatalf("a swap must not report a lifecycle transition: %+v", notice)
603 default:
604 }
605}
606
607// TestSupervisorSwapDropsKeysKeptForPreviousSession types into a session whose
608// link stalls, then swaps to another session: the stalled keys were meant for
609// the first session and must never run in the second.
610func TestSupervisorSwapDropsKeysKeptForPreviousSession(t *testing.T) {
611 picker := newAttachTestPicker()
612 harness := startAttachHarness(t, picker)
613 base := newSessionTestStream()
614 first := &stallingSessionStream{sessionTestStream: base, entered: make(chan struct{}, 1), release: make(chan struct{})}
615 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
616 return first, nil
617 })
618 picker.commit(sessionTestRequest(true))
619 deliverReadyStream(t, base)
620 awaitAttachedState(t, harness.sup)
621 require.Eventually(t, func() bool { return len(base.messages()) >= 2 }, 5*time.Second, time.Millisecond)
622 first.stall.Store(true)
623 harness.reader.push([]byte("rm -rf build\r"))
624 <-first.entered
625
626 var mu sync.Mutex
627 second := newSessionTestStream()
628 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
629 return second, nil
630 })
631 base.deliver(navigationOffer(1))
632 awaitPresentation(t, harness.sup, PresentAttachedPicker)
633 other := sessionTestRequest(true)
634 other.Target = protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{2}, SessionName: "beta"}
635 picker.recordOp(pickerOp{commit: true}, "row", other)
636 // The Detach queues behind the stalled write; let the bounded drain expire
637 // so the unsent keys are preserved, as on a dead link.
638 for {
639 timer := harness.clock.awaitTimer(t)
640 if timer.delay == drainOutboxTimeout {
641 timer.fire()
642 break
643 }
644 }
645 awaitStreamHello(t, &mu, &second)
646 second.deliver(protocol.Welcome{SessionName: "beta"})
647 output := sessionTestOutput(1, "\x1b[Hbeta")
648 output.Context.Route.Target = other.Target
649 second.deliver(output)
650 awaitAttachedState(t, harness.sup)
651
652 harness.reader.push([]byte("z"))
653 require.Eventually(t, func() bool { return sentInput(second) == "z" }, 5*time.Second, time.Millisecond,
654 "second session input = %q", sentInput(second))
655}
656
657// awaitStreamHello waits until the admitted stream has sent its first client
658// message, polling the guarded stream pointer rather than a stale snapshot.
659func awaitStreamHello(t *testing.T, mu *sync.Mutex, stream **sessionTestStream) {
660 t.Helper()
661 require.Eventually(t, func() bool {
662 mu.Lock()
663 defer mu.Unlock()
664 return *stream != nil && len((*stream).messages()) >= 1
665 }, 5*time.Second, time.Millisecond)
666}
667
668// deliverReadyStream scripts Welcome plus one committed initial publication.
669func deliverReadyStream(t *testing.T, stream *sessionTestStream) {
670 t.Helper()
671 stream.deliver(protocol.Welcome{SessionName: "alpha"})
672 stream.deliver(sessionTestOutput(1, "\x1b[Hready"))
673}
674
675// TestSupervisorReducerAttachmentPresentation pins the presentation
676// transitions of one attachment: connecting until the committed publication,
677// attached, then back to the picker without disturbing connectivity.
678func TestSupervisorReducerAttachmentPresentation(t *testing.T) {
679 ready := State{Presentation: PresentPicker, Connectivity: ConnectivityReady, Generation: 2}
680 connecting := reduceSupervisor(ready, supervisorEvent{kind: supervisorAttachBegin})
681 require.Equal(t, PresentConnecting, connecting.Presentation)
682 require.Equal(t, ConnectivityReady, connecting.Connectivity)
683 require.Zero(t, connecting.Attempt)
684
685 attached := reduceSupervisor(connecting, supervisorEvent{kind: supervisorAttached})
686 require.Equal(t, PresentAttached, attached.Presentation)
687 require.Equal(t, ConnectivityReady, attached.Connectivity)
688
689 returned := reduceSupervisor(attached, supervisorEvent{kind: supervisorAttachEnded, err: errAttachmentDeadline})
690 require.Equal(t, PresentPicker, returned.Presentation)
691 require.Equal(t, ConnectivityReady, returned.Connectivity)
692 require.ErrorIs(t, returned.Err, errAttachmentDeadline)
693}
694
695// TestSupervisorAttachmentLocalRemoteParity proves the same supervisor path
696// opens the exact resolved request for a local and a remote selection, with no
697// dialer choice in the client.
698func TestSupervisorUIActionBindingUsesSingleAttachmentPump(t *testing.T) {
699 picker := newAttachTestPicker()
700 var stream *sessionTestStream
701 var streamMu sync.Mutex
702 var ui *UI
703 harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) {
704 ui = NewUI(cfg.Terminal.(ports.UIState), cfg.Clock)
705 cfg.UI = ui
706 })
707 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
708 admitted := newSessionTestStream()
709 streamMu.Lock()
710 stream = admitted
711 streamMu.Unlock()
712 return admitted, nil
713 })
714
715 // Picker ownership has no UI binding, so this would fail if actions could
716 // collide with or bypass the picker's exclusive input claim.
717 _, err := ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: 1, Text: "picker"})
718 var pickerErr *ports.UIError
719 require.ErrorAs(t, err, &pickerErr)
720 require.Equal(t, ports.UIErrUnavailable, pickerErr.Code)
721
722 picker.commit(sessionTestRequest(true))
723 require.Eventually(t, func() bool {
724 streamMu.Lock()
725 defer streamMu.Unlock()
726 return stream != nil && len(stream.messages()) > 0
727 }, 5*time.Second, time.Millisecond)
728 streamMu.Lock()
729 admitted := stream
730 streamMu.Unlock()
731 deliverReadyStream(t, admitted)
732 awaitAttachedState(t, harness.sup)
733 attachedOutput := sessionTestOutput(2, "\x1b[Hattached")
734 attachedOutput.Full = false
735 attachedOutput.Base = 1
736 attachedOutput.New = 2
737 admitted.deliver(attachedOutput)
738
739 var snapshot ports.UISnapshot
740 require.Eventually(t, func() bool {
741 var captureErr error
742 snapshot, captureErr = ui.Capture(ui.Handle())
743 return captureErr == nil && snapshot.Context.Status == ports.UIStatusAttached
744 }, 5*time.Second, time.Millisecond)
745 require.Equal(t, ports.UIStatusAttached, snapshot.Context.Status)
746 require.NotZero(t, snapshot.Context.Generation)
747 actionDone := make(chan error, 1)
748 go func() {
749 _, actionErr := ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: snapshot.Context.Generation, Text: "automated"})
750 actionDone <- actionErr
751 }()
752 var actionID uint64
753 require.Eventually(t, func() bool {
754 var fenced bool
755 for _, message := range admitted.messages() {
756 switch typed := message.(type) {
757 case protocol.Input:
758 if string(typed.Data) == "automated" {
759 actionID = typed.ActionID
760 }
761 case protocol.UIFence:
762 fenced = actionID != 0 && typed.ActionID == actionID
763 }
764 }
765 return fenced
766 }, 5*time.Second, time.Millisecond, "admitted actions must be followed by their UI fence")
767 processedOutput := sessionTestOutput(3, "\x1b[Hprocessed")
768 processedOutput.Full = false
769 processedOutput.Base = 2
770 processedOutput.New = 3
771 processedOutput.Context.Publication = 3
772 admitted.deliver(processedOutput)
773 var boundary ports.UIActionResult
774 require.Eventually(t, func() bool {
775 ui.mu.Lock()
776 boundary = ui.boundary
777 ui.mu.Unlock()
778 return boundary.Revision != 0 && boundary.Context.OutputState == processedOutput.New && boundary.Context.ViewPublication == processedOutput.Context.Publication
779 }, 5*time.Second, time.Millisecond, "the action-specific output must be the committed processing boundary")
780 admitted.deliver(protocol.UIReceipt{ActionID: actionID, Epoch: 1, State: 3, ViewPublication: 3, Outcome: protocol.UIReceiptProcessed})
781 select {
782 case actionErr := <-actionDone:
783 require.NoError(t, actionErr, "receipt routing must complete the admitted action as processed")
784 case <-time.After(5 * time.Second):
785 t.Fatal("processed UI action did not complete")
786 }
787
788 // Ending the stream releases before picker reacquisition. A stale request
789 // would be admitted here if finalization forgot to retire the binding.
790 admitted.deliver(protocol.Detached{Reason: protocol.ReasonSessionKilled})
791 awaitPickerState(t, harness.sup)
792 ui.mu.Lock()
793 require.Nil(t, ui.input, "attachment finalization must revoke the UI pump binding")
794 require.Zero(t, ui.consumer, "attachment finalization must revoke the UI consumer")
795 require.Nil(t, ui.foreground, "attachment finalization must retire the UI foreground")
796 ui.mu.Unlock()
797 _, err = ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: snapshot.Context.Generation, Text: "stale"})
798 var staleErr *ports.UIError
799 require.ErrorAs(t, err, &staleErr)
800 require.Equal(t, ports.UIErrUnavailable, staleErr.Code)
801
802 // Reconnecting installs a fresh UI-owned generation. The prior generation
803 // remains stale even while another attachment is actionable.
804 reconnected := newSessionTestStream()
805 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
806 return reconnected, nil
807 })
808 picker.commit(sessionTestRequest(true))
809 awaitHello(t, reconnected)
810 deliverReadyStream(t, reconnected)
811 awaitAttachedState(t, harness.sup)
812 reconnectedOutput := sessionTestOutput(2, "\x1b[Hreconnected")
813 reconnectedOutput.Full = false
814 reconnectedOutput.Base = 1
815 reconnectedOutput.New = 2
816 reconnected.deliver(reconnectedOutput)
817 require.Eventually(t, func() bool {
818 current, captureErr := ui.Capture(ui.Handle())
819 return captureErr == nil && current.Context.Status == ports.UIStatusAttached && current.Context.Generation != snapshot.Context.Generation
820 }, 5*time.Second, time.Millisecond)
821 _, err = ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: snapshot.Context.Generation, Text: "still-stale"})
822 staleErr = nil
823 require.ErrorAs(t, err, &staleErr)
824 require.Equal(t, ports.UIErrStaleAttachment, staleErr.Code)
825 select {
826 case <-actionDone: // completion may be receipt-driven or retired on release.
827 default:
828 }
829}
830
831// TestSupervisorUIActionReleaseWaitsForInFlightOutput pins GO-201: finalization
832// must not release the UI foreground binding while a foreground output
833// transaction is in flight. The release publishes Picker, so releasing it
834// early would publish an unattached presentation while a newer session frame
835// was still being written, and would return the picker claim before the last
836// frame committed. The mutation this test pins moves releaseForeground before
837// lease.stop in attachmentForeground.finalize.
838func TestSupervisorUIActionReleaseWaitsForInFlightOutput(t *testing.T) {
839 const inFlightState = 2
840 picker := newAttachTestPicker()
841 var (
842 streamMu sync.Mutex
843 stream *sessionTestStream
844 ui *UI
845 )
846 harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) {
847 ui = NewUI(cfg.Terminal.(ports.UIState), cfg.Clock)
848 cfg.UI = ui
849 })
850 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
851 admitted := newSessionTestStream()
852 streamMu.Lock()
853 stream = admitted
854 streamMu.Unlock()
855 return admitted, nil
856 })
857
858 // Installed before the commit, so the supervisor's publication goroutine
859 // always observes them: only the attached frame with inFlightState blocks.
860 terminal := harness.terminal
861 entered := make(chan struct{}, 1)
862 release := make(chan struct{})
863 var releaseOnce sync.Once
864 releaseOutput := func() { releaseOnce.Do(func() { close(release) }) }
865 defer releaseOutput()
866 detached := make(chan struct{}, 4)
867 terminal.publishMatch = func(ctx ports.UIContext) bool {
868 return ctx.Status == ports.UIStatusAttached && ctx.OutputState == inFlightState
869 }
870 terminal.publishEntered = entered
871 terminal.publishRelease = release
872 terminal.detachedSignals = detached
873
874 picker.commit(sessionTestRequest(true))
875 awaitStreamHello(t, &streamMu, &stream)
876 streamMu.Lock()
877 admitted := stream
878 streamMu.Unlock()
879 deliverReadyStream(t, admitted)
880 awaitAttachedState(t, harness.sup)
881
882 // Block one foreground output transaction after BeginOutput and before its
883 // commit, so the send lease is held while finalization begins.
884 inFlight := sessionTestOutput(inFlightState, "\x1b[Hin-flight")
885 inFlight.Full = false
886 inFlight.Base = 1
887 inFlight.New = inFlightState
888 admitted.deliver(inFlight)
889 select {
890 case <-entered:
891 case <-time.After(5 * time.Second):
892 t.Fatal("the foreground output transaction never reached its publication boundary")
893 }
894 require.Equal(t, -1, terminal.eventIndex("drain", ports.UIStatusAttached, inFlightState), "the in-flight frame must still hold its transaction")
895
896 // Cancel the attachment while that transaction is in flight.
897 harness.service.lose(errors.New("attachment lost"))
898 require.Eventually(t, func() bool {
899 return harness.sup.attachments.authority.authorityState() == authorityFinalizing
900 }, 5*time.Second, time.Millisecond, "attachment finalization must begin while the frame is in flight")
901
902 // Finalization cannot complete, and Picker cannot be published, until the
903 // in-flight transaction drains.
904 require.Never(t, func() bool {
905 select {
906 case <-detached:
907 return true
908 default:
909 return false
910 }
911 }, 50*time.Millisecond, time.Millisecond, "Picker must not publish while a foreground transaction is in flight")
912 require.Equal(t, authorityFinalizing, harness.sup.attachments.authority.authorityState(), "finalize still waits on the output transaction")
913 require.False(t, admitted.closedNow(), "the retired stream is released only after the drain")
914 require.False(t, picker.owns(), "the picker claim returns only after the drain")
915 require.Equal(t, PresentAttached, harness.sup.State().Presentation, "the attachment is still presented while its last frame drains")
916 ui.mu.Lock()
917 require.NotNil(t, ui.input, "the UI foreground binding is released only after the drain")
918 require.NotNil(t, ui.foreground, "the UI foreground context is retired only after the drain")
919 ui.mu.Unlock()
920
921 releaseOutput()
922 select {
923 case <-detached:
924 case <-time.After(5 * time.Second):
925 t.Fatal("attachment finalization never published Picker")
926 }
927
928 // The newer frame commits before Picker, and Picker carries no committed
929 // output boundary: an unattached presentation never presents session
930 // metadata.
931 drained := terminal.eventIndex("drain", ports.UIStatusAttached, inFlightState)
932 published := terminal.eventIndex("publish", ports.UIStatusPicker, 0)
933 require.GreaterOrEqual(t, drained, 0, "the in-flight frame must commit once it drains")
934 require.Greater(t, published, drained, "Picker must be published only after the newer frame commits")
935 publications := terminal.publicationList()
936 require.GreaterOrEqual(t, len(publications), 2)
937 last := publications[len(publications)-1]
938 require.Equal(t, ports.UIStatusPicker, last.Status)
939 require.Zero(t, last.OutputState, "Picker carries no committed output boundary")
940 prior := publications[len(publications)-2]
941 require.Equal(t, ports.UIStatusAttached, prior.Status)
942 require.Equal(t, uint64(inFlightState), prior.OutputState)
943
944 // Finalization completes and hands the picker claim back.
945 require.Eventually(t, func() bool {
946 ui.mu.Lock()
947 released := ui.input == nil && ui.consumer == 0 && ui.foreground == nil
948 ui.mu.Unlock()
949 return released && picker.owns() && harness.sup.attachments.authority.authorityState() == authorityIdle && admitted.closedNow()
950 }, 5*time.Second, time.Millisecond, "finalization must complete and return the picker claim")
951 require.Equal(t, PresentPicker, harness.sup.State().Presentation)
952}
953
954func TestSupervisorUIViewUpdateCommitsActionBoundaryWithoutBytes(t *testing.T) {
955 picker := newAttachTestPicker()
956 var ui *UI
957 harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) {
958 ui = NewUI(cfg.Terminal.(ports.UIState), cfg.Clock)
959 cfg.UI = ui
960 })
961 stream := newSessionTestStream()
962 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
963 return stream, nil
964 })
965 picker.commit(sessionTestRequest(true))
966 awaitHello(t, stream)
967 deliverReadyStream(t, stream)
968 awaitAttachedState(t, harness.sup)
969 initialOutput := sessionTestOutput(2, "\x1b[Hmetadata-base")
970 initialOutput.Full = false
971 initialOutput.Base = 1
972 initialOutput.New = 2
973 initialOutput.Context.Publication = 2
974 stream.deliver(initialOutput)
975
976 var snapshot ports.UISnapshot
977 require.Eventually(t, func() bool {
978 var err error
979 snapshot, err = ui.Capture(ui.Handle())
980 return err == nil && snapshot.Context.Status == ports.UIStatusAttached && snapshot.Context.OutputState == 2
981 }, 5*time.Second, time.Millisecond)
982 actionDone := make(chan error, 1)
983 go func() {
984 _, err := ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: snapshot.Context.Generation, Text: "metadata"})
985 actionDone <- err
986 }()
987 var actionID uint64
988 require.Eventually(t, func() bool {
989 for _, message := range stream.messages() {
990 if fence, ok := message.(protocol.UIFence); ok {
991 actionID = fence.ActionID
992 return actionID != 0
993 }
994 }
995 return false
996 }, 5*time.Second, time.Millisecond)
997 view := protocol.ViewContext{Publication: 3, Route: snapshot.Context.Route, TabID: "metadata-tab", FocusedPaneID: "metadata-pane"}
998 view.Route.Target = protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "alpha"}
999 stream.deliver(protocol.UIViewUpdate{Epoch: 1, State: 2, Context: view})
1000 require.Eventually(t, func() bool {
1001 current, err := ui.Capture(ui.Handle())
1002 return err == nil && current.Context.ViewPublication == 3 && current.Context.TabID == "metadata-tab"
1003 }, 5*time.Second, time.Millisecond, "metadata-only update must commit through the foreground transaction")
1004 stream.deliver(protocol.UIReceipt{ActionID: actionID, Epoch: 1, State: 2, ViewPublication: 3, Outcome: protocol.UIReceiptProcessed})
1005 select {
1006 case err := <-actionDone:
1007 require.NoError(t, err)
1008 case <-time.After(5 * time.Second):
1009 t.Fatal("metadata-only committed boundary did not complete action")
1010 }
1011}
1012
1013func TestSupervisorUIViewUpdateRequestsOneCoalescedReset(t *testing.T) {
1014 picker := newAttachTestPicker()
1015 harness := startAttachHarness(t, picker)
1016 stream := newSessionTestStream()
1017 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1018 return stream, nil
1019 })
1020 picker.commit(sessionTestRequest(true))
1021 awaitHello(t, stream)
1022 deliverReadyStream(t, stream)
1023 awaitAttachedState(t, harness.sup)
1024 bad := protocol.UIViewUpdate{Epoch: 9, State: 9, Context: protocol.ViewContext{Publication: 2, Route: protocol.CommittedRouteIdentity{Target: protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "alpha"}}, TabID: "future", FocusedPaneID: "future"}}
1025 stream.deliver(bad)
1026 stream.deliver(bad)
1027 require.Eventually(t, func() bool {
1028 count := 0
1029 for _, message := range stream.messages() {
1030 if _, ok := message.(protocol.OutputResetRequest); ok {
1031 count++
1032 }
1033 }
1034 return count == 1
1035 }, 5*time.Second, time.Millisecond, "future view dependencies must request one coalesced reset")
1036}
1037
1038func TestSupervisorWithoutUIDoesNotBindActions(t *testing.T) {
1039 picker := newAttachTestPicker()
1040 harness := startAttachHarness(t, picker)
1041 require.Nil(t, harness.sup.cfg.UI, "the optional seam must preserve publication-only composition")
1042 require.Nil(t, harness.sup.attachments.actionUI, "mutation that binds an implicit UI changes no-UI behavior")
1043}
1044
1045func TestUIActionWithoutInputReturnsUnavailable(t *testing.T) {
1046 ui := NewUI(&attachTestTerminal{}, newSupervisorTestClock())
1047 _, err := ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: 1, Text: "x"})
1048 var uiErr *ports.UIError
1049 require.ErrorAs(t, err, &uiErr)
1050 require.Equal(t, ports.UIErrUnavailable, uiErr.Code)
1051}
1052
1053func TestSupervisorAttachmentLocalRemoteParity(t *testing.T) {
1054 for _, local := range []bool{true, false} {
1055 name := "remote"
1056 if local {
1057 name = "local"
1058 }
1059 t.Run(name, func(t *testing.T) {
1060 picker := newAttachTestPicker()
1061 harness := startAttachHarness(t, picker)
1062 var (
1063 streamMu sync.Mutex
1064 stream *sessionTestStream
1065 )
1066 harness.service.setOpenStream(func(_ context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1067 admitted := newSessionTestStream()
1068 streamMu.Lock()
1069 stream = admitted
1070 streamMu.Unlock()
1071 return admitted, nil
1072 })
1073 picker.commit(sessionTestRequest(local))
1074
1075 require.Eventually(t, func() bool { return len(harness.service.openedRequests()) == 1 }, 5*time.Second, time.Millisecond)
1076 opened := harness.service.openedRequests()[0]
1077 require.Equal(t, local, opened.Local)
1078 require.Equal(t, ports.BrokerStreamAttachment, opened.Purpose)
1079 require.Equal(t, ports.BrokerAdmissionExact, opened.Admission)
1080 require.Equal(t, "alpha", opened.Target.SessionName)
1081 require.Equal(t, ports.BrokerConnectionID{1}, opened.Connection)
1082 require.Equal(t, ports.BrokerStreamID(1), opened.Stream)
1083 if !local {
1084 require.Equal(t, "user@example", opened.Endpoint)
1085 }
1086
1087 // The stream now runs the typed protocol and commits the initial
1088 // publication before attachment is reported.
1089 streamMu.Lock()
1090 admitted := stream
1091 streamMu.Unlock()
1092 require.NotNil(t, admitted)
1093 require.Eventually(t, func() bool { return len(admitted.messages()) >= 1 }, 5*time.Second, time.Millisecond)
1094 require.Equal(t, PresentConnecting, harness.sup.State().Presentation, "connecting until the publication commits")
1095 require.False(t, picker.owns(), "the picker released input for the attachment")
1096 deliverReadyStream(t, admitted)
1097 awaitAttachedState(t, harness.sup)
1098 // The attached worker's palette query draws nothing; strip it.
1099 require.Equal(t, "\x1b[Hready", strings.ReplaceAll(harness.terminal.written(), paletteColorBatch, ""))
1100 // Two transactions commit on this attachment: the initial frame under the
1101 // pre-attach Connecting presentation, then the committed Attached
1102 // presentation with no bytes of its own.
1103 require.Equal(t, []bool{true, true}, harness.terminal.successes, "both publications committed through the UI transaction")
1104 publications := harness.terminal.publicationList()
1105 require.Equal(t, ports.UIStatusConnecting, publications[0].Status)
1106 require.Equal(t, ports.UIStatusAttached, publications[len(publications)-1].Status)
1107 })
1108 }
1109}
1110
1111// TestSupervisorAttachmentDeadlineElapsedDuringOpenFailsWithoutRestart proves a
1112// deadline that passed during stream open fails the attachment immediately
1113// instead of starting a second budget.
1114func TestSupervisorAttachmentDeadlineElapsedDuringOpenFailsWithoutRestart(t *testing.T) {
1115 picker := newAttachTestPicker()
1116 var (
1117 streamMu sync.Mutex
1118 stream *sessionTestStream
1119 )
1120 harness := startAttachHarness(t, picker)
1121 clock := harness.clock
1122 harness.service.setOpenStream(func(_ context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1123 // The accepted absolute deadline elapsed before this client saw it.
1124 admitted := &sessionTestStreamWithDeadline{
1125 sessionTestStream: newSessionTestStream(),
1126 deadline: clock.Now().Add(-time.Second),
1127 }
1128 streamMu.Lock()
1129 stream = admitted.sessionTestStream
1130 streamMu.Unlock()
1131 return admitted, nil
1132 })
1133 picker.commit(sessionTestRequest(true))
1134
1135 awaitPickerState(t, harness.sup)
1136 require.Eventually(t, func() bool { return harness.sup.State().Err != nil }, 5*time.Second, time.Millisecond)
1137 var typed ports.BrokerError
1138 require.ErrorAs(t, harness.sup.State().Err, &typed)
1139 require.Equal(t, ports.BrokerErrorTimeout, typed.Code)
1140
1141 streamMu.Lock()
1142 require.NotNil(t, stream)
1143 require.True(t, stream.closedNow(), "an expired stream is closed")
1144 streamMu.Unlock()
1145
1146 // Exactly one deadline timer was ever armed, and it was stopped: the
1147 // elapsed accepted deadline never restarted a fresh budget.
1148 require.Equal(t, 1, harness.clock.timerCount(), "the elapsed deadline must not restart a budget")
1149 require.True(t, harness.clock.allStopped(), "the pre-open budget is retired")
1150}
1151
1152// TestSupervisorAttachmentDeadlineNotExtendedAcrossStages proves one absolute
1153// deadline spans the whole attachment: adopting the accepted deadline arms
1154// exactly its remaining time, and crossing Welcome never restarts it.
1155func TestSupervisorAttachmentDeadlineNotExtendedAcrossStages(t *testing.T) {
1156 picker := newAttachTestPicker()
1157 var (
1158 streamMu sync.Mutex
1159 stream *sessionTestStream
1160 )
1161 harness := startAttachHarness(t, picker)
1162 clock := harness.clock
1163 harness.service.setOpenStream(func(_ context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1164 admitted := &sessionTestStreamWithDeadline{
1165 sessionTestStream: newSessionTestStream(),
1166 deadline: clock.Now().Add(5 * time.Second),
1167 }
1168 streamMu.Lock()
1169 stream = admitted.sessionTestStream
1170 streamMu.Unlock()
1171 return admitted, nil
1172 })
1173 picker.commit(sessionTestRequest(true))
1174
1175 awaitStreamHello(t, &streamMu, &stream)
1176
1177 streamMu.Lock()
1178 admitted := stream
1179 streamMu.Unlock()
1180
1181 // Wait until adoption armed the second timer.
1182 require.Eventually(t, func() bool { return harness.clock.timerCount() == 2 }, 5*time.Second, time.Millisecond)
1183 timers := harness.clock.snapshotTimers()
1184 require.Equal(t, protocol.HandshakeTimeout, timers[0].delay, "the pre-open budget bounds stream open")
1185 require.Equal(t, 5*time.Second, timers[1].delay, "the accepted deadline is adopted verbatim, never restarted")
1186 require.True(t, timers[0].stopped(), "the pre-open budget is replaced by the accepted deadline")
1187 require.False(t, timers[1].stopped())
1188
1189 // Crossing Welcome does not arm another deadline.
1190 admitted.deliver(protocol.Welcome{SessionName: "alpha"})
1191 require.Never(t, func() bool { return harness.clock.timerCount() > 2 }, 20*time.Millisecond, 2*time.Millisecond, "no stage may restart the deadline")
1192
1193 // Firing the one adopted deadline ends the attachment as a typed timeout
1194 // before publication, without claiming attachment.
1195 timers[1].fire()
1196 awaitPickerState(t, harness.sup)
1197 require.Eventually(t, func() bool { return harness.sup.State().Err != nil }, 5*time.Second, time.Millisecond)
1198 var typed ports.BrokerError
1199 require.ErrorAs(t, harness.sup.State().Err, &typed)
1200 require.Equal(t, ports.BrokerErrorTimeout, typed.Code)
1201 require.Empty(t, harness.terminal.publications, "a timed-out attachment never published")
1202}
1203
1204// TestSupervisorAttachmentLaterAcceptedDeadlineNeverExtendsBudget proves the
1205// one absolute deadline is the earlier of the client's own pre-open budget and
1206// the admitted stream's accepted deadline. The accepted deadline is anchored
1207// before this client's attempt began, so it is normally the later of the two;
1208// folding it in verbatim would repay the open stage with a fresh budget and let
1209// the attachment outlive the deadline.
1210func TestSupervisorAttachmentLaterAcceptedDeadlineNeverExtendsBudget(t *testing.T) {
1211 picker := newAttachTestPicker()
1212 var (
1213 streamMu sync.Mutex
1214 stream *sessionTestStream
1215 )
1216 harness := startAttachHarness(t, picker)
1217 clock := harness.clock
1218 harness.service.setOpenStream(func(_ context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1219 admitted := &sessionTestStreamWithDeadline{
1220 sessionTestStream: newSessionTestStream(),
1221 deadline: clock.Now().Add(4 * protocol.HandshakeTimeout),
1222 }
1223 streamMu.Lock()
1224 stream = admitted.sessionTestStream
1225 streamMu.Unlock()
1226 return admitted, nil
1227 })
1228 picker.commit(sessionTestRequest(true))
1229
1230 awaitStreamHello(t, &streamMu, &stream)
1231
1232 require.Eventually(t, func() bool { return harness.clock.timerCount() == 2 }, 5*time.Second, time.Millisecond)
1233 timers := harness.clock.snapshotTimers()
1234 require.Equal(t, protocol.HandshakeTimeout, timers[0].delay, "the pre-open budget bounds stream open")
1235 require.Equal(t, protocol.HandshakeTimeout, timers[1].delay,
1236 "a later accepted deadline never repays the open stage with a second budget")
1237 require.True(t, timers[0].stopped())
1238 require.False(t, timers[1].stopped())
1239
1240 // The one deadline still ends the attachment as a typed timeout before
1241 // publication, and it does so no later than the client's own budget.
1242 timers[1].fire()
1243 awaitPickerState(t, harness.sup)
1244 require.Eventually(t, func() bool { return harness.sup.State().Err != nil }, 5*time.Second, time.Millisecond)
1245 var typed ports.BrokerError
1246 require.ErrorAs(t, harness.sup.State().Err, &typed)
1247 require.Equal(t, ports.BrokerErrorTimeout, typed.Code)
1248 require.Empty(t, harness.terminal.publications, "a timed-out attachment never published")
1249}
1250
1251// TestSupervisorAttachmentPreOpenBudgetWithoutProvider proves a stream without
1252// the optional seam keeps the one pre-open budget: no second timer, no restart.
1253func TestSupervisorAttachmentPreOpenBudgetWithoutProvider(t *testing.T) {
1254 picker := newAttachTestPicker()
1255 var (
1256 streamMu sync.Mutex
1257 stream *sessionTestStream
1258 )
1259 harness := startAttachHarness(t, picker)
1260 harness.service.setOpenStream(func(_ context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1261 admitted := newSessionTestStream()
1262 streamMu.Lock()
1263 stream = admitted
1264 streamMu.Unlock()
1265 return admitted, nil
1266 })
1267 picker.commit(sessionTestRequest(true))
1268
1269 awaitStreamHello(t, &streamMu, &stream)
1270
1271 streamMu.Lock()
1272 admitted := stream
1273 streamMu.Unlock()
1274 require.Equal(t, 1, harness.clock.timerCount())
1275 timers := harness.clock.snapshotTimers()
1276 require.Equal(t, protocol.HandshakeTimeout, timers[0].delay)
1277 require.False(t, timers[0].stopped())
1278
1279 timers[0].fire()
1280 awaitPickerState(t, harness.sup)
1281 require.Eventually(t, func() bool { return harness.sup.State().Err != nil }, 5*time.Second, time.Millisecond)
1282 var typed ports.BrokerError
1283 require.ErrorAs(t, harness.sup.State().Err, &typed)
1284 require.Equal(t, ports.BrokerErrorTimeout, typed.Code)
1285 require.True(t, admitted.closedNow(), "the expired stream is closed")
1286}
1287
1288func TestSupervisorAttachmentDeadlineDisarmedAfterPublication(t *testing.T) {
1289 picker := newAttachTestPicker()
1290 var stream *sessionTestStream
1291 var streamMu sync.Mutex
1292 harness := startAttachHarness(t, picker)
1293 harness.service.setOpenStream(func(_ context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1294 admitted := newSessionTestStream()
1295 streamMu.Lock()
1296 stream = admitted
1297 streamMu.Unlock()
1298 return admitted, nil
1299 })
1300 picker.commit(sessionTestRequest(true))
1301 awaitStreamHello(t, &streamMu, &stream)
1302 streamMu.Lock()
1303 admitted := stream
1304 streamMu.Unlock()
1305 deliverReadyStream(t, admitted)
1306 awaitAttachedState(t, harness.sup)
1307
1308 timer := harness.clock.snapshotTimers()[0]
1309 require.True(t, timer.stopped(), "the initial-publication commit disarms the budget")
1310 timer.fire()
1311 // This bounded negative window gives the retired deadline goroutine an
1312 // opportunity to mishandle the late fire; no sleep is used for ordering.
1313 require.Never(t, func() bool {
1314 return harness.sup.State().Presentation != PresentAttached || admitted.closedNow() || harness.sup.State().Err != nil
1315 }, 50*time.Millisecond, time.Millisecond)
1316}
1317
1318func TestAttachmentDeadlineConcurrentExpiryAndRetirement(t *testing.T) {
1319 tests := []struct {
1320 name string
1321 retire func(*attachmentDeadline)
1322 }{
1323 {name: "disarm", retire: (*attachmentDeadline).disarm},
1324 {name: "adopt", retire: func(deadline *attachmentDeadline) {
1325 _ = deadline.adopt(&sessionTestStreamWithDeadline{sessionTestStream: newSessionTestStream(), deadline: time.Unix(2000, 0)})
1326 }},
1327 }
1328 for _, tt := range tests {
1329 t.Run(tt.name, func(t *testing.T) {
1330 clock := newSupervisorTestClock()
1331 deadline := startAttachmentDeadline(context.Background(), clock)
1332 timer := clock.snapshotTimers()[0]
1333 done := make(chan struct{})
1334 go func() {
1335 timer.fire()
1336 tt.retire(deadline)
1337 close(done)
1338 }()
1339 select {
1340 case <-done:
1341 case <-time.After(time.Second):
1342 t.Fatal("concurrent expiry and retirement hung")
1343 }
1344 deadline.finish()
1345 require.True(t, clock.allStopped())
1346 })
1347 }
1348}
1349
1350// testStreamLoss is a physical loss of the stream carrying an attachment.
1351func testStreamLoss() ports.BrokerStreamLost {
1352 return ports.BrokerStreamLost{Connection: ports.BrokerConnectionID{1}, Stream: 1, Epoch: 3, Cause: domain.RemoteFailureTransport, Err: errors.New("attachment lost")}
1353}
1354
1355// resumeTestJitter is the backoff jitter of the resume tests. It keeps every
1356// resume delay distinct from the other timers sharing the fake clock (the
1357// 15s attachment deadline and the 200ms palette deadline).
1358func resumeTestJitter() float64 { return 0.5 }
1359
1360// fireResumeTimerAfter waits for the attempt's backoff timer, runs before, then
1361// fires it.
1362func fireResumeTimerAfter(t *testing.T, clock *supervisorTestClock, attempt int, before func()) {
1363 t.Helper()
1364 want := supervisorBackoffDelay(uint64(attempt), resumeTestJitter)
1365 for {
1366 timer := clock.awaitTimer(t)
1367 if timer.delay == want {
1368 before()
1369 timer.fire()
1370 return
1371 }
1372 }
1373}
1374
1375func TestAttachmentResumeBudget(t *testing.T) {
1376 brief := time.Second
1377 // step is the clock advance between spends: past the outage window, the
1378 // attempt count alone decides; within it, resuming never gives up.
1379 past := resumeOutageWindow / maxAttachmentResumes
1380 tests := []struct {
1381 name string
1382 steps []attachmentOutcomeKind
1383 ups []time.Duration
1384 step time.Duration
1385 want []bool
1386 }{
1387 {name: "flapping resumes exhaust", steps: repeatKinds(attachmentResume, maxAttachmentResumes+1), ups: repeatDurations(brief, maxAttachmentResumes+1), step: past, want: append(repeatBools(true, maxAttachmentResumes), false)},
1388 {name: "stable attachment restores the budget", steps: repeatKinds(attachmentResume, maxAttachmentResumes+2), ups: repeatDurations(resumeStableAttachment, maxAttachmentResumes+2), step: past, want: repeatBools(true, maxAttachmentResumes+2)},
1389 {name: "retries never restore the budget", steps: repeatKinds(attachmentRetry, maxAttachmentResumes+1), ups: repeatDurations(resumeStableAttachment, maxAttachmentResumes+1), step: past, want: append(repeatBools(true, maxAttachmentResumes), false)},
1390 {name: "flaps exhaust inside the outage window", steps: repeatKinds(attachmentResume, maxAttachmentResumes+1), ups: repeatDurations(brief, maxAttachmentResumes+1), step: time.Second, want: append(repeatBools(true, maxAttachmentResumes), false)},
1391 {name: "outage window keeps resuming", steps: repeatKinds(attachmentRetry, 3*maxAttachmentResumes), ups: repeatDurations(brief, 3*maxAttachmentResumes), step: time.Second, want: repeatBools(true, 3*maxAttachmentResumes)},
1392 }
1393 for _, tt := range tests {
1394 t.Run(tt.name, func(t *testing.T) {
1395 var budget attachmentResumeBudget
1396 require.False(t, budget.resuming)
1397 now := time.Unix(0, 0)
1398 for i, kind := range tt.steps {
1399 _, ok := budget.spend(kind, tt.ups[i], now)
1400 now = now.Add(tt.step)
1401 require.Equal(t, tt.want[i], ok, "step %d", i)
1402 require.True(t, budget.resuming)
1403 }
1404 })
1405 }
1406}
1407
1408func repeatKinds(k attachmentOutcomeKind, n int) []attachmentOutcomeKind {
1409 out := make([]attachmentOutcomeKind, n)
1410 for i := range out {
1411 out[i] = k
1412 }
1413 return out
1414}
1415
1416func repeatDurations(d time.Duration, n int) []time.Duration {
1417 out := make([]time.Duration, n)
1418 for i := range out {
1419 out[i] = d
1420 }
1421 return out
1422}
1423
1424func repeatBools(b bool, n int) []bool {
1425 out := make([]bool, n)
1426 for i := range out {
1427 out[i] = b
1428 }
1429 return out
1430}
1431
1432// TestSupervisorAttachmentLossResumesSameSession is the regression for a
1433// remote attachment that died with attachment_lost and left the client stuck:
1434// a stream lost after attachment reconnects to the exact session it was
1435// showing, retries only transport-class failures, and returns to the picker
1436// with a single notice when the resume gives up.
1437func TestSupervisorAttachmentLossResumesSameSession(t *testing.T) {
1438 type step uint8
1439 const (
1440 // attachThenLose attaches the resumed stream, stays up long enough to
1441 // count as stable, then loses it again.
1442 attachThenLose step = iota + 1
1443 // attachThenFlap attaches the resumed stream and loses it right away.
1444 attachThenFlap
1445 // attachAndStay attaches the resumed stream and keeps it.
1446 attachAndStay
1447 // loseBeforeAttach loses the resumed stream before it attaches.
1448 loseBeforeAttach
1449 // refuse fails the resumed open with a non-transport refusal.
1450 refuse
1451 // openUnavailable fails the resumed open with a transport-class error.
1452 openUnavailable
1453 // cancelDuringBackoff ends the client while the backoff runs.
1454 cancelDuringBackoff
1455 )
1456 repeat := func(s step, n int) []step {
1457 steps := make([]step, n)
1458 for i := range steps {
1459 steps[i] = s
1460 }
1461 return steps
1462 }
1463 cases := []struct {
1464 name string
1465 steps []step
1466 // wantPicker is the final outcome: back on the picker with exactly one
1467 // DestinationFailed notice, or still attached with none.
1468 wantPicker bool
1469 }{
1470 {name: "resume attaches again", steps: []step{attachAndStay}},
1471 {name: "every stable resume restores the budget", steps: append(repeat(attachThenLose, maxAttachmentResumes+1), attachAndStay)},
1472 {name: "a flapping session gives up", steps: repeat(attachThenFlap, maxAttachmentResumes), wantPicker: true},
1473 {name: "transient failures then attach", steps: append(repeat(loseBeforeAttach, maxAttachmentResumes-1), attachAndStay)},
1474 {name: "exhausted returns to picker", steps: repeat(loseBeforeAttach, maxAttachmentResumes), wantPicker: true},
1475 {name: "refusal is final", steps: []step{refuse}, wantPicker: true},
1476 {name: "unavailable opens then attach", steps: append(repeat(openUnavailable, maxAttachmentResumes-1), attachAndStay)},
1477 {name: "unavailable opens exhaust", steps: repeat(openUnavailable, maxAttachmentResumes), wantPicker: true},
1478 {name: "cancel during backoff never resumes", steps: []step{cancelDuringBackoff}},
1479 }
1480 for _, tc := range cases {
1481 t.Run(tc.name, func(t *testing.T) {
1482 picker := newAttachTestPicker()
1483 notices := make(chan LifecycleNotice, 64)
1484 harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) {
1485 cfg.NotifyLifecycle = func(notice LifecycleNotice) { notices <- notice }
1486 cfg.Jitter = resumeTestJitter
1487 })
1488 streams := make(chan *sessionTestStream, len(tc.steps)+2)
1489 var refusing, unavailable atomic.Bool
1490 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1491 if refusing.Load() {
1492 return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Text: "session gone"}
1493 }
1494 if unavailable.Load() {
1495 return nil, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "route down"}
1496 }
1497 stream := newSessionTestStream()
1498 streams <- stream
1499 return stream, nil
1500 })
1501 picker.commit(sessionTestRequest(false))
1502 first := <-streams
1503 deliverReadyStream(t, first)
1504 awaitAttachedState(t, harness.sup)
1505 first.fail(testStreamLoss())
1506
1507 attempt := 1
1508 for i, s := range tc.steps {
1509 if s == cancelDuringBackoff {
1510 backoff := supervisorBackoffDelay(1, resumeTestJitter)
1511 for harness.clock.awaitTimer(t).delay != backoff {
1512 }
1513 harness.cancel()
1514 <-harness.runDone
1515 require.Len(t, harness.service.openedRequests(), 1, "no resume after cancellation")
1516 return
1517 }
1518 refusing.Store(s == refuse)
1519 unavailable.Store(s == openUnavailable)
1520 opened := len(harness.service.openedRequests())
1521 // Giving up also requires the outage window to pass. The
1522 // clock moves only once the last attempt's backoff is
1523 // scheduled, so no attempt measures it as uptime.
1524 last := tc.wantPicker && i == len(tc.steps)-1
1525 fireResumeTimerAfter(t, harness.clock, attempt, func() {
1526 if last {
1527 advancePickerClock(harness.clock, resumeOutageWindow)
1528 }
1529 })
1530 if s == refuse {
1531 break
1532 }
1533 if s == openUnavailable {
1534 // The failed open is retried on Connecting, without a notice.
1535 // Wait for it before the next step changes the open result.
1536 require.Eventually(t, func() bool { return len(harness.service.openedRequests()) > opened }, 5*time.Second, time.Millisecond)
1537 attempt++
1538 continue
1539 }
1540 resumed := <-streams
1541 if s == loseBeforeAttach {
1542 resumed.fail(testStreamLoss())
1543 attempt++
1544 continue
1545 }
1546 require.Eventually(t, func() bool { return len(resumed.messages()) > 0 }, 5*time.Second, time.Millisecond)
1547 deliverReadyStream(t, resumed)
1548 awaitAttachedState(t, harness.sup)
1549 switch s {
1550 case attachThenLose:
1551 advancePickerClock(harness.clock, resumeStableAttachment)
1552 resumed.fail(testStreamLoss())
1553 attempt = 1
1554 case attachThenFlap:
1555 resumed.fail(testStreamLoss())
1556 attempt++
1557 }
1558 }
1559
1560 requests := harness.service.openedRequests()
1561 for _, resume := range requests[1:] {
1562 require.Equal(t, ports.BrokerAdmissionExact, resume.Admission)
1563 require.Equal(t, sessionTestOutput(1, "").Context.Route.Target, resume.Target, "resume names the committed session")
1564 require.Equal(t, requests[0].Endpoint, resume.Endpoint)
1565 require.NotEqual(t, requests[0].Stream, resume.Stream, "resume uses a fresh stream identity")
1566 }
1567 if !tc.wantPicker {
1568 awaitAttachedState(t, harness.sup)
1569 require.False(t, picker.owns(), "the picker never took the terminal")
1570 require.Empty(t, notices, "a resume that attaches reports nothing")
1571 return
1572 }
1573 awaitPickerState(t, harness.sup)
1574 require.Eventually(t, picker.owns, 5*time.Second, time.Millisecond)
1575 failed := 0
1576 for len(notices) > 0 {
1577 if (<-notices).Kind == LifecycleNoticeDestinationFailed {
1578 failed++
1579 }
1580 }
1581 require.Equal(t, 1, failed, "the resume reports its failure once")
1582 })
1583 }
1584}
1585
1586// TestSupervisorAttachmentCancellationRetiresGeneration proves cancellation
1587// mid-attach closes the stream, ends the run, and never reports attachment.
1588func TestSupervisorAttachmentCancellationRetiresGeneration(t *testing.T) {
1589 picker := newAttachTestPicker()
1590 var (
1591 streamMu sync.Mutex
1592 stream *sessionTestStream
1593 )
1594 harness := startAttachHarness(t, picker)
1595 harness.service.setOpenStream(func(_ context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1596 admitted := newSessionTestStream()
1597 streamMu.Lock()
1598 stream = admitted
1599 streamMu.Unlock()
1600 return admitted, nil
1601 })
1602 picker.commit(sessionTestRequest(true))
1603 require.Eventually(t, func() bool { return len(harness.service.openedRequests()) == 1 }, 5*time.Second, time.Millisecond)
1604
1605 harness.cancel()
1606 require.ErrorIs(t, harness.waitRun(t), context.Canceled)
1607 streamMu.Lock()
1608 admitted := stream
1609 streamMu.Unlock()
1610 require.True(t, admitted.closedNow(), "the supervisor close authority released the stream")
1611 require.Equal(t, PresentTerminating, harness.sup.State().Presentation)
1612 require.Empty(t, harness.terminal.publications, "a cancelled attachment never published")
1613}
1614
1615// TestSupervisorConnectingWhileAcquisitionBlockedPublishesNoSessionMetadata pins
1616// the blocked-acquisition presentation: while the logical stream open is still
1617// in flight, the published presentation is Connecting with the run's handle and
1618// no session metadata or actionable generation, and capture and wait remain
1619// usable in that state.
1620func TestSupervisorConnectingWhileAcquisitionBlockedPublishesNoSessionMetadata(t *testing.T) {
1621 picker := newAttachTestPicker()
1622 release := make(chan struct{})
1623 var releaseOnce sync.Once
1624 defer releaseOnce.Do(func() { close(release) })
1625 var ui *UI
1626 harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) {
1627 ui = NewUI(cfg.Terminal.(ports.UIState), cfg.Clock)
1628 cfg.UI = ui
1629 })
1630 harness.service.setOpenStream(func(ctx context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1631 select {
1632 case <-release:
1633 return newSessionTestStream(), nil
1634 case <-ctx.Done():
1635 return nil, ctx.Err()
1636 }
1637 })
1638
1639 picker.commit(sessionTestRequest(true))
1640 require.Eventually(t, func() bool { return len(harness.service.openedRequests()) == 1 }, 5*time.Second, time.Millisecond, "the acquisition is in flight")
1641
1642 // The acquisition is blocked, so the run publishes Connecting and stays
1643 // usable: capture and wait both work in that presentation.
1644 connecting := ports.UIStatusConnecting
1645 matched, err := ui.Wait(t.Context(), ports.UIWaitRequest{Attachment: ui.Handle(), Expect: ports.UIExpect{Status: &connecting}})
1646 require.NoError(t, err)
1647 require.Equal(t, ports.UIStatusConnecting, matched.Context.Status)
1648 require.Zero(t, matched.Context.Generation, "a blocked acquisition never publishes an actionable generation")
1649 require.Equal(t, protocol.ExactSessionTarget{}, matched.Context.Route.Target, "Connecting presents no session identity")
1650 require.Equal(t, ui.Handle(), matched.Context.AttachmentHandle, "Connecting still presents the run's stable handle")
1651 snapshot, err := ui.Capture(ui.Handle())
1652 require.NoError(t, err)
1653 require.Equal(t, ports.UIStatusConnecting, snapshot.Context.Status)
1654
1655 // Input actions are attachment actions: refused while the attachment has not
1656 // committed.
1657 _, err = ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: 1, Keys: []string{"Escape"}})
1658 var actionErr *ports.UIError
1659 require.ErrorAs(t, err, &actionErr)
1660 require.Equal(t, ports.UIErrUnavailable, actionErr.Code)
1661
1662 releaseOnce.Do(func() { close(release) })
1663 harness.cancel()
1664 require.ErrorIs(t, harness.waitRun(t), context.Canceled)
1665}
1666
1667// TestSupervisorPickerInputNeverReachesSession proves picker bytes produce no
1668// stream open, no session write, and no terminal write.
1669func TestSupervisorPickerInputNeverReachesSession(t *testing.T) {
1670 picker := newAttachTestPicker()
1671 harness := startAttachHarness(t, picker)
1672 opened := false
1673 harness.service.setOpenStream(func(_ context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1674 opened = true
1675 return newSessionTestStream(), nil
1676 })
1677
1678 harness.reader.push([]byte("j"))
1679 require.Eventually(t, func() bool { return picker.consumedCount() >= 1 }, 5*time.Second, time.Millisecond)
1680 require.False(t, opened, "picker input must never open a stream")
1681 require.Zero(t, picker.resolveCount(), "picker input must never resolve a selection")
1682 require.Empty(t, harness.service.openedRequests())
1683 _, flushes, _ := harness.terminal.counts()
1684 require.Zero(t, flushes, "picker input must never flush session output")
1685 require.Empty(t, harness.terminal.written(), "picker input must never write session bytes")
1686}
1687
1688// TestSupervisorAttachmentReachableFromRealPicker drives the real client-owned
1689// picker: a committed row resolves to one exact stream request and its
1690// publication attaches.
1691func TestSupervisorAttachmentReachableFromRealPicker(t *testing.T) {
1692 ctx, cancel := context.WithCancel(context.Background())
1693 defer cancel()
1694
1695 clock := newSupervisorTestClock()
1696 reader := newPickerChunkReader()
1697 t.Cleanup(reader.close)
1698 terminal := &attachTestTerminal{in: reader, available: true}
1699 service := newSupervisorTestService(ports.BrokerConnectionID{1})
1700 service.hub.publish(pickerTestSnapshot(3, 1, clock.Now(), "alpha"))
1701 admittedStreams := make(chan *sessionTestStream, 1)
1702 service.setOpenStream(func(_ context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1703 admitted := newSessionTestStream()
1704 admittedStreams <- admitted
1705 return admitted, nil
1706 })
1707
1708 controller := newPickerController(clock, pickerTestFreshness, terminalcap.ColorCapabilities{Mode: terminalcap.TrueColor})
1709 connector := newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) { return service, nil })
1710 sup := mustSupervisor(t, SupervisorConfig{
1711 Connector: connector,
1712 Terminal: terminal,
1713 Clock: clock,
1714 Jitter: func() float64 { return 0 },
1715 Picker: controller,
1716 })
1717 runErr := make(chan error, 1)
1718 go func() { runErr <- sup.Run(ctx) }()
1719
1720 require.Eventually(t, func() bool { return controller.Catalogue().Revision() == 1 }, 5*time.Second, time.Millisecond)
1721 require.Equal(t, []string{"alpha"}, pickerSessionLabels(controller.Catalogue().Lines()))
1722
1723 // The user commits the row under the cursor through the shared reader.
1724 reader.push([]byte("\r"))
1725 require.Eventually(t, func() bool { return len(service.openedRequests()) == 1 }, 5*time.Second, time.Millisecond)
1726 opened := service.openedRequests()[0]
1727 require.Equal(t, "alpha", opened.Target.SessionName)
1728 require.True(t, opened.Local)
1729 require.Equal(t, ports.BrokerAdmissionExact, opened.Admission)
1730 require.Equal(t, ports.BrokerStreamID(1), opened.Stream)
1731
1732 var stream *sessionTestStream
1733 select {
1734 case stream = <-admittedStreams:
1735 case <-time.After(5 * time.Second):
1736 t.Fatal("the resolved stream was never admitted")
1737 }
1738 deliverReadyStream(t, stream)
1739 awaitAttachedState(t, sup)
1740
1741 cancel()
1742 require.ErrorIs(t, <-runErr, context.Canceled)
1743 require.True(t, stream.closedNow())
1744 require.Equal(t, 1, terminal.restores, "the supervisor restored raw mode exactly once")
1745}
1746
1747// TestSupervisorAttachmentConcurrentCommitAndCancellation deterministically
1748// retires attachment authority at each blocking boundary under review.
1749func TestSupervisorAttachmentConcurrentCommitAndCancellation(t *testing.T) {
1750 tests := []struct {
1751 name string
1752 run func(*testing.T, *attachTestHarness, *attachTestPicker) *sessionTestStream
1753 }{
1754 {name: "open in flight", run: func(t *testing.T, harness *attachTestHarness, picker *attachTestPicker) *sessionTestStream {
1755 entered := make(chan struct{})
1756 release := make(chan struct{})
1757 stream := newSessionTestStream()
1758 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1759 close(entered)
1760 <-release
1761 return stream, nil
1762 })
1763 picker.commit(sessionTestRequest(true))
1764 <-entered
1765 harness.cancel()
1766 close(release)
1767 return stream
1768 }},
1769 {name: "publication commit in flight", run: func(t *testing.T, harness *attachTestHarness, picker *attachTestPicker) *sessionTestStream {
1770 stream := newSessionTestStream()
1771 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1772 return stream, nil
1773 })
1774 entered := make(chan struct{}, 1)
1775 release := make(chan struct{})
1776 harness.terminal.publishEntered = entered
1777 harness.terminal.publishRelease = release
1778 picker.commit(sessionTestRequest(true))
1779 awaitHello(t, stream)
1780 stream.deliver(protocol.Welcome{SessionName: "alpha"})
1781 stream.deliver(sessionTestOutput(1, "initial"))
1782 <-entered
1783 harness.cancel()
1784 close(release)
1785 return stream
1786 }},
1787 }
1788 for _, tt := range tests {
1789 t.Run(tt.name, func(t *testing.T) {
1790 picker := newAttachTestPicker()
1791 harness := startAttachHarness(t, picker)
1792 stream := tt.run(t, harness, picker)
1793 // Cancellation revokes picker authority immediately, before teardown
1794 // completes. Inject work in that teardown window and prove neither path
1795 // can consume it.
1796 resolved, consumed := picker.resolveCount(), picker.consumedCount()
1797 picker.commit(sessionTestRequest(true))
1798 harness.reader.push([]byte("retired"))
1799 require.False(t, picker.owns())
1800 require.Equal(t, resolved, picker.resolveCount(), "revoked picker operations must not resolve")
1801 require.Equal(t, consumed, picker.consumedCount(), "revoked input must not be consumed")
1802 require.ErrorIs(t, harness.waitRun(t), context.Canceled)
1803 // Teardown has fully run: a cancelled attachment must not hand input
1804 // back to the picker on its way out.
1805 require.False(t, picker.owns(), "a terminated run must not re-acquire the picker")
1806 require.True(t, stream.closedNow(), "the admitted stream must be released")
1807 require.NotEqual(t, PresentAttached, harness.sup.State().Presentation)
1808 })
1809 }
1810}