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