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 // pendingPickerKey retains a commit observed while connecting. It is
438 // revalidated against the adopted service just like a ready-phase commit.
439 pendingPickerKey string
440 preview previewManager
441
442 mu sync.Mutex
443 state State
444
445 // attachments is the single foreground host every committed attachment
446 // runs through. The supervisor owns it for the whole run
447 // and retains close authority over each stream it admits.
448 attachments *attachmentHost
449 // nextAttachment identifies one attachment run within the connection
450 // attempt generation. Logical stream IDs are allocated by the broker
451 // service, the single allocator on its own connection, so the supervisor
452 // keeps no allocator of its own.
453 nextAttachment atomic.Uint64
454 // navigation is the supervisor's private copy of the composition's intent,
455 // consumed exactly once. navigationConsumed is set before the intent is
456 // resolved, so a refusal, a cancellation, a timeout, or a reconnection can
457 // never re-arm it.
458 navigation InitialNavigation
459 navigationConsumed bool
460 // clientID is the stable client identity carried in every Hello this
461 // supervisor sends, across reconnects and attachments.
462 clientID [16]byte
463 // theme retains the terminal-reported colors across attachments, so a
464 // replacement attachment restores them before its own palette query ends.
465 theme terminalThemeState
466 // tabs remembers navigation only for this client process.
467 tabs sessionTabMemory
468 // capabilities is the outer terminal's probed and enabled capabilities,
469 // written once in Run before any attachment starts.
470 capabilities terminalCapabilities
471 // readySub is the adopted connection's subscription while the ready phase
472 // runs, so the picker overlay over a live attachment keeps folding broker
473 // publications. It is only touched from the run goroutine.
474 readySub ports.BrokerSubscription
475 // pendingSwap is the request the picker overlay or a daemon navigation
476 // committed to another target while an attachment was live. It is only
477 // touched from the run goroutine and consumed by settleAttachment.
478 pendingSwap *pickerAttachmentTarget
479 // pendingInPlace is the choice the live attachment is switching to in
480 // place on its own daemon. A refusal turns it into pendingSwap. Only
481 // touched from the run goroutine; cleared when the attachment settles.
482 pendingInPlace *pendingInPlace
483 // inPlaceSeq numbers in-place choices so a late outcome for an older one
484 // is never applied to a newer one.
485 inPlaceSeq uint64
486 // kills runs the picker's `x` operations off the run goroutine.
487 kills pickerKills
488 // routes is the client route ledger published to the serving daemon;
489 // routesSent is the attachment that received its latest snapshot. Both are
490 // only touched from the run goroutine.
491 routes *routeLedger
492 routesSent AttachmentToken
493}
494
495// NewSupervisor validates the required dependencies and returns a supervisor
496// parked on the picker with no connection. It does not touch the terminal; Run
497// enters raw mode.
498func NewSupervisor(cfg SupervisorConfig) (*Supervisor, error) {
499 if err := cfg.SessionEnvironment.Validate(); err != nil {
500 return nil, fmt.Errorf("client: invalid session environment: %w", err)
501 }
502 cfg.SessionEnvironment = cfg.SessionEnvironment.Clone()
503 if supervisorNil(cfg.Connector) {
504 return nil, errors.New("supervisor requires a broker connector")
505 }
506 if supervisorNil(cfg.Terminal) {
507 return nil, errors.New("supervisor requires a terminal")
508 }
509 if supervisorNil(cfg.Clock) {
510 return nil, errors.New("supervisor requires a clock")
511 }
512 if cfg.Render == nil {
513 cfg.Render = func(State) {}
514 }
515 if supervisorNil(cfg.Picker) {
516 // Normalize a typed-nil picker so Run's nil check and the input
517 // lifetime's consumer check are both safe.
518 cfg.Picker = nil
519 }
520 if cfg.Jitter == nil {
521 cfg.Jitter = rand.Float64
522 }
523 if cfg.ResolveInitialNavigation != nil {
524 if cfg.InitialNavigation != (InitialNavigation{}) {
525 return nil, errors.New("supervisor accepts either an initial navigation or a resolver, not both")
526 }
527 } else if err := cfg.InitialNavigation.Validate(); err != nil {
528 return nil, fmt.Errorf("supervisor initial navigation: %w", err)
529 }
530 logger := cfg.Logger
531 if logger == nil {
532 logger = slog.New(slog.NewTextHandler(io.Discard, nil))
533 }
534 supervisor := &Supervisor{
535 cfg: cfg,
536 logger: logger,
537 state: State{Presentation: PresentPicker, Connectivity: ConnectivityDisconnected},
538 clientID: newClientID(),
539 navigation: cfg.InitialNavigation,
540 preview: previewManager{clock: cfg.Clock},
541 }
542 // The host is the supervisor's existing foreground grant. It owns no raw
543 // mode and starts no reader: the supervisor keeps its one terminal input
544 // lifetime, so the attachment path adds neither a second reader nor a
545 // second writer. The optional UI binds to this host's existing pump claim;
546 // it never starts another terminal reader.
547 supervisor.attachments = newAttachmentHost(attachmentHostConfig{
548 Terminal: cfg.Terminal,
549 Clock: cfg.Clock,
550 Actions: cfg.LifecycleActions,
551 ActionUI: cfg.UI,
552 OnAttached: func(AttachmentToken) { supervisor.transition(supervisorEvent{kind: supervisorAttached}) },
553 })
554 return supervisor, nil
555}
556
557// initialNavigationPending reports whether an initial navigation other than
558// the picker is still armed. A resolver may still resolve to the picker; the
559// navigation then settles back to it without an attachment.
560func (s *Supervisor) initialNavigationPending() bool {
561 s.mu.Lock()
562 defer s.mu.Unlock()
563 if s.navigationConsumed {
564 return false
565 }
566 return s.cfg.ResolveInitialNavigation != nil || s.navigation.Kind != InitialNavigationPicker
567}
568
569// State returns the current immutable projection.
570func (s *Supervisor) State() State {
571 s.mu.Lock()
572 defer s.mu.Unlock()
573 return s.state
574}
575
576// Run owns the terminal for exactly its lifetime. It enters raw mode before it
577// connects to the broker, starts the single terminal input reader, and then
578// drives connect/subscribe/retry until the terminal reaches EOF, the process
579// context is cancelled, or a terminal broker failure settles the process. Raw
580// mode is restored exactly once on every exit path.
581func (s *Supervisor) Run(ctx context.Context) (retErr error) {
582 restore, err := s.cfg.Terminal.EnterRaw()
583 if err != nil {
584 return fmt.Errorf("entering raw mode: %w", err)
585 }
586 if restore == nil {
587 restore = func() error { return nil }
588 }
589 restoreTerminal := sync.OnceValue(restore)
590 defer func() {
591 if rerr := restoreTerminal(); rerr != nil && retErr == nil {
592 retErr = fmt.Errorf("restoring terminal: %w", rerr)
593 }
594 }()
595
596 input := startTerminalInputLifetimeWith(s.cfg.Terminal.In(), s.cfg.Picker, func(pump *terminalInputPump) {
597 s.detectTerminalCapabilities(ctx, pump)
598 })
599 defer input.stop()
600 s.attachments.setInput(input.pump)
601 s.attachments.startGeometry(ctx)
602 defer s.attachments.stopGeometry()
603
604 // The active connection is owned across loop iterations. retire is
605 // idempotent so every exit path closes exactly the connection it adopted.
606 var (
607 service ports.BrokerNavigator
608 sub ports.BrokerSubscription
609 )
610 retire := func() {
611 s.retirePickerKill()
612 s.preview.close(s.cfg.Picker)
613 s.readySub = nil
614 if sub != nil {
615 sub.Close()
616 sub = nil
617 }
618 if service != nil {
619 _ = service.Close()
620 service = nil
621 }
622 }
623 defer retire()
624
625 for {
626 // The generation is claimed before the connector runs, so the attempt is
627 // always attributable to the state that started it and a completion can be
628 // matched against the generation that produced it.
629 s.transition(supervisorEvent{kind: supervisorBeginAttempt, navigating: s.initialNavigationPending()})
630 generation := s.State().Generation
631 attemptCtx, cancelAttempt := context.WithCancel(ctx)
632 outcome := make(chan supervisorAttempt, 1)
633 go func() { outcome <- s.connectAttempt(attemptCtx, generation) }()
634
635 result, terminated, termErr := s.awaitAttempt(ctx, input, outcome)
636 if terminated {
637 // Connect must honor cancellation, so joining the attempt here both
638 // retires a late superseded success and leaves no goroutine behind.
639 cancelAttempt()
640 late := <-outcome
641 late.retire()
642 s.transition(supervisorEvent{kind: supervisorTerminal, err: termErr})
643 return termErr
644 }
645 cancelAttempt()
646 adopted, cause := s.adoptAttempt(ctx, result)
647 if !adopted {
648 // The cancelled or superseded attempt is retired rather than adopted, so
649 // it cannot drive the state of a newer attempt and a cancelled run never
650 // publishes a retry cadence, renders ConnectivityRetryWait, or notifies a
651 // connectivity failure for it.
652 result.retire()
653 if cause != nil {
654 s.transition(supervisorEvent{kind: supervisorTerminal, err: cause})
655 return cause
656 }
657 continue
658 }
659
660 if result.err != nil {
661 if supervisorTerminalFailure(result.err) {
662 s.transition(supervisorEvent{kind: supervisorTerminal, err: result.err})
663 return result.err
664 }
665 if !supervisorRetryable(result.err) {
666 // Retrying can never fix this. Return to the picker with the
667 // visible error and stop reconnecting; the process still owns
668 // the terminal until the user exits or it is cancelled.
669 s.transition(supervisorEvent{kind: supervisorNonRetryable, err: result.err})
670 s.notify(result.err)
671 terminalErr := s.awaitTermination(ctx, input)
672 s.transition(supervisorEvent{kind: supervisorTerminal, err: terminalErr})
673 return terminalErr
674 }
675 if terminated, termErr := s.retryFailure(ctx, input, supervisorTransientFailure, result.err); terminated {
676 return termErr
677 }
678 continue
679 }
680
681 service, sub = result.service, result.sub
682 s.readySub = sub
683 if s.cfg.Picker != nil {
684 // Render the first committed publication immediately, before waiting
685 // for the next one; the picker never shows a stale empty catalogue
686 // while an established connection already has state.
687 s.cfg.Picker.ApplySnapshot(service.Snapshot())
688 s.renderCurrent()
689 }
690 wasReadyLost := s.State().ReadyLost
691 s.transition(supervisorEvent{kind: supervisorReady})
692 s.refreshPreview(service)
693 if wasReadyLost {
694 s.notifyLifecycle(LifecycleNoticeBrokerReconnected)
695 s.mu.Lock()
696 s.state.ReadyLost = false
697 s.mu.Unlock()
698 }
699 if terminated, termErr := s.runInitialNavigation(ctx, input, service); terminated {
700 retire()
701 s.transition(supervisorEvent{kind: supervisorTerminal, err: termErr})
702 return termErr
703 }
704 // A navigation left armed (the broker was lost while it waited for an
705 // observation) keeps Connecting: the next attempt takes it again.
706 if !s.initialNavigationPending() {
707 s.transition(supervisorEvent{kind: supervisorNavigationSettled})
708 }
709
710 // Ready phase: fold publications, admit committed attachments one at a
711 // time, and return to the picker after each. A committed attachment
712 // never replaces the broker connection; only a broker loss leaves this
713 // loop, and cancellation or terminal EOF ends the run.
714 var lossErr error
715 ready:
716 for {
717 ready := s.awaitReady(ctx, input, service, sub)
718 if ready.terminated {
719 retire()
720 s.transition(supervisorEvent{kind: supervisorTerminal, err: ready.termErr})
721 return ready.termErr
722 }
723 if ready.commitKey == "" {
724 lossErr = ready.lossErr
725 break ready
726 }
727 if terminated, termErr := s.runCommittedAttachment(ctx, input, service, ready.commitKey); terminated {
728 retire()
729 s.transition(supervisorEvent{kind: supervisorTerminal, err: termErr})
730 return termErr
731 }
732 }
733 retire()
734 if cause := ctx.Err(); cause != nil {
735 // Parent cancellation wins over a loss that settled at the same instant,
736 // exactly as it does at the attempt settle. The connection was already
737 // released above, so the loss is simply never turned into a retry
738 // cadence, a ConnectivityRetryWait render, or a notification for a run
739 // that was already cancelled.
740 s.transition(supervisorEvent{kind: supervisorTerminal, err: cause})
741 return cause
742 }
743 s.notifyLifecycle(LifecycleNoticeBrokerLost)
744 if terminated, termErr := s.retryFailure(ctx, input, supervisorBrokerLoss, normalizeUnavailable(lossErr)); terminated {
745 return termErr
746 }
747 }
748}
749
750// supervisorAttempt is the settled outcome of one connect/subscribe/first
751// publication attempt. generation is the attempt's generation, claimed before
752// the connector ran, so the completion is always attributable to the attempt
753// that produced it (see adoptAttempt's generation fence).
754type supervisorAttempt struct {
755 generation uint64
756 service ports.BrokerNavigator
757 sub ports.BrokerSubscription
758 err error
759}
760
761// retire closes whatever the attempt produced. It is safe on a partially built
762// or empty attempt, and it treats a typed-nil service or subscription as absent
763// rather than calling into it.
764func (a supervisorAttempt) retire() {
765 if !supervisorNil(a.sub) {
766 a.sub.Close()
767 }
768 if !supervisorNil(a.service) {
769 _ = a.service.Close()
770 }
771}
772
773// adoptAttempt decides whether one settled attempt may be adopted, and reports the
774// parent cancellation that must settle the run instead. The decision is made under
775// the state lock, atomically with respect to the state it consults, so an attempt
776// that settled after the parent cancelled the run is never adopted even though
777// awaitAttempt's select may have taken its outcome over an already-ready
778// ctx.Done. The generation fence is enforced at the same decision point: a
779// superseded attempt is rejected too. It is a defensive invariant, not a
780// currently reachable production path, because Run joins each attempt before it
781// advances; it is kept so a future driver that admits more than one attempt at a
782// time can never let a late completion drive the state of a newer attempt.
783func (s *Supervisor) adoptAttempt(ctx context.Context, result supervisorAttempt) (adopted bool, cause error) {
784 s.mu.Lock()
785 defer s.mu.Unlock()
786 if cause := ctx.Err(); cause != nil {
787 return false, cause
788 }
789 if result.generation != s.state.Generation {
790 return false, nil
791 }
792 return true, nil
793}
794
795// retryFailure records one retryable connectivity failure, surfaces it, and
796// waits the cadence for the next attempt. It is the shared tail of the transient
797// connect/subscribe failure and the established-connection loss paths, so both
798// select the same observable transition, notification, and backoff.
799func (s *Supervisor) retryFailure(ctx context.Context, input *terminalInputLifetime, kind supervisorEventKind, err error) (bool, error) {
800 if cause := ctx.Err(); cause != nil {
801 s.transition(supervisorEvent{kind: supervisorTerminal, err: cause})
802 return true, cause
803 }
804 s.transition(supervisorEvent{kind: kind, err: err})
805 s.notify(err)
806 if terminated, termErr := s.waitBackoff(ctx, input); terminated {
807 s.transition(supervisorEvent{kind: supervisorTerminal, err: termErr})
808 return true, termErr
809 }
810 return false, nil
811}
812
813// connectAttempt connects once, subscribes, and waits for the first committed
814// publication. It returns a live service and subscription only when the
815// connection is Ready; every failure closes whatever it opened before
816// returning, so a caller has nothing to clean up on the error path. Every
817// connect, subscribe, or first-publication failure is stamped with generation
818// and normalized to a typed unavailable failure so the run loop never observes
819// an untyped cause.
820func (s *Supervisor) connectAttempt(ctx context.Context, generation uint64) supervisorAttempt {
821 s.logger.Debug("client_broker_connect_begin", "generation", generation)
822 service, err := s.cfg.Connector.Connect(ctx)
823 if err != nil {
824 failure := normalizeUnavailable(err)
825 s.logger.Warn("client_broker_connect_failed", "generation", generation, "code", brokerErrorCode(failure), "error", failure, "cause", errors.Unwrap(failure))
826 return supervisorAttempt{generation: generation, err: failure}
827 }
828 if supervisorNil(service) {
829 return supervisorAttempt{generation: generation, err: ports.BrokerError{
830 Code: ports.BrokerErrorUnavailable,
831 Text: "broker connector returned no service",
832 }}
833 }
834 sub, err := service.Subscribe()
835 if err != nil {
836 _ = service.Close()
837 failure := normalizeUnavailable(err)
838 s.logger.Warn("client_broker_subscribe_failed", "generation", generation, "code", brokerErrorCode(failure), "error", failure, "cause", errors.Unwrap(failure))
839 return supervisorAttempt{generation: generation, err: failure}
840 }
841 if supervisorNil(sub) {
842 // A service that reports no subscription cannot ever become Ready, so it
843 // is a typed unavailable failure rather than a nil dereference on the
844 // first Changed() call.
845 _ = service.Close()
846 return supervisorAttempt{generation: generation, err: ports.BrokerError{
847 Code: ports.BrokerErrorUnavailable,
848 Text: "broker service returned no subscription",
849 }}
850 }
851 for {
852 if service.Snapshot().Epoch != 0 {
853 return supervisorAttempt{generation: generation, service: service, sub: sub}
854 }
855 select {
856 case <-sub.Changed():
857 continue
858 case <-service.Done():
859 cause := normalizeUnavailable(service.Err())
860 s.logger.Warn("client_broker_initial_publication_lost", "generation", generation, "code", brokerErrorCode(cause), "error", cause, "cause", errors.Unwrap(cause))
861 sub.Close()
862 _ = service.Close()
863 return supervisorAttempt{generation: generation, err: cause}
864 case <-ctx.Done():
865 sub.Close()
866 _ = service.Close()
867 return supervisorAttempt{generation: generation, err: ctx.Err()}
868 }
869 }
870}
871
872// awaitAttempt waits for the in-flight attempt to settle, or for the run to be
873// terminated first.
874func (s *Supervisor) awaitAttempt(ctx context.Context, input *terminalInputLifetime, outcome <-chan supervisorAttempt) (supervisorAttempt, bool, error) {
875 for {
876 select {
877 case result := <-outcome:
878 return result, false, nil
879 case <-s.pickerOps():
880 if s.takePickerClose() {
881 return supervisorAttempt{}, true, nil
882 }
883 case <-ctx.Done():
884 return supervisorAttempt{}, true, ctx.Err()
885 case err := <-input.EOF():
886 return supervisorAttempt{}, true, terminalReadCause(err)
887 case <-s.presentationInvalidation():
888 s.renderResizeInvalidation()
889 case <-s.spinnerTick():
890 s.cfg.Spinner.AdvanceSpinner(s.State())
891 }
892 }
893}
894
895// readyOutcome is the result of one wait inside the established-connection
896// phase: either the broker connection was lost, the run was terminated, or a
897// committed picker selection is ready to be attached.
898type readyOutcome struct {
899 // commitKey is the exact catalogue key captured with a commit decision. A
900 // non-empty value names one attachment to admit; the ready loop never
901 // resolves anything else.
902 commitKey string
903 // lossErr is the broker service's stable terminal cause when the
904 // established connection was lost.
905 lossErr error
906 // terminated reports that the run itself must end, with termErr carrying
907 // the process cancellation or terminal-read cause.
908 terminated bool
909 termErr error
910}
911
912// awaitReady waits for an established connection to be lost, for a committed
913// picker selection, or for the run to be terminated first. Publications are
914// folded into the picker as they arrive; a commit decision wakes the loop so
915// the supervisor can admit exactly the row the user committed. It opens no
916// stream and starts no terminal work itself.
917func (s *Supervisor) awaitReady(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, sub ports.BrokerSubscription) readyOutcome {
918 if key := s.pendingPickerKey; key != "" {
919 s.pendingPickerKey = ""
920 return readyOutcome{commitKey: key}
921 }
922 var changed <-chan struct{}
923 var ops <-chan struct{}
924 if s.cfg.Picker != nil {
925 if !supervisorNil(sub) {
926 changed = sub.Changed()
927 }
928 ops = s.cfg.Picker.OpsReady()
929 }
930 notices := noticeWake{clock: s.cfg.Clock}
931 defer notices.stop()
932 for {
933 select {
934 case <-notices.arm(s.cfg.Picker):
935 s.renderCurrent()
936 case <-service.Done():
937 return readyOutcome{lossErr: service.Err()}
938 case <-changed:
939 s.cfg.Picker.ApplySnapshot(service.Snapshot())
940 s.refreshPreview(service)
941 s.renderCurrent()
942 case <-s.preview.changed():
943 if s.preview.publish(s.cfg.Picker) {
944 s.renderCurrent()
945 }
946 case outcome := <-s.kills.results():
947 s.finishPickerKill(service, outcome)
948 case <-ops:
949 op, key := s.cfg.Picker.TakeOp()
950 if op.kill && !op.commit && !op.close {
951 s.startPickerKill(service, key)
952 s.renderCurrent()
953 continue
954 }
955 if !op.close && !op.commit {
956 s.refreshPreview(service)
957 s.renderCurrent()
958 continue
959 }
960 if op.close {
961 if op.exit {
962 // Ctrl+C is the explicit exit. With no attachment to
963 // return to, it is the only close that ends the process.
964 s.preview.close(s.cfg.Picker)
965 return readyOutcome{terminated: true}
966 }
967 // Escape and q cancel; with no attachment there is nothing to
968 // cancel back to, so the picker stays and says how to leave.
969 s.notifyPicker(pickerExitHint)
970 s.renderCurrent()
971 continue
972 }
973 if op.commit && key != "" {
974 s.preview.close(s.cfg.Picker)
975 return readyOutcome{commitKey: key}
976 }
977 case <-ctx.Done():
978 return readyOutcome{terminated: true, termErr: ctx.Err()}
979 case err := <-input.EOF():
980 return readyOutcome{terminated: true, termErr: terminalReadCause(err)}
981 case <-s.presentationInvalidation():
982 s.refreshPreview(service)
983 s.renderResizeInvalidation()
984 }
985 }
986}
987
988func (s *Supervisor) refreshPreview(service ports.BrokerNavigator) {
989 if s == nil || s.cfg.Picker == nil {
990 return
991 }
992 geometry, err := s.cfg.Terminal.Geometry()
993 if err != nil || !geometry.Valid() {
994 s.preview.close(s.cfg.Picker)
995 return
996 }
997 s.preview.refresh(service, s.cfg.Picker, geometry.Size)
998}
999
1000// pickerOps is nil when no picker owns presentation decisions.
1001func (s *Supervisor) pickerOps() <-chan struct{} {
1002 if s.cfg.Picker == nil {
1003 return nil
1004 }
1005 return s.cfg.Picker.OpsReady()
1006}
1007
1008// takePickerClose honors the explicit exit while offline without losing a
1009// racing commit. A plain cancel (Escape, q) has no attachment to return to and
1010// never ends the process.
1011func (s *Supervisor) takePickerClose() bool {
1012 op, key := s.cfg.Picker.TakeOp()
1013 if op.commit && key != "" {
1014 s.pendingPickerKey = key
1015 }
1016 if op.close && !op.exit {
1017 s.notifyPicker(pickerExitHint)
1018 s.renderCurrent()
1019 }
1020 if op.kill && !op.commit && !op.close && key != "" {
1021 // A kill needs the broker's control stream; it is refused, not queued.
1022 s.notifyPicker(pickerKillRefused("couldn't kill: broker unavailable"))
1023 s.renderCurrent()
1024 }
1025 return op.close && op.exit
1026}
1027
1028// pickerPresentationHost is the optional presentation surface of the real
1029// picker. Scripted pickers may omit it.
1030type pickerPresentationHost interface {
1031 Notify(n domain.Notification)
1032 invalidatePresentation()
1033}
1034
1035// notifyPicker shows one bounded client-local notice when the picker
1036// supports it.
1037func (s *Supervisor) notifyPicker(n domain.Notification) {
1038 if host, ok := s.cfg.Picker.(pickerPresentationHost); ok {
1039 host.Notify(n)
1040 }
1041}
1042
1043// noticeDeadliner reports when the picker's notice stack next changes.
1044type noticeDeadliner interface {
1045 noticeDeadline() (time.Time, bool)
1046}
1047
1048// noticeWake wakes the ready loop when a picker notice expires, so the stack
1049// advances and freed cells are blanked without waiting for other input.
1050type noticeWake struct {
1051 clock ports.Clock
1052 timer ports.Timer
1053 due time.Time
1054}
1055
1056// arm returns a channel that fires at the next notice deadline, or nil when
1057// no notice will change on its own.
1058func (w *noticeWake) arm(picker pickerHost) <-chan time.Time {
1059 host, ok := picker.(noticeDeadliner)
1060 if !ok || supervisorNil(w.clock) {
1061 return nil
1062 }
1063 due, ok := host.noticeDeadline()
1064 if !ok {
1065 w.stop()
1066 return nil
1067 }
1068 if w.timer == nil || !due.Equal(w.due) {
1069 w.stop()
1070 w.timer = w.clock.NewTimer(max(due.Sub(w.clock.Now()), 0))
1071 w.due = due
1072 }
1073 return w.timer.C()
1074}
1075
1076func (w *noticeWake) stop() {
1077 stopSupervisorTimer(w.timer)
1078 w.timer = nil
1079 w.due = time.Time{}
1080}
1081
1082// pickerExitHint tells an unattached user how to leave the picker.
1083var pickerExitHint = domain.Notification{Code: domain.NoticeUser, Severity: domain.NoticeInfo, Message: "no session attached: press Ctrl+C to quit"}
1084
1085// invalidatePickerPresentation forces the next picker frame to redraw the
1086// whole box after another owner wrote the terminal.
1087func (s *Supervisor) invalidatePickerPresentation() {
1088 if host, ok := s.cfg.Picker.(pickerPresentationHost); ok {
1089 host.invalidatePresentation()
1090 }
1091}
1092
1093// awaitTermination parks the supervisor on the picker after a non-retryable
1094// failure until the terminal ends or the process is cancelled. A clean terminal
1095// EOF returns nil; every other cause is reported.
1096func (s *Supervisor) awaitTermination(ctx context.Context, input *terminalInputLifetime) error {
1097 notices := noticeWake{clock: s.cfg.Clock}
1098 defer notices.stop()
1099 for {
1100 select {
1101 case <-s.pickerOps():
1102 if s.takePickerClose() {
1103 return nil
1104 }
1105 case <-ctx.Done():
1106 return ctx.Err()
1107 case err := <-input.EOF():
1108 return terminalReadCause(err)
1109 case <-s.presentationInvalidation():
1110 s.renderResizeInvalidation()
1111 case <-notices.arm(s.cfg.Picker):
1112 s.renderCurrent()
1113 }
1114 }
1115}
1116
1117// waitBackoff waits one equal-jitter retry interval for the current failure
1118// count, or settles on cancellation or terminal EOF. The timer is always
1119// stopped and drained, so no timer outlives the attempt transition.
1120func (s *Supervisor) waitBackoff(ctx context.Context, input *terminalInputLifetime) (bool, error) {
1121 delay := supervisorBackoffDelay(s.State().Attempt, s.cfg.Jitter)
1122 if delay <= 0 {
1123 return false, nil
1124 }
1125 timer := s.cfg.Clock.NewTimer(delay)
1126 defer stopSupervisorTimer(timer)
1127 notices := noticeWake{clock: s.cfg.Clock}
1128 defer notices.stop()
1129 for {
1130 select {
1131 case <-s.pickerOps():
1132 if s.takePickerClose() {
1133 return true, nil
1134 }
1135 case <-timer.C():
1136 return false, nil
1137 case <-ctx.Done():
1138 return true, ctx.Err()
1139 case err := <-input.EOF():
1140 return true, terminalReadCause(err)
1141 case <-s.presentationInvalidation():
1142 s.renderResizeInvalidation()
1143 case <-s.spinnerTick():
1144 s.cfg.Spinner.AdvanceSpinner(s.State())
1145 case <-notices.arm(s.cfg.Picker):
1146 s.renderCurrent()
1147 }
1148 }
1149}
1150
1151// supervisorBackoffDelay is the equal-jitter exponential interval for one
1152// failure count: the cap doubles from supervisorRetryInitial to
1153// supervisorRetryMax, and the interval is half the cap plus a uniform fraction
1154// of the other half, so attempts spread across [cap/2, cap). It is pure so the
1155// cadence can be tested exactly.
1156func supervisorBackoffDelay(attempt uint64, jitter func() float64) time.Duration {
1157 if attempt == 0 {
1158 return 0
1159 }
1160 limit := supervisorRetryInitial
1161 for i := uint64(1); i < attempt && limit < supervisorRetryMax; i++ {
1162 limit *= 2
1163 }
1164 if limit > supervisorRetryMax {
1165 limit = supervisorRetryMax
1166 }
1167 half := limit / 2
1168 fraction := jitter()
1169 if !(fraction >= 0) {
1170 // NaN is unordered: both fraction < 0 and fraction > 1 are false for
1171 // it, so an explicit non-negative test maps NaN and every negative value
1172 // to the minimum fraction. Without this, a NaN would reach the duration
1173 // conversion and yield an undefined interval instead of the documented
1174 // half-cap minimum.
1175 fraction = 0
1176 }
1177 if fraction > 1 {
1178 fraction = 1
1179 }
1180 return half + time.Duration(fraction*float64(half))
1181}
1182
1183// terminalReadCause maps a terminal read end to the Run result: a clean EOF is
1184// an orderly exit (nil), any other read failure is reported.
1185func terminalReadCause(err error) error {
1186 if err == nil || errors.Is(err, io.EOF) {
1187 return nil
1188 }
1189 return err
1190}
1191
1192// stopSupervisorTimer stops and drains a timer, supporting buffered timer
1193// adapters with pre-Go-1.23 reset semantics.
1194func stopSupervisorTimer(timer ports.Timer) {
1195 if timer == nil {
1196 return
1197 }
1198 if !timer.Stop() {
1199 select {
1200 case <-timer.C():
1201 default:
1202 }
1203 }
1204}
1205
1206// supervisorTerminalFailure reports whether a broker failure must end the
1207// process. Only the domain's terminal codes do; every other failure returns the
1208// client to the picker.
1209func supervisorTerminalFailure(err error) bool {
1210 var typed ports.BrokerError
1211 return errors.As(err, &typed) && typed.Terminal()
1212}
1213
1214// supervisorRetryable reports whether a failure is worth another attempt after
1215// backoff. An incompatible carriage or a conflicting policy can never succeed
1216// by retrying, so it is surfaced once and automatic reconnection stops; every
1217// other failure is treated as transient.
1218func supervisorRetryable(err error) bool {
1219 var typed ports.BrokerError
1220 if errors.As(err, &typed) {
1221 switch typed.Code {
1222 case ports.BrokerErrorIncompatible, ports.BrokerErrorConflictingPolicy:
1223 return false
1224 }
1225 }
1226 return true
1227}
1228
1229// normalizeUnavailable presents one connect, subscribe, or connection-loss
1230// failure as a typed ports.BrokerError. A cause that is already a BrokerError is
1231// returned unchanged, so its own code (for example incompatible or conflicting
1232// policy) still selects the non-retryable path; every untyped failure, including
1233// a nil cause, becomes an unavailable BrokerError that preserves the original
1234// error as its local-only Cause.
1235func brokerErrorCode(err error) string {
1236 var typed ports.BrokerError
1237 if errors.As(err, &typed) {
1238 return typed.Code.String()
1239 }
1240 return "unknown"
1241}
1242
1243func normalizeUnavailable(err error) error {
1244 if err == nil {
1245 return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker connection lost"}
1246 }
1247 var typed ports.BrokerError
1248 if errors.As(err, &typed) {
1249 return err
1250 }
1251 var admission ports.BrokerAdmissionError
1252 if errors.As(err, &admission) {
1253 return err
1254 }
1255 return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker connection lost", Cause: err}
1256}
1257
1258// notify surfaces a connectivity failure through the optional notifier.
1259func (s *Supervisor) notifyLifecycle(kind LifecycleNoticeKind) {
1260 if s != nil {
1261 s.logger.Debug("client_lifecycle_notice", "kind", kind)
1262 }
1263 if s != nil && s.cfg.NotifyLifecycle != nil {
1264 s.cfg.NotifyLifecycle(LifecycleNotice{Kind: kind})
1265 }
1266}
1267
1268func (s *Supervisor) notify(err error) {
1269 if err != nil {
1270 s.logger.Warn("client_connectivity_failure", "state", s.State().Connectivity.String(), "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
1271 }
1272 if s.cfg.Notify == nil || err == nil {
1273 return
1274 }
1275 s.cfg.Notify(s.State(), err)
1276}
1277
1278// transition applies one reducer event under the supervisor mutex and renders
1279// the resulting state outside it, so a renderer that reads State back never
1280// deadlocks.
1281//
1282// A presentation change also publishes the new presentation through the UI
1283// owner inside the same fence, so a capture observes the presentation the
1284// supervisor actually entered instead of an older attachment context. The
1285// publication is deliberately limited to the unattached presentations the
1286// supervisor owns: Picker and Connecting carry the run's handle and status with
1287// every session field zeroed, while Attached is published by the admitted
1288// attachment foreground only, after its first frame was written, flushed, and
1289// committed, carrying the real action generation. Terminating closes the UI
1290// service and publishes nothing.
1291func (s *Supervisor) transition(event supervisorEvent) {
1292 s.mu.Lock()
1293 previous := s.state.Presentation
1294 s.state = reduceSupervisor(s.state, event)
1295 state := s.state
1296 s.mu.Unlock()
1297 if state.Presentation != previous || event.kind == supervisorBeginAttempt || event.kind == supervisorTransientFailure || event.kind == supervisorBrokerLoss {
1298 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))
1299 }
1300 if state.Presentation != previous {
1301 s.publishPresentation(state.Presentation)
1302 }
1303 if s.cfg.Spinner != nil && state.Presentation != PresentConnecting && state.Connectivity != ConnectivityRetryWait && state.Connectivity != ConnectivityConnectingBroker {
1304 s.cfg.Spinner.StopSpinner()
1305 }
1306 if s.cfg.Render != nil {
1307 s.cfg.Render(state)
1308 }
1309}
1310
1311func (s *Supervisor) spinnerTick() <-chan time.Time {
1312 if s.cfg.Spinner == nil {
1313 return nil
1314 }
1315 return s.cfg.Spinner.Spinner()
1316}
1317
1318// publishPresentation publishes one supervisor-owned presentation through the
1319// optional UI owner. Attached and terminating are refused: Attached belongs to
1320// the admitted attachment foreground, and terminating ends the service.
1321func (s *Supervisor) publishPresentation(presentation Presentation) {
1322 if s == nil || s.cfg.UI == nil {
1323 return
1324 }
1325 switch presentation {
1326 case PresentPicker:
1327 _ = s.cfg.UI.publishPresentation(ports.UIStatusPicker)
1328 case PresentConnecting:
1329 _ = s.cfg.UI.publishPresentation(ports.UIStatusConnecting)
1330 }
1331}
1332
1333// renderCurrent repaints presentation data that changed without a supervisor
1334// state transition, notably a picker catalogue publication.
1335func (s *Supervisor) renderCurrent() {
1336 if s == nil || s.cfg.Render == nil {
1337 return
1338 }
1339 s.cfg.Render(s.State())
1340}
1341
1342// presentationInvalidation returns the coalesced resize invalidation the
1343// supervisor's serialized waits select on, or nil when there is no attachment
1344// host to collect resizes. A nil signal is simply never ready, so a supervisor
1345// with no host (the zero value, or one whose host was detached) parks on its
1346// remaining arms instead of dereferencing a nil host. Selecting on a nil
1347// channel is safe in both directions. The returned channel is receive-only: the
1348// supervisor only ever consumes an invalidation, and the attachment settlement
1349// wait deliberately does not select on it at all, so an invalidation raised
1350// while a foreground is admitted stays buffered for the next picker state.
1351func (s *Supervisor) presentationInvalidation() <-chan struct{} {
1352 if s == nil || s.attachments == nil {
1353 return nil
1354 }
1355 return s.attachments.presentationUpdate
1356}
1357
1358// renderResizeInvalidation runs only on the supervisor's serialized control
1359// path, so resize collection cannot write the terminal and an attached
1360// foreground is never painted over. The composition's renderer is invoked only
1361// for a presentation PickerPresentation admits, which is exactly PresentPicker;
1362// it samples Terminal.Geometry at render time. A pre-attachment connecting
1363// presentation is refused here for the same reason as an attached one: the
1364// admitted foreground already owns the same terminal writer.
1365func (s *Supervisor) renderResizeInvalidation() {
1366 if s == nil || s.cfg.Render == nil {
1367 return
1368 }
1369 state := s.State()
1370 if PickerPresentation(state) {
1371 s.cfg.Render(state)
1372 }
1373}
1374
1375// terminalInputLifetime owns the single read of the controlling terminal. It is
1376// started once and never replaced: a bare io.Reader cannot be interrupted, so
1377// the goroutine may outlive Run when stdin never reaches EOF, exactly like the
1378// attach client's input pump. It never closes caller-owned input. When a picker
1379// consumer is supplied each read is handed to it from this one
1380// reader, so the picker never starts a second reader; without a consumer the
1381// bytes are discarded when no consumer is supplied.
1382type terminalInputLifetime struct {
1383 eof chan error
1384 once sync.Once
1385 consumer pickerInputConsumer
1386 pump *terminalInputPump
1387
1388 mu sync.Mutex
1389 pickerID uint64
1390 // resumeOverflowed spans every watch of one outage: once held keys
1391 // overflowed, later phases drop keys too, so the resumed session never
1392 // receives input with a gap. The picker taking input clears it.
1393 resumeOverflowed atomic.Bool
1394}
1395
1396// startTerminalInputLifetime starts the single terminal read. A nil reader is
1397// treated as an immediate orderly EOF. A nil consumer discards read bytes; a
1398// non-nil consumer owns them for the picker presentation.
1399func startTerminalInputLifetime(in io.Reader, consumer pickerInputConsumer) *terminalInputLifetime {
1400 return startTerminalInputLifetimeWith(in, consumer, nil)
1401}
1402
1403// startTerminalInputLifetimeWith runs beforeConsumers on the started pump
1404// before the picker may claim it, so a terminal probe sees its responses
1405// first and hands every other byte on to the picker.
1406func startTerminalInputLifetimeWith(in io.Reader, consumer pickerInputConsumer, beforeConsumers func(*terminalInputPump)) *terminalInputLifetime {
1407 lifetime := &terminalInputLifetime{eof: make(chan error, 1), consumer: consumer}
1408 if supervisorNil(in) {
1409 lifetime.finish(io.EOF)
1410 return lifetime
1411 }
1412 lifetime.pump = newTerminalInputPump(in)
1413 lifetime.pump.start()
1414 if beforeConsumers != nil {
1415 beforeConsumers(lifetime.pump)
1416 }
1417 if consumer != nil {
1418 lifetime.acquirePicker()
1419 }
1420 go lifetime.runPicker()
1421 go func() {
1422 <-lifetime.pump.exited
1423 // Publish the cause the reader actually ended with. The reader enqueues
1424 // its final result for the consumer, but this watcher can win that race,
1425 // and reporting a read failure as an orderly EOF would return a clean
1426 // status for a terminal that failed.
1427 lifetime.finish(lifetime.pump.cause())
1428 }()
1429 return lifetime
1430}
1431
1432func (l *terminalInputLifetime) acquirePicker() {
1433 if l == nil || l.pump == nil || l.consumer == nil {
1434 return
1435 }
1436 l.mu.Lock()
1437 if l.pickerID != 0 {
1438 l.mu.Unlock()
1439 return
1440 }
1441 if id, ok := l.pump.tryClaim(); ok {
1442 l.pickerID = id
1443 l.resumeOverflowed.Store(false)
1444 l.mu.Unlock()
1445 return
1446 }
1447 l.mu.Unlock()
1448 // A finalizing attachment still owns the claim. Retry when that claim is
1449 // actually released rather than permanently abandoning picker input.
1450 go func() {
1451 select {
1452 case <-l.pump.claimChanged:
1453 l.acquirePicker()
1454 case <-l.pump.done:
1455 }
1456 }()
1457}
1458
1459func (l *terminalInputLifetime) releasePicker() {
1460 if l == nil || l.pump == nil {
1461 return
1462 }
1463 l.mu.Lock()
1464 id := l.pickerID
1465 l.pickerID = 0
1466 l.mu.Unlock()
1467 if id != 0 {
1468 l.pump.ack(id)
1469 // Drop picker-owned bytes that were queued but never presented before
1470 // changing owner class; they are not session input.
1471 l.pump.dropOwned(id)
1472 l.pump.revoke(id)
1473 }
1474}
1475
1476func (l *terminalInputLifetime) runPicker() {
1477 if l == nil || l.pump == nil {
1478 return
1479 }
1480 for {
1481 l.mu.Lock()
1482 id := l.pickerID
1483 l.mu.Unlock()
1484 if id == 0 {
1485 select {
1486 case <-l.pump.claimChanged:
1487 case <-l.pump.done:
1488 return
1489 }
1490 continue
1491 }
1492 result, ok := l.pump.take(context.Background(), id)
1493 if !ok {
1494 select {
1495 case <-l.pump.readyFor(id):
1496 case <-l.pump.claimChanged:
1497 case <-l.pump.done:
1498 return
1499 }
1500 continue
1501 }
1502 if len(result.data) != 0 {
1503 l.consumer.ConsumeTerminalRead(result.data)
1504 }
1505 l.pump.ack(id)
1506 if result.err != nil {
1507 l.finish(result.err)
1508 return
1509 }
1510 }
1511}
1512
1513func (l *terminalInputLifetime) stop() {
1514 if l == nil || l.pump == nil {
1515 return
1516 }
1517 l.pump.stop()
1518 l.pump.suspend()
1519}
1520
1521// resumeHeldInputLimit bounds keys kept while a lost attachment resumes.
1522// Past it further keys are dropped, like a full terminal buffer, until the
1523// picker takes input again.
1524const resumeHeldInputLimit = 64 << 10
1525
1526// resumeWatch reads the terminal while a lost attachment waits to resume.
1527// cancelled closes on Ctrl-C or a lone Esc; any other bytes are handed back
1528// to the pump on release, so the resumed attachment receives them in order.
1529type resumeWatch struct {
1530 cancelled chan struct{}
1531 stop chan struct{}
1532 done chan struct{}
1533 lifetime *terminalInputLifetime
1534 id uint64
1535 clock ports.Clock
1536 logger *slog.Logger
1537
1538 // backlog counts deliveries already waiting at claim: session input the
1539 // lost attachment kept, held verbatim and never read as a cancel key.
1540 backlog int
1541 releaseOnce sync.Once
1542
1543 mu sync.Mutex
1544 held []byte
1545}
1546
1547// heldLen reports how many kept bytes the watch holds.
1548func (w *resumeWatch) heldLen() int {
1549 w.mu.Lock()
1550 defer w.mu.Unlock()
1551 return len(w.held)
1552}
1553
1554// hold keeps data within the bound.
1555func (w *resumeWatch) hold(data []byte) {
1556 w.mu.Lock()
1557 defer w.mu.Unlock()
1558 overflowed := &w.lifetime.resumeOverflowed
1559 if overflowed.Load() || len(w.held)+len(data) > resumeHeldInputLimit {
1560 if len(data) != 0 {
1561 overflowed.Store(true)
1562 }
1563 return
1564 }
1565 w.held = append(w.held, data...)
1566}
1567
1568// watchResume claims terminal input for one resume backoff. When the claim
1569// is unavailable it returns a watch that never cancels.
1570func (l *terminalInputLifetime) watchResume(clock ports.Clock, logger *slog.Logger) *resumeWatch {
1571 w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l, clock: clock, logger: logger}
1572 if l == nil || l.pump == nil {
1573 close(w.done)
1574 return w
1575 }
1576 id, backlog, ok := l.pump.claimWithBacklog()
1577 if !ok {
1578 close(w.done)
1579 return w
1580 }
1581 w.id = id
1582 w.backlog = backlog
1583 go w.run()
1584 return w
1585}
1586
1587func (w *resumeWatch) run() {
1588 defer close(w.done)
1589 pump := w.lifetime.pump
1590 var decoder resumeInputDecoder
1591 var escape ports.Timer
1592 defer func() {
1593 if escape != nil {
1594 escape.Stop()
1595 }
1596 w.hold(decoder.flush())
1597 }()
1598 stop := w.stop
1599 for {
1600 var escapeC <-chan time.Time
1601 if escape != nil {
1602 escapeC = escape.C()
1603 }
1604 result, ok := pump.take(context.Background(), w.id)
1605 if !ok {
1606 select {
1607 case <-pump.readyFor(w.id):
1608 case <-escapeC:
1609 if decoder.loneEscape() {
1610 close(w.cancelled)
1611 }
1612 return
1613 case <-stop:
1614 if !decoder.undecided() {
1615 return
1616 }
1617 // Resolve an escape typed just before the phase ended: the
1618 // next watch would hold it as backlog, so a cancel pressed at
1619 // the boundary would reach the session instead.
1620 stop = nil
1621 if escape == nil {
1622 escape = w.clock.NewTimer(resumeEscapeDelay)
1623 }
1624 case <-pump.done:
1625 return
1626 }
1627 continue
1628 }
1629 if result.err != nil {
1630 // Leave the final read for the terminal EOF watcher.
1631 pump.ack(w.id)
1632 w.hold(decoder.flush())
1633 w.hold(result.data)
1634 w.lifetime.finish(result.err)
1635 return
1636 }
1637 pump.ack(w.id)
1638 if w.backlog > 0 {
1639 w.backlog--
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 if err := s.cfg.Terminal.EnableKittyKeyboard(kittykey.OuterFlags); err != nil {
1727 s.logger.Warn("enabling kitty keyboard protocol failed", "err", err)
1728 caps.KittyKeyboard = false
1729 }
1730 } else {
1731 caps.KittyKeyboard = false
1732 }
1733 s.capabilities = caps
1734}