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