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