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