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