1package client
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "io"
8 "log/slog"
9 "math/rand/v2"
10 "reflect"
11 "sync"
12 "sync/atomic"
13 "time"
14
15 "github.com/bnema/vev/internal/domain"
16 "github.com/bnema/vev/internal/domain/terminalcap"
17 "github.com/bnema/vev/internal/ports"
18 "github.com/bnema/vev/internal/usecase/keys/kittykey"
19)
20
21// Autonomous client supervisor.
22//
23// The supervisor is the terminal-owning owner of one
24// autonomous client process. It enters raw mode once, before it ever reaches
25// for the broker, and owns a single terminal input lifetime for the whole run.
26// While the broker connection is down it renders the picker and keeps
27// reconnecting. One terminal input pump serves both picker and attachment;
28// the supervisor coordinates their lifetimes without decoding session output.
29//
30// Connectivity is one attempt at a time. An attempt connects, subscribes, and
31// waits for the first committed publication before it is Ready; readiness
32// resets the equal-jitter exponential retry cadence (100ms doubling to a 2s
33// cap). A transient connect, subscribe, or snapshot failure, and the loss of an
34// established connection observed through BrokerService.Done/Err while no
35// logical stream is held, both retire the old subscription and service, wait
36// the cadence, and reconnect under a fresh generation. A failure that can never
37// succeed by retrying (an incompatible carriage or a conflicting policy) is
38// surfaced once and stops automatic reconnection; only the domain's terminal
39// codes end the process.
40//
41// A connection attempt is joined before Run returns. Since the connector
42// contract requires Connect to honor cancellation, a completion that arrives
43// after the supervisor has begun terminating is still drained and its service
44// and subscription are closed, so a late superseded success never leaks. Raw
45// mode is restored exactly once on every exit path, terminal EOF and process
46// cancellation included.
47//
48// Parent cancellation settles the run: an attempt that settles at the same
49// instant as cancellation is retired instead of adopted, and the run ends with
50// ctx.Err without publishing a retry cadence, rendering ConnectivityRetryWait,
51// or notifying a connectivity failure.
52
53// Supervisor retry cadence bounds. The exponential cap doubles from the initial
54// interval to the maximum; the equal-jitter interval is then half the cap plus
55// a uniform fraction of the other half.
56const (
57 supervisorRetryInitial = 100 * time.Millisecond
58 supervisorRetryMax = 2 * time.Second
59)
60
61// Presentation is what the autonomous client is presenting on the terminal.
62type Presentation uint8
63
64const (
65 // PresentPicker shows the session picker.
66 PresentPicker Presentation = iota + 1
67 // PresentConnecting shows the attach transition while connecting.
68 PresentConnecting
69 // PresentAttached shows a live attachment.
70 PresentAttached
71 // PresentTerminating is the terminal state: raw mode is being restored and
72 // the process is leaving.
73 PresentTerminating
74 // PresentAttachedPicker composes the client picker over a live attachment.
75 // The logical attachment stays attached: its output
76 // is applied and acknowledged but not written while the picker owns the
77 // terminal and its input. Cancel returns to the same attachment without
78 // reconnecting.
79 PresentAttachedPicker
80)
81
82func (p Presentation) String() string {
83 switch p {
84 case PresentPicker:
85 return "picker"
86 case PresentConnecting:
87 return "connecting"
88 case PresentAttached:
89 return "attached"
90 case PresentTerminating:
91 return "terminating"
92 case PresentAttachedPicker:
93 return "attached_picker"
94 default:
95 return "unknown"
96 }
97}
98
99// Connectivity is the broker-connection lifecycle.
100type Connectivity uint8
101
102const (
103 // ConnectivityDisconnected holds no broker connection: either before the
104 // first attempt, after a non-retryable failure, or while terminating.
105 ConnectivityDisconnected Connectivity = iota + 1
106 // ConnectivityConnectingBroker is one in-flight connect/subscribe/first
107 // publication attempt.
108 ConnectivityConnectingBroker
109 // ConnectivityReady holds an established connection with its first
110 // publication committed.
111 ConnectivityReady
112 // ConnectivityRetryWait waits the retry cadence before the next attempt.
113 ConnectivityRetryWait
114)
115
116func (c Connectivity) String() string {
117 switch c {
118 case ConnectivityDisconnected:
119 return "disconnected"
120 case ConnectivityConnectingBroker:
121 return "connecting_broker"
122 case ConnectivityReady:
123 return "ready"
124 case ConnectivityRetryWait:
125 return "retry_wait"
126 default:
127 return "unknown"
128 }
129}
130
131// State is the immutable projection of the supervisor.
132type State struct {
133 Presentation Presentation
134 Connectivity Connectivity
135 // Attempt counts consecutive broker-connection failures since the last
136 // Ready. It is zero before the first attempt and after Ready, and drives
137 // the equal-jitter exponential backoff.
138 Attempt uint64
139 // Generation identifies the current connection attempt. It increases before
140 // every attempt, so a settled attempt carries the generation that produced
141 // it. Run joins each attempt before it advances, so every completion today
142 // carries the current generation: the comparison in adoptAttempt is a
143 // defensive invariant for a future concurrent driver, not a path this
144 // driver can currently reach.
145 Generation uint64
146 // Err is the most recent visible connectivity failure, nil while healthy
147 // and before any failure.
148 Err error
149 // ReadyLost is set only when an established ready generation was lost and
150 // cleared by the replacement's committed publication.
151 ReadyLost bool
152 // SlowConnecting marks a Connecting presentation that is expected to be
153 // slow (a remote attachment or a resume), so its notice is shown at once.
154 // The notice cannot be painted later: the attachment foreground owns the
155 // terminal during the handshake.
156 SlowConnecting bool
157}
158
159// supervisorEventKind is one input to the pure supervisor reducer.
160type supervisorEventKind uint8
161
162const (
163 // supervisorBeginAttempt starts one connect/subscribe attempt.
164 supervisorBeginAttempt supervisorEventKind = iota + 1
165 // supervisorReady marks the first committed publication of the current
166 // attempt.
167 supervisorReady
168 // supervisorTransientFailure is a connect, subscribe, or first-publication
169 // failure that is worth retrying.
170 supervisorTransientFailure
171 // supervisorBrokerLoss is the loss of an established connection observed
172 // through BrokerService.Done/Err while no logical stream is held.
173 supervisorBrokerLoss
174 // supervisorAttachBegin marks the admission of one committed attachment
175 // stream: the picker released input and the supervisor is presenting the
176 // connecting state until the initial publication is committed.
177 supervisorAttachBegin
178 // supervisorAttached marks a committed initial publication: the worker
179 // proved the full output frame was written, flushed,
180 // and UI-committed. Welcome alone never produces it.
181 supervisorAttached
182 // supervisorAttachEnded returns a settled attachment to the picker without
183 // restarting the process. event.err carries the typed failure, if any.
184 supervisorAttachEnded
185 // supervisorNonRetryable is a failure that retrying can never fix.
186 supervisorNonRetryable
187 // supervisorNavigationSettled ends a pending initial navigation's
188 // Connecting presentation when no attachment took over the terminal.
189 supervisorNavigationSettled
190 // supervisorTerminal ends the process: terminal EOF, process cancellation,
191 // or a terminal broker failure.
192 supervisorTerminal
193 // supervisorOverlayOpened composes the picker over the live attachment.
194 supervisorOverlayOpened
195 // supervisorOverlayClosed returns from the picker overlay to the same live
196 // attachment.
197 supervisorOverlayClosed
198)
199
200type supervisorEvent struct {
201 kind supervisorEventKind
202 err error
203 // navigating marks a supervisorBeginAttempt made while an initial
204 // navigation is still pending: the user asked for a session, not the
205 // picker, so the attempt presents Connecting instead of flashing the
206 // picker until the navigation settles.
207 navigating bool
208 // slow marks a supervisorAttachBegin whose Connecting notice must not be
209 // delayed; see State.SlowConnecting.
210 slow bool
211}
212
213// reduceSupervisor is the pure connectivity/presentation reducer. It is
214// side-effect free so the state machine can be exhaustively table-tested; the
215// driver applies it under a mutex and renders the result.
216func reduceSupervisor(state State, event supervisorEvent) State {
217 switch event.kind {
218 case supervisorBeginAttempt:
219 state.Presentation = PresentPicker
220 state.SlowConnecting = false
221 if event.navigating {
222 state.Presentation = PresentConnecting
223 }
224 state.Connectivity = ConnectivityConnectingBroker
225 state.Generation++
226 state.Err = nil
227 case supervisorReady:
228 state.Connectivity = ConnectivityReady
229 state.Attempt = 0
230 state.Err = nil
231 case supervisorTransientFailure:
232 state.Connectivity = ConnectivityRetryWait
233 state.Attempt++
234 state.Err = event.err
235 case supervisorBrokerLoss:
236 state.ReadyLost = state.Connectivity == ConnectivityReady
237 state.Connectivity = ConnectivityRetryWait
238 state.Attempt++
239 state.Err = event.err
240 case supervisorAttachBegin:
241 // Connecting is the honest presentation until the committed initial
242 // publication: the broker connection is still ready and the attempt
243 // cadence is untouched.
244 state.Presentation = PresentConnecting
245 state.SlowConnecting = event.slow
246 state.Err = nil
247 case supervisorAttached:
248 state.Presentation = PresentAttached
249 state.Err = nil
250 case supervisorAttachEnded:
251 // A settled attachment always returns to the picker in the same
252 // process, attached or not; a stream that fails after attachment must
253 // never leave the attached presentation showing.
254 state.Presentation = PresentPicker
255 state.Err = event.err
256 case supervisorNonRetryable:
257 // The run stays on the picker with the visible error; a pending
258 // initial navigation can no longer present Connecting.
259 state.Presentation = PresentPicker
260 state.Connectivity = ConnectivityDisconnected
261 state.Err = event.err
262 case supervisorNavigationSettled:
263 // An initial navigation that ended without an attachment (a resolved
264 // picker, or a loss before it could be taken) leaves Connecting.
265 if state.Presentation == PresentConnecting {
266 state.Presentation = PresentPicker
267 }
268 case supervisorOverlayOpened:
269 // Only a committed attachment can host the overlay: a connecting
270 // foreground has not proven its first frame yet.
271 if state.Presentation == PresentAttached {
272 state.Presentation = PresentAttachedPicker
273 state.Err = nil
274 }
275 case supervisorOverlayClosed:
276 if state.Presentation == PresentAttachedPicker {
277 state.Presentation = PresentAttached
278 }
279 case supervisorTerminal:
280 state.Presentation = PresentTerminating
281 state.Connectivity = ConnectivityDisconnected
282 state.Err = event.err
283 }
284 // SlowConnecting describes only the Connecting that AttachBegin entered.
285 if state.Presentation != PresentConnecting {
286 state.SlowConnecting = false
287 }
288 return state
289}
290
291// LifecycleNoticeKind is one bounded autonomous lifecycle transition.
292type LifecycleNoticeKind uint8
293
294const (
295 LifecycleNoticeDetachToPicker LifecycleNoticeKind = iota + 1
296 LifecycleNoticeDetachAndExit
297 LifecycleNoticeSessionEnded
298 LifecycleNoticeDestinationFailed
299 LifecycleNoticeBrokerLost
300 LifecycleNoticeBrokerReconnected
301 LifecycleNoticeOutcomeUnknown
302 // LifecycleNoticeSelectionUnavailable reports a local refusal: the picker
303 // could not resolve the committed selection against the current catalogue,
304 // or initial navigation had nothing to resolve. No destination was dialed,
305 // so reporting it as a destination failure would name the wrong cause.
306 LifecycleNoticeSelectionUnavailable
307)
308
309// LifecycleNotice carries classification only; adapter diagnostics and free
310// text never cross this presentation seam.
311type LifecycleNotice struct {
312 Kind LifecycleNoticeKind
313}
314
315// SupervisorConfig supplies the autonomous client's dependencies. The
316// connector, terminal, and clock are required; render, notify, and jitter are
317// injectable seams with safe defaults. Render, notify, and jitter must not
318// block: the supervisor drives them from its single run goroutine.
319// SpinnerPresentation advances a transition frame on the serialized control
320// path, never from a competing terminal writer goroutine.
321type SpinnerPresentation interface {
322 Spinner() <-chan time.Time
323 AdvanceSpinner(State)
324 StopSpinner()
325}
326
327type SupervisorConfig struct {
328 // Logger receives bounded client lifecycle diagnostics. Optional.
329 Logger *slog.Logger
330 // Connector establishes the broker connection. Required.
331 Connector ports.BrokerConnector
332 // Terminal is the controlling terminal the supervisor owns. Required.
333 Terminal ports.Terminal
334 // Clock supplies the retry timers. Required; use ports.Clock, not the wall
335 // clock.
336 Clock ports.Clock
337 // Clipboard is the client-side image clipboard for remote attachments.
338 Clipboard ports.ClipboardReader
339 // Render asks the composition to paint the current state. It is called for
340 // visible transitions, after picker catalogue publications, and for every
341 // resize invalidation the supervisor's serialized waits consume. The
342 // composition paints only for the states PickerPresentation admits, which is
343 // exactly PresentPicker, using Picker.Render and Picker.RenderNotice and
344 // sampling Terminal.Geometry at render time so a resize repaint observes the
345 // latest size. PresentConnecting is deliberately not admitted: an admitted
346 // attachment foreground owns the terminal writer from Begin, before the
347 // attached marker and the initial publication, so painting then would
348 // overwrite session output. An attached foreground still owns the terminal
349 // and a terminating process is leaving it, so neither is ever painted over
350 // by a resize invalidation. Optional; the default renders nothing.
351 Render func(State)
352 Spinner SpinnerPresentation
353 // Notify surfaces a connectivity failure without leaving the picker.
354 // Optional; the default notifies nothing.
355 Notify func(State, error)
356 // NotifyLifecycle reports typed, bounded lifecycle transitions.
357 NotifyLifecycle func(LifecycleNotice)
358 // Jitter returns an equal-jitter fraction in [0,1). It is the only
359 // randomness seam so tests can pin the retry cadence exactly. Optional; the
360 // default is a process-local random source.
361 Jitter func() float64
362 // UI optionally enables supervisor-owned action admission. Each attachment
363 // is bound only after it owns the pump claim; no UI preserves the prior
364 // publication-only behavior with no admitted automation.
365 UI *UI
366 // Picker is the client-owned picker. When set, the
367 // supervisor folds every broker publication into it and hands it every
368 // terminal read from the same single input lifetime used for EOF detection,
369 // so the picker never starts a second reader. When nil the supervisor drops
370 // terminal input through the same pump. Optional.
371 Picker pickerHost
372 // AttachmentEnvironment is composition-owned process context copied into
373 // each attachment Hello. The use case deliberately does not inspect the
374 // process environment; omitted fields retain their protocol zero values.
375 AttachmentEnvironment AttachmentEnvironment
376 SessionEnvironment SessionEnvironment
377 // LifecycleActions carries explicit attachment lifecycle requests. A nil
378 // channel disables external lifecycle actions. The actions are typed so
379 // detach-to-picker can never be confused with detach-and-exit.
380 LifecycleActions <-chan AttachmentLifecycleAction
381 // InitialNavigation is the closed, one-shot navigation intent performed
382 // after the first committed catalogue publication. Its zero value is
383 // InitialNavigationPicker, which opens nothing, so a configuration that
384 // omits it is safe. No-argument composition sets the ephemeral local
385 // creation explicitly. The supervisor copies the value; it never shares a
386 // mutable pointer with the composition.
387 InitialNavigation InitialNavigation
388 // ResolveInitialNavigation optionally translates a composition-owned CLI
389 // target (an attach name or a remote endpoint) into an exact navigation
390 // identity. It is a translation of a target into identity, not an owner of
391 // navigation: it performs no I/O and is consulted at most once, against a
392 // defensive copy of the first committed publication of the adopted
393 // connection. When set, InitialNavigation must be exactly zero.
394 ResolveInitialNavigation InitialNavigationResolver
395}
396
397// AttachmentEnvironment is the composition seam for client environment data
398// carried by attachment Hello messages.
399type AttachmentEnvironment struct {
400 TermEnv string
401 Cwd string
402 // Color is the terminal's detected color capability. The zero value is
403 // TrueColor; composition fills it from terminal detection.
404 Color terminalcap.ColorCapabilities
405 // PID is this client's process ID, sent on local attaches only.
406 PID uint32
407 // ProbeTerminal enables the one bounded capability probe of a real outer
408 // terminal (Kitty graphics and keyboard). Virtual terminals leave it off.
409 ProbeTerminal bool
410 // KittyKeyboard enables the kitty keyboard protocol when the probe finds
411 // it, which makes Ctrl+1..9 recent-session switching available.
412 KittyKeyboard bool
413 // EchoPredict selects predictive local echo on remote attachments.
414 EchoPredict domain.EchoPredictMode
415}
416
417// AttachmentLifecycleAction is an explicit process/attachment decision.
418type AttachmentLifecycleActionKind uint8
419
420const (
421 AttachmentDetachToPicker AttachmentLifecycleActionKind = iota + 1
422 AttachmentDetachAndExit
423)
424
425// AttachmentLifecycleAction targets exactly one granted attachment.
426type AttachmentLifecycleAction struct {
427 Token AttachmentToken
428 Kind AttachmentLifecycleActionKind
429}
430
431// Supervisor owns one autonomous client process: raw mode, one terminal input
432// lifetime, and the broker-connectivity lifecycle.
433type Supervisor struct {
434 cfg SupervisorConfig
435 logger *slog.Logger
436
437 // picker groups the picker's per-run state: the pending commit key, the
438 // preview subscription, and the off-goroutine kill runner.
439 picker pickerSession
440
441 mu sync.Mutex
442 state State
443
444 // attachments is the single foreground host every committed attachment
445 // runs through. The supervisor owns it for the whole run
446 // and retains close authority over each stream it admits.
447 attachments *attachmentHost
448 // nextAttachment identifies one attachment run within the connection
449 // attempt generation. Logical stream IDs are allocated by the broker
450 // service, the single allocator on its own connection, so the supervisor
451 // keeps no allocator of its own.
452 nextAttachment atomic.Uint64
453 // navigation is the supervisor's private copy of the composition's intent,
454 // consumed exactly once. navigationConsumed is set before the intent is
455 // resolved, so a refusal, a cancellation, a timeout, or a reconnection can
456 // never re-arm it.
457 navigation InitialNavigation
458 navigationConsumed bool
459 // clientID is the stable client identity carried in every Hello this
460 // supervisor sends, across reconnects and attachments.
461 clientID [16]byte
462 // theme retains the terminal-reported colors across attachments, so a
463 // replacement attachment restores them before its own palette query ends.
464 theme terminalThemeState
465 // tabs remembers navigation only for this client process.
466 tabs sessionTabMemory
467 // capabilities is the outer terminal's probed and enabled capabilities,
468 // written once in Run before any attachment starts.
469 capabilities terminalCapabilities
470 // nav is the run goroutine's navigation bookkeeping: the ready-phase
471 // subscription, pending swap and in-place choices, and the route ledger.
472 nav navigationState
473}
474
475// NewSupervisor validates the required dependencies and returns a supervisor
476// parked on the picker with no connection. It does not touch the terminal; Run
477// enters raw mode.
478func NewSupervisor(cfg SupervisorConfig) (*Supervisor, error) {
479 if err := cfg.SessionEnvironment.Validate(); err != nil {
480 return nil, fmt.Errorf("client: invalid session environment: %w", err)
481 }
482 cfg.SessionEnvironment = cfg.SessionEnvironment.Clone()
483 if supervisorNil(cfg.Connector) {
484 return nil, errors.New("supervisor requires a broker connector")
485 }
486 if supervisorNil(cfg.Terminal) {
487 return nil, errors.New("supervisor requires a terminal")
488 }
489 if supervisorNil(cfg.Clock) {
490 return nil, errors.New("supervisor requires a clock")
491 }
492 if cfg.Render == nil {
493 cfg.Render = func(State) {}
494 }
495 if supervisorNil(cfg.Picker) {
496 // Normalize a typed-nil picker so Run's nil check and the input
497 // lifetime's consumer check are both safe.
498 cfg.Picker = nil
499 }
500 if cfg.Jitter == nil {
501 cfg.Jitter = rand.Float64
502 }
503 if cfg.ResolveInitialNavigation != nil {
504 if cfg.InitialNavigation != (InitialNavigation{}) {
505 return nil, errors.New("supervisor accepts either an initial navigation or a resolver, not both")
506 }
507 } else if err := cfg.InitialNavigation.Validate(); err != nil {
508 return nil, fmt.Errorf("supervisor initial navigation: %w", err)
509 }
510 logger := cfg.Logger
511 if logger == nil {
512 logger = slog.New(slog.NewTextHandler(io.Discard, nil))
513 }
514 supervisor := &Supervisor{
515 cfg: cfg,
516 logger: logger,
517 state: State{Presentation: PresentPicker, Connectivity: ConnectivityDisconnected},
518 clientID: newClientID(),
519 navigation: cfg.InitialNavigation,
520 picker: pickerSession{preview: previewManager{clock: cfg.Clock}},
521 }
522 // The host is the supervisor's existing foreground grant. It owns no raw
523 // mode and starts no reader: the supervisor keeps its one terminal input
524 // lifetime, so the attachment path adds neither a second reader nor a
525 // second writer. The optional UI binds to this host's existing pump claim;
526 // it never starts another terminal reader.
527 supervisor.attachments = newAttachmentHost(attachmentHostConfig{
528 Terminal: cfg.Terminal,
529 Clock: cfg.Clock,
530 Actions: cfg.LifecycleActions,
531 ActionUI: cfg.UI,
532 OnAttached: func(AttachmentToken) { supervisor.transition(supervisorEvent{kind: supervisorAttached}) },
533 })
534 return supervisor, nil
535}
536
537// initialNavigationPending reports whether an initial navigation other than
538// the picker is still armed. A resolver may still resolve to the picker; the
539// navigation then settles back to it without an attachment.
540func (s *Supervisor) initialNavigationPending() bool {
541 s.mu.Lock()
542 defer s.mu.Unlock()
543 if s.navigationConsumed {
544 return false
545 }
546 return s.cfg.ResolveInitialNavigation != nil || s.navigation.Kind != InitialNavigationPicker
547}
548
549// State returns the current immutable projection.
550func (s *Supervisor) State() State {
551 s.mu.Lock()
552 defer s.mu.Unlock()
553 return s.state
554}
555
556// Run owns the terminal for exactly its lifetime. It enters raw mode before it
557// connects to the broker, starts the single terminal input reader, and then
558// drives connect/subscribe/retry until the terminal reaches EOF, the process
559// context is cancelled, or a terminal broker failure settles the process. Raw
560// mode is restored exactly once on every exit path.
561func (s *Supervisor) Run(ctx context.Context) (retErr error) {
562 restore, err := s.cfg.Terminal.EnterRaw()
563 if err != nil {
564 return fmt.Errorf("entering raw mode: %w", err)
565 }
566 if restore == nil {
567 restore = func() error { return nil }
568 }
569 restoreTerminal := sync.OnceValue(restore)
570 defer func() {
571 if rerr := restoreTerminal(); rerr != nil && retErr == nil {
572 retErr = fmt.Errorf("restoring terminal: %w", rerr)
573 }
574 }()
575
576 input := startTerminalInputLifetimeWith(s.cfg.Terminal.In(), s.cfg.Picker, func(pump *terminalInputPump) {
577 s.detectTerminalCapabilities(ctx, pump)
578 })
579 defer input.stop()
580 s.attachments.setInput(input.pump)
581 s.attachments.startGeometry(ctx)
582 defer s.attachments.stopGeometry()
583
584 // The active connection is owned across loop iterations. retire is
585 // idempotent so every exit path closes exactly the connection it adopted.
586 var (
587 service ports.BrokerNavigator
588 sub ports.BrokerSubscription
589 )
590 retire := func() {
591 s.retirePickerKill()
592 s.picker.preview.close(s.cfg.Picker)
593 s.nav.readySub = nil
594 if sub != nil {
595 sub.Close()
596 sub = nil
597 }
598 if service != nil {
599 _ = service.Close()
600 service = nil
601 }
602 }
603 defer retire()
604
605 for {
606 // The generation is claimed before the connector runs, so the attempt is
607 // always attributable to the state that started it and a completion can be
608 // matched against the generation that produced it.
609 s.transition(supervisorEvent{kind: supervisorBeginAttempt, navigating: s.initialNavigationPending()})
610 generation := s.State().Generation
611 attemptCtx, cancelAttempt := context.WithCancel(ctx)
612 outcome := make(chan supervisorAttempt, 1)
613 go func() { outcome <- s.connectAttempt(attemptCtx, generation) }()
614
615 result, terminated, termErr := s.awaitAttempt(ctx, input, outcome)
616 if terminated {
617 // Connect must honor cancellation, so joining the attempt here both
618 // retires a late superseded success and leaves no goroutine behind.
619 cancelAttempt()
620 late := <-outcome
621 late.retire()
622 s.transition(supervisorEvent{kind: supervisorTerminal, err: termErr})
623 return termErr
624 }
625 cancelAttempt()
626 adopted, cause := s.adoptAttempt(ctx, result)
627 if !adopted {
628 // The cancelled or superseded attempt is retired rather than adopted, so
629 // it cannot drive the state of a newer attempt and a cancelled run never
630 // publishes a retry cadence, renders ConnectivityRetryWait, or notifies a
631 // connectivity failure for it.
632 result.retire()
633 if cause != nil {
634 s.transition(supervisorEvent{kind: supervisorTerminal, err: cause})
635 return cause
636 }
637 continue
638 }
639
640 if result.err != nil {
641 if supervisorTerminalFailure(result.err) {
642 s.transition(supervisorEvent{kind: supervisorTerminal, err: result.err})
643 return result.err
644 }
645 if !supervisorRetryable(result.err) {
646 // Retrying can never fix this. Return to the picker with the
647 // visible error and stop reconnecting; the process still owns
648 // the terminal until the user exits or it is cancelled.
649 s.transition(supervisorEvent{kind: supervisorNonRetryable, err: result.err})
650 s.notify(result.err)
651 terminalErr := s.awaitTermination(ctx, input)
652 s.transition(supervisorEvent{kind: supervisorTerminal, err: terminalErr})
653 return terminalErr
654 }
655 if terminated, termErr := s.retryFailure(ctx, input, supervisorTransientFailure, result.err); terminated {
656 return termErr
657 }
658 continue
659 }
660
661 service, sub = result.service, result.sub
662 s.nav.readySub = sub
663 if s.cfg.Picker != nil {
664 // Render the first committed publication immediately, before waiting
665 // for the next one; the picker never shows a stale empty catalogue
666 // while an established connection already has state.
667 s.cfg.Picker.ApplySnapshot(service.Snapshot())
668 s.renderCurrent()
669 }
670 wasReadyLost := s.State().ReadyLost
671 s.transition(supervisorEvent{kind: supervisorReady})
672 s.refreshPreview(service)
673 if wasReadyLost {
674 s.notifyLifecycle(LifecycleNoticeBrokerReconnected)
675 s.mu.Lock()
676 s.state.ReadyLost = false
677 s.mu.Unlock()
678 }
679 if terminated, termErr := s.runInitialNavigation(ctx, input, service); terminated {
680 retire()
681 s.transition(supervisorEvent{kind: supervisorTerminal, err: termErr})
682 return termErr
683 }
684 // A navigation left armed (the broker was lost while it waited for an
685 // observation) keeps Connecting: the next attempt takes it again.
686 if !s.initialNavigationPending() {
687 s.transition(supervisorEvent{kind: supervisorNavigationSettled})
688 }
689
690 // Ready phase: fold publications, admit committed attachments one at a
691 // time, and return to the picker after each. A committed attachment
692 // never replaces the broker connection; only a broker loss leaves this
693 // loop, and cancellation or terminal EOF ends the run.
694 var lossErr error
695 ready:
696 for {
697 ready := s.awaitReady(ctx, input, service, sub)
698 if ready.terminated {
699 retire()
700 s.transition(supervisorEvent{kind: supervisorTerminal, err: ready.termErr})
701 return ready.termErr
702 }
703 if ready.commitKey == "" {
704 lossErr = ready.lossErr
705 break ready
706 }
707 if terminated, termErr := s.runCommittedAttachment(ctx, input, service, ready.commitKey); terminated {
708 retire()
709 s.transition(supervisorEvent{kind: supervisorTerminal, err: termErr})
710 return termErr
711 }
712 }
713 retire()
714 if cause := ctx.Err(); cause != nil {
715 // Parent cancellation wins over a loss that settled at the same instant,
716 // exactly as it does at the attempt settle. The connection was already
717 // released above, so the loss is simply never turned into a retry
718 // cadence, a ConnectivityRetryWait render, or a notification for a run
719 // that was already cancelled.
720 s.transition(supervisorEvent{kind: supervisorTerminal, err: cause})
721 return cause
722 }
723 s.notifyLifecycle(LifecycleNoticeBrokerLost)
724 if terminated, termErr := s.retryFailure(ctx, input, supervisorBrokerLoss, normalizeUnavailable(lossErr)); terminated {
725 return termErr
726 }
727 }
728}
729
730// supervisorAttempt is the settled outcome of one connect/subscribe/first
731// publication attempt. generation is the attempt's generation, claimed before
732// the connector ran, so the completion is always attributable to the attempt
733// that produced it (see adoptAttempt's generation fence).
734type supervisorAttempt struct {
735 generation uint64
736 service ports.BrokerNavigator
737 sub ports.BrokerSubscription
738 err error
739}
740
741// retire closes whatever the attempt produced. It is safe on a partially built
742// or empty attempt, and it treats a typed-nil service or subscription as absent
743// rather than calling into it.
744func (a supervisorAttempt) retire() {
745 if !supervisorNil(a.sub) {
746 a.sub.Close()
747 }
748 if !supervisorNil(a.service) {
749 _ = a.service.Close()
750 }
751}
752
753// adoptAttempt decides whether one settled attempt may be adopted, and reports the
754// parent cancellation that must settle the run instead. The decision is made under
755// the state lock, atomically with respect to the state it consults, so an attempt
756// that settled after the parent cancelled the run is never adopted even though
757// awaitAttempt's select may have taken its outcome over an already-ready
758// ctx.Done. The generation fence is enforced at the same decision point: a
759// superseded attempt is rejected too. It is a defensive invariant, not a
760// currently reachable production path, because Run joins each attempt before it
761// advances; it is kept so a future driver that admits more than one attempt at a
762// time can never let a late completion drive the state of a newer attempt.
763func (s *Supervisor) adoptAttempt(ctx context.Context, result supervisorAttempt) (adopted bool, cause error) {
764 s.mu.Lock()
765 defer s.mu.Unlock()
766 if cause := ctx.Err(); cause != nil {
767 return false, cause
768 }
769 if result.generation != s.state.Generation {
770 return false, nil
771 }
772 return true, nil
773}
774
775// retryFailure records one retryable connectivity failure, surfaces it, and
776// waits the cadence for the next attempt. It is the shared tail of the transient
777// connect/subscribe failure and the established-connection loss paths, so both
778// select the same observable transition, notification, and backoff.
779func (s *Supervisor) retryFailure(ctx context.Context, input *terminalInputLifetime, kind supervisorEventKind, err error) (bool, error) {
780 if cause := ctx.Err(); cause != nil {
781 s.transition(supervisorEvent{kind: supervisorTerminal, err: cause})
782 return true, cause
783 }
784 s.transition(supervisorEvent{kind: kind, err: err})
785 s.notify(err)
786 if terminated, termErr := s.waitBackoff(ctx, input); terminated {
787 s.transition(supervisorEvent{kind: supervisorTerminal, err: termErr})
788 return true, termErr
789 }
790 return false, nil
791}
792
793// connectAttempt connects once, subscribes, and waits for the first committed
794// publication. It returns a live service and subscription only when the
795// connection is Ready; every failure closes whatever it opened before
796// returning, so a caller has nothing to clean up on the error path. Every
797// connect, subscribe, or first-publication failure is stamped with generation
798// and normalized to a typed unavailable failure so the run loop never observes
799// an untyped cause.
800func (s *Supervisor) connectAttempt(ctx context.Context, generation uint64) supervisorAttempt {
801 s.logger.Debug("client_broker_connect_begin", "generation", generation)
802 service, err := s.cfg.Connector.Connect(ctx)
803 if err != nil {
804 failure := normalizeUnavailable(err)
805 s.logger.Warn("client_broker_connect_failed", "generation", generation, "code", brokerErrorCode(failure), "error", failure, "cause", errors.Unwrap(failure))
806 return supervisorAttempt{generation: generation, err: failure}
807 }
808 if supervisorNil(service) {
809 return supervisorAttempt{generation: generation, err: ports.BrokerError{
810 Code: ports.BrokerErrorUnavailable,
811 Text: "broker connector returned no service",
812 }}
813 }
814 sub, err := service.Subscribe()
815 if err != nil {
816 _ = service.Close()
817 failure := normalizeUnavailable(err)
818 s.logger.Warn("client_broker_subscribe_failed", "generation", generation, "code", brokerErrorCode(failure), "error", failure, "cause", errors.Unwrap(failure))
819 return supervisorAttempt{generation: generation, err: failure}
820 }
821 if supervisorNil(sub) {
822 // A service that reports no subscription cannot ever become Ready, so it
823 // is a typed unavailable failure rather than a nil dereference on the
824 // first Changed() call.
825 _ = service.Close()
826 return supervisorAttempt{generation: generation, err: ports.BrokerError{
827 Code: ports.BrokerErrorUnavailable,
828 Text: "broker service returned no subscription",
829 }}
830 }
831 for {
832 if service.Snapshot().Epoch != 0 {
833 return supervisorAttempt{generation: generation, service: service, sub: sub}
834 }
835 select {
836 case <-sub.Changed():
837 continue
838 case <-service.Done():
839 cause := normalizeUnavailable(service.Err())
840 s.logger.Warn("client_broker_initial_publication_lost", "generation", generation, "code", brokerErrorCode(cause), "error", cause, "cause", errors.Unwrap(cause))
841 sub.Close()
842 _ = service.Close()
843 return supervisorAttempt{generation: generation, err: cause}
844 case <-ctx.Done():
845 sub.Close()
846 _ = service.Close()
847 return supervisorAttempt{generation: generation, err: ctx.Err()}
848 }
849 }
850}
851
852// awaitAttempt waits for the in-flight attempt to settle, or for the run to be
853// terminated first.
854func (s *Supervisor) awaitAttempt(ctx context.Context, input *terminalInputLifetime, outcome <-chan supervisorAttempt) (supervisorAttempt, bool, error) {
855 for {
856 select {
857 case result := <-outcome:
858 return result, false, nil
859 case <-s.pickerOps():
860 if s.takePickerClose() {
861 return supervisorAttempt{}, true, nil
862 }
863 case <-ctx.Done():
864 return supervisorAttempt{}, true, ctx.Err()
865 case err := <-input.EOF():
866 return supervisorAttempt{}, true, terminalReadCause(err)
867 case <-s.presentationInvalidation():
868 s.renderResizeInvalidation()
869 case <-s.spinnerTick():
870 s.cfg.Spinner.AdvanceSpinner(s.State())
871 }
872 }
873}
874
875// readyOutcome is the result of one wait inside the established-connection
876// phase: either the broker connection was lost, the run was terminated, or a
877// committed picker selection is ready to be attached.
878type readyOutcome struct {
879 // commitKey is the exact catalogue key captured with a commit decision. A
880 // non-empty value names one attachment to admit; the ready loop never
881 // resolves anything else.
882 commitKey string
883 // lossErr is the broker service's stable terminal cause when the
884 // established connection was lost.
885 lossErr error
886 // terminated reports that the run itself must end, with termErr carrying
887 // the process cancellation or terminal-read cause.
888 terminated bool
889 termErr error
890}
891
892// awaitReady waits for an established connection to be lost, for a committed
893// picker selection, or for the run to be terminated first. Publications are
894// folded into the picker as they arrive; a commit decision wakes the loop so
895// the supervisor can admit exactly the row the user committed. It opens no
896// stream and starts no terminal work itself.
897func (s *Supervisor) awaitReady(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, sub ports.BrokerSubscription) readyOutcome {
898 if key := s.picker.pendingKey; key != "" {
899 s.picker.pendingKey = ""
900 return readyOutcome{commitKey: key}
901 }
902 var changed <-chan struct{}
903 var ops <-chan struct{}
904 if s.cfg.Picker != nil {
905 if !supervisorNil(sub) {
906 changed = sub.Changed()
907 }
908 ops = s.cfg.Picker.OpsReady()
909 }
910 notices := noticeWake{clock: s.cfg.Clock}
911 defer notices.stop()
912 for {
913 select {
914 case <-notices.arm(s.cfg.Picker):
915 s.renderCurrent()
916 case <-service.Done():
917 return readyOutcome{lossErr: service.Err()}
918 case <-changed:
919 s.cfg.Picker.ApplySnapshot(service.Snapshot())
920 s.refreshPreview(service)
921 s.renderCurrent()
922 case <-s.picker.preview.changed():
923 if s.picker.preview.publish(s.cfg.Picker) {
924 s.renderCurrent()
925 }
926 case outcome := <-s.picker.kills.results():
927 s.finishPickerKill(service, outcome)
928 case <-ops:
929 op, key := s.cfg.Picker.TakeOp()
930 if op.kill && !op.commit && !op.close {
931 s.startPickerKill(service, key)
932 s.renderCurrent()
933 continue
934 }
935 if !op.close && !op.commit {
936 s.refreshPreview(service)
937 s.renderCurrent()
938 continue
939 }
940 if op.close {
941 if op.exit {
942 // Ctrl+C is the explicit exit. With no attachment to
943 // return to, it is the only close that ends the process.
944 s.picker.preview.close(s.cfg.Picker)
945 return readyOutcome{terminated: true}
946 }
947 // Escape and q cancel; with no attachment there is nothing to
948 // cancel back to, so the picker stays and says how to leave.
949 s.notifyPicker(pickerExitHint)
950 s.renderCurrent()
951 continue
952 }
953 if op.commit && key != "" {
954 s.picker.preview.close(s.cfg.Picker)
955 return readyOutcome{commitKey: key}
956 }
957 case <-ctx.Done():
958 return readyOutcome{terminated: true, termErr: ctx.Err()}
959 case err := <-input.EOF():
960 return readyOutcome{terminated: true, termErr: terminalReadCause(err)}
961 case <-s.presentationInvalidation():
962 s.refreshPreview(service)
963 s.renderResizeInvalidation()
964 }
965 }
966}
967
968func (s *Supervisor) refreshPreview(service ports.BrokerNavigator) {
969 if s == nil || s.cfg.Picker == nil {
970 return
971 }
972 geometry, err := s.cfg.Terminal.Geometry()
973 if err != nil || !geometry.Valid() {
974 s.picker.preview.close(s.cfg.Picker)
975 return
976 }
977 s.picker.preview.refresh(service, s.cfg.Picker, geometry.Size)
978}
979
980// pickerOps is nil when no picker owns presentation decisions.
981func (s *Supervisor) pickerOps() <-chan struct{} {
982 if s.cfg.Picker == nil {
983 return nil
984 }
985 return s.cfg.Picker.OpsReady()
986}
987
988// takePickerClose honors the explicit exit while offline without losing a
989// racing commit. A plain cancel (Escape, q) has no attachment to return to and
990// never ends the process.
991func (s *Supervisor) takePickerClose() bool {
992 op, key := s.cfg.Picker.TakeOp()
993 if op.commit && key != "" {
994 s.picker.pendingKey = key
995 }
996 if op.close && !op.exit {
997 s.notifyPicker(pickerExitHint)
998 s.renderCurrent()
999 }
1000 if op.kill && !op.commit && !op.close && key != "" {
1001 // A kill needs the broker's control stream; it is refused, not queued.
1002 s.notifyPicker(pickerKillRefused("couldn't kill: broker unavailable"))
1003 s.renderCurrent()
1004 }
1005 return op.close && op.exit
1006}
1007
1008// pickerPresentationHost is the optional presentation surface of the real
1009// picker. Scripted pickers may omit it.
1010type pickerPresentationHost interface {
1011 Notify(n domain.Notification)
1012 invalidatePresentation()
1013}
1014
1015// notifyPicker shows one bounded client-local notice when the picker
1016// supports it.
1017func (s *Supervisor) notifyPicker(n domain.Notification) {
1018 if host, ok := s.cfg.Picker.(pickerPresentationHost); ok {
1019 host.Notify(n)
1020 }
1021}
1022
1023// noticeDeadliner reports when the picker's notice stack next changes.
1024type noticeDeadliner interface {
1025 noticeDeadline() (time.Time, bool)
1026}
1027
1028// noticeWake wakes the ready loop when a picker notice expires, so the stack
1029// advances and freed cells are blanked without waiting for other input.
1030type noticeWake struct {
1031 clock ports.Clock
1032 timer ports.Timer
1033 due time.Time
1034}
1035
1036// arm returns a channel that fires at the next notice deadline, or nil when
1037// no notice will change on its own.
1038func (w *noticeWake) arm(picker pickerHost) <-chan time.Time {
1039 host, ok := picker.(noticeDeadliner)
1040 if !ok || supervisorNil(w.clock) {
1041 return nil
1042 }
1043 due, ok := host.noticeDeadline()
1044 if !ok {
1045 w.stop()
1046 return nil
1047 }
1048 if w.timer == nil || !due.Equal(w.due) {
1049 w.stop()
1050 w.timer = w.clock.NewTimer(max(due.Sub(w.clock.Now()), 0))
1051 w.due = due
1052 }
1053 return w.timer.C()
1054}
1055
1056func (w *noticeWake) stop() {
1057 stopSupervisorTimer(w.timer)
1058 w.timer = nil
1059 w.due = time.Time{}
1060}
1061
1062// pickerExitHint tells an unattached user how to leave the picker.
1063var pickerExitHint = domain.Notification{Code: domain.NoticeUser, Severity: domain.NoticeInfo, Message: "no session attached: press Ctrl+C to quit"}
1064
1065// invalidatePickerPresentation forces the next picker frame to redraw the
1066// whole box after another owner wrote the terminal.
1067func (s *Supervisor) invalidatePickerPresentation() {
1068 if host, ok := s.cfg.Picker.(pickerPresentationHost); ok {
1069 host.invalidatePresentation()
1070 }
1071}
1072
1073// awaitTermination parks the supervisor on the picker after a non-retryable
1074// failure until the terminal ends or the process is cancelled. A clean terminal
1075// EOF returns nil; every other cause is reported.
1076func (s *Supervisor) awaitTermination(ctx context.Context, input *terminalInputLifetime) error {
1077 notices := noticeWake{clock: s.cfg.Clock}
1078 defer notices.stop()
1079 for {
1080 select {
1081 case <-s.pickerOps():
1082 if s.takePickerClose() {
1083 return nil
1084 }
1085 case <-ctx.Done():
1086 return ctx.Err()
1087 case err := <-input.EOF():
1088 return terminalReadCause(err)
1089 case <-s.presentationInvalidation():
1090 s.renderResizeInvalidation()
1091 case <-notices.arm(s.cfg.Picker):
1092 s.renderCurrent()
1093 }
1094 }
1095}
1096
1097// waitBackoff waits one equal-jitter retry interval for the current failure
1098// count, or settles on cancellation or terminal EOF. The timer is always
1099// stopped and drained, so no timer outlives the attempt transition.
1100func (s *Supervisor) waitBackoff(ctx context.Context, input *terminalInputLifetime) (bool, error) {
1101 delay := supervisorBackoffDelay(s.State().Attempt, s.cfg.Jitter)
1102 if delay <= 0 {
1103 return false, nil
1104 }
1105 timer := s.cfg.Clock.NewTimer(delay)
1106 defer stopSupervisorTimer(timer)
1107 notices := noticeWake{clock: s.cfg.Clock}
1108 defer notices.stop()
1109 for {
1110 select {
1111 case <-s.pickerOps():
1112 if s.takePickerClose() {
1113 return true, nil
1114 }
1115 case <-timer.C():
1116 return false, nil
1117 case <-ctx.Done():
1118 return true, ctx.Err()
1119 case err := <-input.EOF():
1120 return true, terminalReadCause(err)
1121 case <-s.presentationInvalidation():
1122 s.renderResizeInvalidation()
1123 case <-s.spinnerTick():
1124 s.cfg.Spinner.AdvanceSpinner(s.State())
1125 case <-notices.arm(s.cfg.Picker):
1126 s.renderCurrent()
1127 }
1128 }
1129}
1130
1131// supervisorBackoffDelay is the equal-jitter exponential interval for one
1132// failure count: the cap doubles from supervisorRetryInitial to
1133// supervisorRetryMax, and the interval is half the cap plus a uniform fraction
1134// of the other half, so attempts spread across [cap/2, cap). It is pure so the
1135// cadence can be tested exactly.
1136func supervisorBackoffDelay(attempt uint64, jitter func() float64) time.Duration {
1137 if attempt == 0 {
1138 return 0
1139 }
1140 limit := supervisorRetryInitial
1141 for i := uint64(1); i < attempt && limit < supervisorRetryMax; i++ {
1142 limit *= 2
1143 }
1144 if limit > supervisorRetryMax {
1145 limit = supervisorRetryMax
1146 }
1147 half := limit / 2
1148 fraction := jitter()
1149 if !(fraction >= 0) {
1150 // NaN is unordered: both fraction < 0 and fraction > 1 are false for
1151 // it, so an explicit non-negative test maps NaN and every negative value
1152 // to the minimum fraction. Without this, a NaN would reach the duration
1153 // conversion and yield an undefined interval instead of the documented
1154 // half-cap minimum.
1155 fraction = 0
1156 }
1157 if fraction > 1 {
1158 fraction = 1
1159 }
1160 return half + time.Duration(fraction*float64(half))
1161}
1162
1163// terminalReadCause maps a terminal read end to the Run result: a clean EOF is
1164// an orderly exit (nil), any other read failure is reported.
1165func terminalReadCause(err error) error {
1166 if err == nil || errors.Is(err, io.EOF) {
1167 return nil
1168 }
1169 return err
1170}
1171
1172// stopSupervisorTimer stops and drains a timer, supporting buffered timer
1173// adapters with pre-Go-1.23 reset semantics.
1174func stopSupervisorTimer(timer ports.Timer) {
1175 if timer == nil {
1176 return
1177 }
1178 if !timer.Stop() {
1179 select {
1180 case <-timer.C():
1181 default:
1182 }
1183 }
1184}
1185
1186// supervisorTerminalFailure reports whether a broker failure must end the
1187// process. Only the domain's terminal codes do; every other failure returns the
1188// client to the picker.
1189func supervisorTerminalFailure(err error) bool {
1190 var typed ports.BrokerError
1191 return errors.As(err, &typed) && typed.Terminal()
1192}
1193
1194// supervisorRetryable reports whether a failure is worth another attempt after
1195// backoff. An incompatible carriage or a conflicting policy can never succeed
1196// by retrying, so it is surfaced once and automatic reconnection stops; every
1197// other failure is treated as transient.
1198func supervisorRetryable(err error) bool {
1199 var typed ports.BrokerError
1200 if errors.As(err, &typed) {
1201 switch typed.Code {
1202 case ports.BrokerErrorIncompatible, ports.BrokerErrorConflictingPolicy:
1203 return false
1204 }
1205 }
1206 return true
1207}
1208
1209// normalizeUnavailable presents one connect, subscribe, or connection-loss
1210// failure as a typed ports.BrokerError. A cause that is already a BrokerError is
1211// returned unchanged, so its own code (for example incompatible or conflicting
1212// policy) still selects the non-retryable path; every untyped failure, including
1213// a nil cause, becomes an unavailable BrokerError that preserves the original
1214// error as its local-only Cause.
1215func brokerErrorCode(err error) string {
1216 var typed ports.BrokerError
1217 if errors.As(err, &typed) {
1218 return typed.Code.String()
1219 }
1220 return "unknown"
1221}
1222
1223func normalizeUnavailable(err error) error {
1224 if err == nil {
1225 return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker connection lost"}
1226 }
1227 var typed ports.BrokerError
1228 if errors.As(err, &typed) {
1229 return err
1230 }
1231 var admission ports.BrokerAdmissionError
1232 if errors.As(err, &admission) {
1233 return err
1234 }
1235 return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker connection lost", Cause: err}
1236}
1237
1238// notify surfaces a connectivity failure through the optional notifier.
1239func (s *Supervisor) notifyLifecycle(kind LifecycleNoticeKind) {
1240 if s != nil {
1241 s.logger.Debug("client_lifecycle_notice", "kind", kind)
1242 }
1243 if s != nil && s.cfg.NotifyLifecycle != nil {
1244 s.cfg.NotifyLifecycle(LifecycleNotice{Kind: kind})
1245 }
1246}
1247
1248func (s *Supervisor) notify(err error) {
1249 if err != nil {
1250 s.logger.Warn("client_connectivity_failure", "state", s.State().Connectivity.String(), "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
1251 }
1252 if s.cfg.Notify == nil || err == nil {
1253 return
1254 }
1255 s.cfg.Notify(s.State(), err)
1256}
1257
1258// transition applies one reducer event under the supervisor mutex and renders
1259// the resulting state outside it, so a renderer that reads State back never
1260// deadlocks.
1261//
1262// A presentation change also publishes the new presentation through the UI
1263// owner inside the same fence, so a capture observes the presentation the
1264// supervisor actually entered instead of an older attachment context. The
1265// publication is deliberately limited to the unattached presentations the
1266// supervisor owns: Picker and Connecting carry the run's handle and status with
1267// every session field zeroed, while Attached is published by the admitted
1268// attachment foreground only, after its first frame was written, flushed, and
1269// committed, carrying the real action generation. Terminating closes the UI
1270// service and publishes nothing.
1271func (s *Supervisor) transition(event supervisorEvent) {
1272 s.mu.Lock()
1273 previous := s.state.Presentation
1274 s.state = reduceSupervisor(s.state, event)
1275 state := s.state
1276 s.mu.Unlock()
1277 if state.Presentation != previous || event.kind == supervisorBeginAttempt || event.kind == supervisorTransientFailure || event.kind == supervisorBrokerLoss {
1278 s.logger.Debug("client_state_transition", "from", previous.String(), "to", state.Presentation.String(), "connectivity", state.Connectivity.String(), "attempt", state.Attempt, "generation", state.Generation, "error", event.err, "cause", errors.Unwrap(event.err))
1279 }
1280 if state.Presentation != previous {
1281 s.publishPresentation(state.Presentation)
1282 }
1283 if s.cfg.Spinner != nil && state.Presentation != PresentConnecting && state.Connectivity != ConnectivityRetryWait && state.Connectivity != ConnectivityConnectingBroker {
1284 s.cfg.Spinner.StopSpinner()
1285 }
1286 if s.cfg.Render != nil {
1287 s.cfg.Render(state)
1288 }
1289}
1290
1291func (s *Supervisor) spinnerTick() <-chan time.Time {
1292 if s.cfg.Spinner == nil {
1293 return nil
1294 }
1295 return s.cfg.Spinner.Spinner()
1296}
1297
1298// publishPresentation publishes one supervisor-owned presentation through the
1299// optional UI owner. Attached and terminating are refused: Attached belongs to
1300// the admitted attachment foreground, and terminating ends the service.
1301func (s *Supervisor) publishPresentation(presentation Presentation) {
1302 if s == nil || s.cfg.UI == nil {
1303 return
1304 }
1305 switch presentation {
1306 case PresentPicker:
1307 _ = s.cfg.UI.publishPresentation(ports.UIStatusPicker)
1308 case PresentConnecting:
1309 _ = s.cfg.UI.publishPresentation(ports.UIStatusConnecting)
1310 }
1311}
1312
1313// renderCurrent repaints presentation data that changed without a supervisor
1314// state transition, notably a picker catalogue publication.
1315func (s *Supervisor) renderCurrent() {
1316 if s == nil || s.cfg.Render == nil {
1317 return
1318 }
1319 s.cfg.Render(s.State())
1320}
1321
1322// presentationInvalidation returns the coalesced resize invalidation the
1323// supervisor's serialized waits select on, or nil when there is no attachment
1324// host to collect resizes. A nil signal is simply never ready, so a supervisor
1325// with no host (the zero value, or one whose host was detached) parks on its
1326// remaining arms instead of dereferencing a nil host. Selecting on a nil
1327// channel is safe in both directions. The returned channel is receive-only: the
1328// supervisor only ever consumes an invalidation, and the attachment settlement
1329// wait deliberately does not select on it at all, so an invalidation raised
1330// while a foreground is admitted stays buffered for the next picker state.
1331func (s *Supervisor) presentationInvalidation() <-chan struct{} {
1332 if s == nil || s.attachments == nil {
1333 return nil
1334 }
1335 return s.attachments.presentationUpdate
1336}
1337
1338// renderResizeInvalidation runs only on the supervisor's serialized control
1339// path, so resize collection cannot write the terminal and an attached
1340// foreground is never painted over. The composition's renderer is invoked only
1341// for a presentation PickerPresentation admits, which is exactly PresentPicker;
1342// it samples Terminal.Geometry at render time. A pre-attachment connecting
1343// presentation is refused here for the same reason as an attached one: the
1344// admitted foreground already owns the same terminal writer.
1345func (s *Supervisor) renderResizeInvalidation() {
1346 if s == nil || s.cfg.Render == nil {
1347 return
1348 }
1349 state := s.State()
1350 if PickerPresentation(state) {
1351 s.cfg.Render(state)
1352 }
1353}
1354
1355// terminalInputLifetime owns the single read of the controlling terminal. It is
1356// started once and never replaced: a bare io.Reader cannot be interrupted, so
1357// the goroutine may outlive Run when stdin never reaches EOF, exactly like the
1358// attach client's input pump. It never closes caller-owned input. When a picker
1359// consumer is supplied each read is handed to it from this one
1360// reader, so the picker never starts a second reader; without a consumer the
1361// bytes are discarded when no consumer is supplied.
1362type terminalInputLifetime struct {
1363 eof chan error
1364 once sync.Once
1365 consumer pickerInputConsumer
1366 pump *terminalInputPump
1367
1368 mu sync.Mutex
1369 pickerID uint64
1370 // resumeOverflowed spans every watch of one outage: once held keys
1371 // overflowed, later phases drop keys too, so the resumed session never
1372 // receives input with a gap. The picker taking input clears it.
1373 resumeOverflowed atomic.Bool
1374}
1375
1376// startTerminalInputLifetime starts the single terminal read. A nil reader is
1377// treated as an immediate orderly EOF. A nil consumer discards read bytes; a
1378// non-nil consumer owns them for the picker presentation.
1379func startTerminalInputLifetime(in io.Reader, consumer pickerInputConsumer) *terminalInputLifetime {
1380 return startTerminalInputLifetimeWith(in, consumer, nil)
1381}
1382
1383// startTerminalInputLifetimeWith runs beforeConsumers on the started pump
1384// before the picker may claim it, so a terminal probe sees its responses
1385// first and hands every other byte on to the picker.
1386func startTerminalInputLifetimeWith(in io.Reader, consumer pickerInputConsumer, beforeConsumers func(*terminalInputPump)) *terminalInputLifetime {
1387 lifetime := &terminalInputLifetime{eof: make(chan error, 1), consumer: consumer}
1388 if supervisorNil(in) {
1389 lifetime.finish(io.EOF)
1390 return lifetime
1391 }
1392 lifetime.pump = newTerminalInputPump(in)
1393 lifetime.pump.start()
1394 if beforeConsumers != nil {
1395 beforeConsumers(lifetime.pump)
1396 }
1397 if consumer != nil {
1398 lifetime.acquirePicker()
1399 }
1400 go lifetime.runPicker()
1401 go func() {
1402 <-lifetime.pump.exited
1403 // Publish the cause the reader actually ended with. The reader enqueues
1404 // its final result for the consumer, but this watcher can win that race,
1405 // and reporting a read failure as an orderly EOF would return a clean
1406 // status for a terminal that failed.
1407 lifetime.finish(lifetime.pump.cause())
1408 }()
1409 return lifetime
1410}
1411
1412func (l *terminalInputLifetime) acquirePicker() {
1413 if l == nil || l.pump == nil || l.consumer == nil {
1414 return
1415 }
1416 l.mu.Lock()
1417 if l.pickerID != 0 {
1418 l.mu.Unlock()
1419 return
1420 }
1421 if id, ok := l.pump.tryClaim(); ok {
1422 l.pickerID = id
1423 l.resumeOverflowed.Store(false)
1424 l.mu.Unlock()
1425 return
1426 }
1427 l.mu.Unlock()
1428 // A finalizing attachment still owns the claim. Retry when that claim is
1429 // actually released rather than permanently abandoning picker input.
1430 go func() {
1431 select {
1432 case <-l.pump.claimChanged:
1433 l.acquirePicker()
1434 case <-l.pump.done:
1435 }
1436 }()
1437}
1438
1439func (l *terminalInputLifetime) releasePicker() {
1440 if l == nil || l.pump == nil {
1441 return
1442 }
1443 l.mu.Lock()
1444 id := l.pickerID
1445 l.pickerID = 0
1446 l.mu.Unlock()
1447 if id != 0 {
1448 l.pump.ack(id)
1449 // Drop picker-owned bytes that were queued but never presented before
1450 // changing owner class; they are not session input.
1451 l.pump.dropOwned(id)
1452 l.pump.revoke(id)
1453 }
1454}
1455
1456func (l *terminalInputLifetime) runPicker() {
1457 if l == nil || l.pump == nil {
1458 return
1459 }
1460 for {
1461 l.mu.Lock()
1462 id := l.pickerID
1463 l.mu.Unlock()
1464 if id == 0 {
1465 select {
1466 case <-l.pump.claimChanged:
1467 case <-l.pump.done:
1468 return
1469 }
1470 continue
1471 }
1472 result, ok := l.pump.take(context.Background(), id)
1473 if !ok {
1474 select {
1475 case <-l.pump.readyFor(id):
1476 case <-l.pump.claimChanged:
1477 case <-l.pump.done:
1478 return
1479 }
1480 continue
1481 }
1482 if len(result.data) != 0 {
1483 l.consumer.ConsumeTerminalRead(result.data)
1484 }
1485 l.pump.ack(id)
1486 if result.err != nil {
1487 l.finish(result.err)
1488 return
1489 }
1490 }
1491}
1492
1493func (l *terminalInputLifetime) stop() {
1494 if l == nil || l.pump == nil {
1495 return
1496 }
1497 l.pump.stop()
1498 l.pump.suspend()
1499}
1500
1501// resumeHeldInputLimit bounds keys kept while a lost attachment resumes.
1502// Past it further keys are dropped, like a full terminal buffer, until the
1503// picker takes input again.
1504const resumeHeldInputLimit = 64 << 10
1505
1506// resumeWatch reads the terminal while a lost attachment waits to resume.
1507// cancelled closes on Ctrl-C or a lone Esc; any other bytes are handed back
1508// to the pump on release, so the resumed attachment receives them in order.
1509type resumeWatch struct {
1510 cancelled chan struct{}
1511 stop chan struct{}
1512 done chan struct{}
1513 lifetime *terminalInputLifetime
1514 id uint64
1515 clock ports.Clock
1516 logger *slog.Logger
1517
1518 // backlog marks deliveries already waiting at claim: session input the
1519 // lost attachment or an earlier phase kept, held verbatim and never read
1520 // as a cancel key.
1521 backlog claimBacklog
1522 releaseOnce sync.Once
1523
1524 mu sync.Mutex
1525 held []byte
1526}
1527
1528// heldLen reports how many kept bytes the watch holds.
1529func (w *resumeWatch) heldLen() int {
1530 w.mu.Lock()
1531 defer w.mu.Unlock()
1532 return len(w.held)
1533}
1534
1535// hold keeps newly typed data within the bound. After an overflow in this
1536// outage it drops every new key, so the session never sees a gap.
1537func (w *resumeWatch) hold(data []byte) {
1538 w.holdInput(data, false)
1539}
1540
1541// holdResidual keeps the residual an earlier owner handed back. It was kept
1542// before any overflow, so it precedes the dropped keys and the sticky
1543// overflow must not discard it.
1544func (w *resumeWatch) holdResidual(data []byte) {
1545 w.holdInput(data, true)
1546}
1547
1548func (w *resumeWatch) holdInput(data []byte, residual bool) {
1549 w.mu.Lock()
1550 defer w.mu.Unlock()
1551 overflowed := &w.lifetime.resumeOverflowed
1552 if (!residual && overflowed.Load()) || len(w.held)+len(data) > resumeHeldInputLimit {
1553 if len(data) != 0 {
1554 overflowed.Store(true)
1555 }
1556 return
1557 }
1558 w.held = append(w.held, data...)
1559}
1560
1561// watchResume claims terminal input for one resume backoff. When the claim
1562// is unavailable it returns a watch that never cancels.
1563func (l *terminalInputLifetime) watchResume(clock ports.Clock, logger *slog.Logger) *resumeWatch {
1564 w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l, clock: clock, logger: logger}
1565 if l == nil || l.pump == nil {
1566 close(w.done)
1567 return w
1568 }
1569 id, backlog, ok := l.pump.claimWithBacklog()
1570 if !ok {
1571 close(w.done)
1572 return w
1573 }
1574 w.id = id
1575 w.backlog = backlog
1576 go w.run()
1577 return w
1578}
1579
1580func (w *resumeWatch) run() {
1581 defer close(w.done)
1582 pump := w.lifetime.pump
1583 var decoder resumeInputDecoder
1584 var escape ports.Timer
1585 defer func() {
1586 if escape != nil {
1587 escape.Stop()
1588 }
1589 w.hold(decoder.flush())
1590 }()
1591 stop := w.stop
1592 for {
1593 var escapeC <-chan time.Time
1594 if escape != nil {
1595 escapeC = escape.C()
1596 }
1597 result, ok := pump.take(context.Background(), w.id)
1598 if !ok {
1599 select {
1600 case <-pump.readyFor(w.id):
1601 case <-escapeC:
1602 if decoder.loneEscape() {
1603 close(w.cancelled)
1604 }
1605 return
1606 case <-stop:
1607 if !decoder.undecided() {
1608 return
1609 }
1610 // Resolve an escape typed just before the phase ended: the
1611 // next watch would hold it as backlog, so a cancel pressed at
1612 // the boundary would reach the session instead.
1613 stop = nil
1614 if escape == nil {
1615 escape = w.clock.NewTimer(resumeEscapeDelay)
1616 }
1617 case <-pump.done:
1618 return
1619 }
1620 continue
1621 }
1622 if result.err != nil {
1623 // Leave the final read for the terminal EOF watcher.
1624 pump.ack(w.id)
1625 w.hold(decoder.flush())
1626 w.hold(result.data)
1627 w.lifetime.finish(result.err)
1628 return
1629 }
1630 pump.ack(w.id)
1631 if w.backlog.residual {
1632 w.backlog.residual = false
1633 w.holdResidual(result.data)
1634 continue
1635 }
1636 if w.backlog.pending {
1637 // Read after the earlier owner stopped: newly typed, so an
1638 // overflow drops it, but it is still never a cancel key.
1639 w.backlog.pending = false
1640 w.hold(result.data)
1641 continue
1642 }
1643 held, cancel := decoder.feed(result.data)
1644 w.hold(held)
1645 if cancel {
1646 close(w.cancelled)
1647 return
1648 }
1649 if escape != nil {
1650 escape.Stop()
1651 escape = nil
1652 }
1653 if stop == nil && !decoder.undecided() {
1654 return
1655 }
1656 if decoder.loneEscape() || stop == nil {
1657 escape = w.clock.NewTimer(resumeEscapeDelay)
1658 }
1659 }
1660}
1661
1662// release stops the watch and hands kept keys back to the pump, unless the
1663// user cancelled, in which case they are dropped with the session. An escape
1664// still undecided is resolved first, so it may cancel. Later calls are no-ops.
1665func (w *resumeWatch) release() {
1666 if w.id == 0 {
1667 return
1668 }
1669 w.releaseOnce.Do(w.releaseClaim)
1670}
1671
1672func (w *resumeWatch) releaseClaim() {
1673 close(w.stop)
1674 <-w.done
1675 pump := w.lifetime.pump
1676 select {
1677 case <-w.cancelled:
1678 default:
1679 w.mu.Lock()
1680 if !pump.appendResidual(w.id, w.held, resumeHeldInputLimit) && len(w.held) != 0 {
1681 // Replaying only part of the held keys would leave a gap.
1682 w.lifetime.resumeOverflowed.Store(true)
1683 if w.logger != nil {
1684 w.logger.Warn("client_resume_input_dropped", "bytes", len(w.held))
1685 }
1686 }
1687 w.mu.Unlock()
1688 }
1689 pump.revoke(w.id)
1690}
1691
1692// finish publishes the terminal cause exactly once.
1693func (l *terminalInputLifetime) finish(err error) {
1694 l.once.Do(func() { l.eof <- err })
1695}
1696
1697// EOF is closed-result channel delivering the terminal read cause exactly once.
1698func (l *terminalInputLifetime) EOF() <-chan error { return l.eof }
1699
1700// supervisorNil reports whether a dependency is nil in any shape. A typed nil
1701// satisfies a plain != nil comparison, so the guard would defer the failure to
1702// the first method call, which panics; the reflection check keeps it total.
1703func supervisorNil(dependency any) bool {
1704 if dependency == nil {
1705 return true
1706 }
1707 value := reflect.ValueOf(dependency)
1708 switch value.Kind() {
1709 case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Ptr, reflect.Slice:
1710 return value.IsNil()
1711 default:
1712 return false
1713 }
1714}
1715
1716// detectTerminalCapabilities probes the outer terminal once and enables the
1717// kitty keyboard protocol when it is supported and wanted. It runs in Run
1718// before any input consumer or attachment exists.
1719func (s *Supervisor) detectTerminalCapabilities(ctx context.Context, pump *terminalInputPump) {
1720 env := s.cfg.AttachmentEnvironment
1721 if !env.ProbeTerminal {
1722 return
1723 }
1724 caps := probeTerminalCapabilities(ctx, s.cfg.Terminal, s.cfg.Clock, pump)
1725 if caps.KittyKeyboard && env.KittyKeyboard {
1726 // Normalize before the terminal switches, so no read escapes it.
1727 pump.keys.enable()
1728 if err := s.cfg.Terminal.EnableKittyKeyboard(kittykey.RequestFlags); err != nil {
1729 pump.keys.disable()
1730 s.logger.Warn("enabling kitty keyboard protocol failed", "err", err)
1731 caps.KittyKeyboard = false
1732 }
1733 } else {
1734 caps.KittyKeyboard = false
1735 }
1736 s.capabilities = caps
1737}