1package client
2
3import (
4 "crypto/sha256"
5 "encoding/base64"
6 "encoding/binary"
7 "strconv"
8 "strings"
9 "sync"
10 "time"
11
12 "github.com/bnema/vev/internal/domain"
13 "github.com/bnema/vev/internal/ports"
14 "github.com/bnema/vev/internal/protocol"
15 "github.com/bnema/vev/internal/protocol/catalogue"
16)
17
18// Client-owned picker catalogue.
19//
20// The autonomous supervisor owns the broker connection and the single terminal
21// input lifetime; this file owns the catalogue the picker presents, projected
22// from the broker's own observations rather than from a serving-daemon
23// catalogue. It projects broker observations instead of the daemon-served
24// protocol.PickerSnapshot: rows are derived from ports.BrokerSnapshot, the
25// opaque keys stay client-local, and a selection resolves to an exact
26// ports.BrokerOpenStreamRequest instead of a daemon-side key handoff.
27//
28// The projection preserves configured authority and observed state separately,
29// exactly as ports.BrokerDaemonObservation does:
30//
31// - daemon identity is the local flag or the remote endpoint plus the
32// registration incarnation; a re-registered endpoint is an identity
33// replacement and its rows never survive.
34// - session identity is the catalogue lifecycle ID; a same-name replacement
35// resolves as "replaced", never by name.
36// - freshness is the LastSuccess window against the injected clock.
37// - compatibility is the observed protocol version against the policy's
38// required version.
39// - availability is the observed RemoteAvailability.
40//
41// Updates are monotonic: a publication that does not supersede the applied
42// epoch/revision is ignored, and a per-daemon observation that regresses
43// (older LastSuccess and no newer failure episode, same incarnation) keeps the
44// newer observed state while still refreshing configured authority. Row order
45// is stable: hosts keep their first-seen position and unrelated updates never
46// reshuffle them. Nothing here attaches to a daemon, starts one, or writes a
47// session.
48
49const (
50 // pickerCatalogueSourceID names the single client-owned picker source.
51 pickerCatalogueSourceID = "broker"
52 // pickerCatalogueDefaultFreshness bounds how old a LastSuccess may be before
53 // a row is displayed as stale. Freshness is presentation information and
54 // never a resolution lock: a stale row is dimmed and badged, but an explicit
55 // attach or creation may still be attempted, because only the destination
56 // revalidates the exact identity.
57 pickerCatalogueDefaultFreshness = 30 * time.Second
58 // pickerCatalogueMaxRetiredRefs bounds the retired-key fence used to
59 // distinguish "gone" from "replaced" after a row leaves the projection.
60 pickerCatalogueMaxRetiredRefs = 128
61)
62
63// pickerCatalogueErrorCode is the closed refusal taxonomy for a client-owned
64// picker selection. The zero value is never produced: a refusal always names
65// why the exact selection cannot be resolved.
66type pickerCatalogueErrorCode uint8
67
68const (
69 pickerCatalogueUnknown pickerCatalogueErrorCode = iota + 1
70 pickerCatalogueGone
71 pickerCatalogueReplaced
72 pickerCatalogueIncompatible
73 pickerCatalogueUnavailable
74 pickerCatalogueEpochStale
75 pickerCatalogueInvalidName
76 pickerCatalogueNoSelection
77)
78
79func (c pickerCatalogueErrorCode) String() string {
80 switch c {
81 case pickerCatalogueUnknown:
82 return "unknown"
83 case pickerCatalogueGone:
84 return "gone"
85 case pickerCatalogueReplaced:
86 return "replaced"
87 case pickerCatalogueIncompatible:
88 return "incompatible"
89 case pickerCatalogueUnavailable:
90 return "unavailable"
91 case pickerCatalogueEpochStale:
92 return "stale_epoch"
93 case pickerCatalogueInvalidName:
94 return "invalid_name"
95 case pickerCatalogueNoSelection:
96 return "no_selection"
97 default:
98 return "unknown"
99 }
100}
101
102// pickerCatalogueError is a typed, bounded refusal. It never carries a raw
103// endpoint, credential, or adapter diagnostic as display text.
104type pickerCatalogueError struct {
105 Code pickerCatalogueErrorCode
106 Text string
107 // Notice is the bounded user-facing sentence shown for this refusal when
108 // the generic "picker selection <code>" wording would name the wrong
109 // thing, as for a remote host known to be failing.
110 Notice string
111}
112
113// noticeText is the sentence a picker toast shows for this refusal.
114func (e pickerCatalogueError) noticeText() string {
115 if e.Notice != "" {
116 return e.Notice
117 }
118 return e.Error()
119}
120
121func (e pickerCatalogueError) Error() string {
122 if e.Text == "" {
123 return "picker selection " + e.Code.String()
124 }
125 return "picker selection " + e.Code.String() + ": " + e.Text
126}
127
128// pickerSelectionKind names one closed admission variant a resolved selection
129// can carry.
130type pickerSelectionKind uint8
131
132const (
133 // pickerSelectionExact attaches to one exact session lifecycle.
134 pickerSelectionExact pickerSelectionKind = iota + 1
135 // pickerSelectionCreateNamed creates one named session.
136 pickerSelectionCreateNamed
137 // pickerSelectionCreateEphemeral creates one ephemeral session.
138 pickerSelectionCreateEphemeral
139 // pickerSelectionAttachNamed attaches to a session the daemon resolves by
140 // name. It is produced only by an initial CLI navigation, never a row.
141 pickerSelectionAttachNamed
142)
143
144// pickerSelectionRef is the client's exact selection identity, captured when a
145// row was projected. Resolution revalidates it against the latest snapshot; it
146// is never reconstructed from a display label.
147type pickerSelectionRef struct {
148 kind pickerSelectionKind
149 epoch ports.BrokerEpoch
150 local bool
151 endpoint string
152 registration domain.RemoteRegistration
153 lifecycle domain.SessionLifecycleID
154 name string
155 createName string
156 // tab names one tab row inside the session; its zero value is a
157 // session-level selection.
158 tab pickerTabRef
159}
160
161// pickerTabRef is the exact tab a tab row names, captured when it was
162// projected: the stable ID when the catalogue carries one, and the ordinal,
163// raw name, and tab count a stopped remote selector needs otherwise.
164type pickerTabRef struct {
165 present bool
166 id domain.TabStableID
167 index int
168 name string
169 count int
170}
171
172// pickerResolveBase carries the live broker connection identity a resolved
173// request must be fenced to. The catalogue supplies everything else.
174type pickerResolveBase struct {
175 Connection ports.BrokerConnectionID
176 Stream ports.BrokerStreamID
177 Env []string
178}
179
180// pickerHostState is one projected daemon and its newest accepted
181// observation.
182type pickerHostState struct {
183 observation ports.BrokerDaemonObservation
184}
185
186// pickerCatalogueConfig supplies the fresh-window clock. Clock is optional and
187// falls back to the system clock; Freshness defaults to
188// pickerCatalogueDefaultFreshness.
189type pickerCatalogueConfig struct {
190 Clock ports.Clock
191 Freshness time.Duration
192}
193
194// pickerCatalogue is the client-owned, monotonic projection of broker daemon
195// observations into ordered picker rows. It is safe for concurrent use: the
196// supervisor applies publications from its run path while the picker loop
197// resolves selections.
198type pickerCatalogue struct {
199 clock ports.Clock
200 freshness time.Duration
201
202 mu sync.Mutex
203 snapshot ports.BrokerSnapshot
204 hosts map[string]*pickerHostState
205 order []string
206 refs map[string]pickerSelectionRef
207 retired map[string]pickerSelectionRef
208 retiredOrder []string
209 // lines/cursor are the grouped projection; recent/recentCursor the flat
210 // recency projection. Both carry the same keys and actions.
211 lines []protocol.PickerLine
212 cursor protocol.PickerCursor
213 recent []protocol.PickerLine
214 recentCursor protocol.PickerCursor
215 // current is the attachment the picker is presented over; it only
216 // decides where a fresh cursor starts.
217 current pickerCurrent
218}
219
220// newPickerCatalogue builds an empty projection.
221func newPickerCatalogue(cfg pickerCatalogueConfig) *pickerCatalogue {
222 clock := cfg.Clock
223 if supervisorNil(clock) {
224 clock = systemClock{}
225 }
226 freshness := cfg.Freshness
227 if freshness <= 0 {
228 freshness = pickerCatalogueDefaultFreshness
229 }
230 return &pickerCatalogue{
231 clock: clock,
232 freshness: freshness,
233 hosts: make(map[string]*pickerHostState),
234 refs: make(map[string]pickerSelectionRef),
235 retired: make(map[string]pickerSelectionRef),
236 }
237}
238
239// Apply folds one broker publication into the projection. It reports whether
240// the projection changed. The publication is the projection's trust boundary:
241// a snapshot that fails its own contract is rejected outright and the previous
242// projection is kept. A valid publication that does not supersede the applied
243// epoch/revision is also ignored, so a slow or out-of-order snapshot can never
244// regress a row.
245func (c *pickerCatalogue) Apply(snapshot ports.BrokerSnapshot) bool {
246 if c == nil {
247 return false
248 }
249 c.mu.Lock()
250 defer c.mu.Unlock()
251 if snapshot.Epoch == 0 || snapshot.Revision == 0 {
252 return false
253 }
254 if err := snapshot.Validate(); err != nil {
255 return false
256 }
257 if !snapshot.Supersedes(c.snapshot) {
258 return false
259 }
260 if c.snapshot.Epoch != 0 && snapshot.Epoch != c.snapshot.Epoch {
261 // A new broker process is a full resynchronization: every key, row, and
262 // retired fence from the previous epoch is invalid.
263 c.resetLocked()
264 }
265 c.snapshot = snapshot.Clone()
266 c.mergeHostsLocked()
267 c.projectLocked()
268 return true
269}
270
271// Epoch reports the applied broker epoch, or zero before the first publication.
272func (c *pickerCatalogue) Epoch() ports.BrokerEpoch {
273 if c == nil {
274 return 0
275 }
276 c.mu.Lock()
277 defer c.mu.Unlock()
278 return c.snapshot.Epoch
279}
280
281// Revision reports the applied broker revision, or zero before the first
282// publication.
283func (c *pickerCatalogue) Revision() ports.BrokerRevision {
284 if c == nil {
285 return 0
286 }
287 c.mu.Lock()
288 defer c.mu.Unlock()
289 return c.snapshot.Revision
290}
291
292// Snapshot returns a defensive copy of the applied broker snapshot.
293func (c *pickerCatalogue) Snapshot() ports.BrokerSnapshot {
294 if c == nil {
295 return ports.BrokerSnapshot{}
296 }
297 c.mu.Lock()
298 defer c.mu.Unlock()
299 return c.snapshot.Clone()
300}
301
302// Lines returns a copy of the current ordered rows.
303func (c *pickerCatalogue) Lines() []protocol.PickerLine {
304 if c == nil {
305 return nil
306 }
307 c.mu.Lock()
308 defer c.mu.Unlock()
309 return append([]protocol.PickerLine(nil), c.lines...)
310}
311
312// Cursor returns the projection's cursor hint.
313func (c *pickerCatalogue) Cursor() protocol.PickerCursor {
314 if c == nil {
315 return protocol.PickerCursor{}
316 }
317 c.mu.Lock()
318 defer c.mu.Unlock()
319 return c.cursor
320}
321
322// projections returns the recent and grouped projections together, so both
323// always describe the same publication.
324func (c *pickerCatalogue) projections() (recent, grouped protocol.PickerProjection) {
325 if c == nil {
326 return protocol.PickerProjection{}, protocol.PickerProjection{}
327 }
328 c.mu.Lock()
329 defer c.mu.Unlock()
330 recent = protocol.PickerProjection{Lines: append([]protocol.PickerLine(nil), c.recent...), Cursor: c.recentCursor}
331 grouped = protocol.PickerProjection{Lines: append([]protocol.PickerLine(nil), c.lines...), Cursor: c.cursor}
332 return recent, grouped
333}
334
335// RecentLines returns a copy of the flat recency projection.
336func (c *pickerCatalogue) RecentLines() []protocol.PickerLine {
337 if c == nil {
338 return nil
339 }
340 c.mu.Lock()
341 defer c.mu.Unlock()
342 return append([]protocol.PickerLine(nil), c.recent...)
343}
344
345// SetCurrent records the attachment the picker is presented over (or none)
346// and recomputes the cursor hints.
347func (c *pickerCatalogue) SetCurrent(current pickerCurrent) {
348 if c == nil {
349 return
350 }
351 c.mu.Lock()
352 defer c.mu.Unlock()
353 if c.current == current {
354 return
355 }
356 c.current = current
357 if c.snapshot.Epoch != 0 {
358 c.projectLocked()
359 }
360}
361
362// Ref returns the exact selection identity recorded for one row key, including
363// a retired key that has left the current projection.
364func (c *pickerCatalogue) Ref(key string) (pickerSelectionRef, bool) {
365 if c == nil {
366 return pickerSelectionRef{}, false
367 }
368 c.mu.Lock()
369 defer c.mu.Unlock()
370 if ref, ok := c.refs[key]; ok {
371 return ref, true
372 }
373 ref, ok := c.retired[key]
374 return ref, ok
375}
376
377// Resolve revalidates one row key against the LATEST applied snapshot and
378// returns the exact broker stream request it names. It refuses with a typed
379// refusal when the identity is gone, replaced, stale, incompatible, or
380// unavailable, and it never invents identity.
381func (c *pickerCatalogue) Resolve(key string, base pickerResolveBase) (ports.BrokerOpenStreamRequest, error) {
382 if c == nil {
383 return ports.BrokerOpenStreamRequest{}, pickerCatalogueError{Code: pickerCatalogueUnknown, Text: "no catalogue"}
384 }
385 c.mu.Lock()
386 defer c.mu.Unlock()
387 ref, ok := c.refs[key]
388 if !ok {
389 ref, ok = c.retired[key]
390 }
391 if !ok {
392 return ports.BrokerOpenStreamRequest{}, pickerCatalogueError{Code: pickerCatalogueUnknown, Text: "picker row is not in the catalogue"}
393 }
394 return c.resolveRefLocked(ref, base)
395}
396
397// ResolveTarget is the commit resolution of one published row: the row's
398// selection kind, the exact broker stream request, and the tab it names, all
399// revalidated under one lock against the same publication. Only a session,
400// tab, or create row is a destination; any other kind is refused.
401func (c *pickerCatalogue) ResolveTarget(key string, base pickerResolveBase) (ports.BrokerOpenStreamRequest, attachmentTab, error) {
402 if c == nil {
403 return ports.BrokerOpenStreamRequest{}, attachmentTab{}, pickerCatalogueError{Code: pickerCatalogueUnknown, Text: "no catalogue"}
404 }
405 c.mu.Lock()
406 defer c.mu.Unlock()
407 ref, ok := c.refs[key]
408 if !ok {
409 ref, ok = c.retired[key]
410 }
411 if !ok {
412 return ports.BrokerOpenStreamRequest{}, attachmentTab{}, pickerCatalogueError{Code: pickerCatalogueUnknown, Text: "picker row is not in the catalogue"}
413 }
414 switch ref.kind {
415 case pickerSelectionExact, pickerSelectionCreateNamed, pickerSelectionCreateEphemeral:
416 default:
417 return ports.BrokerOpenStreamRequest{}, attachmentTab{}, pickerCatalogueError{Code: pickerCatalogueUnavailable, Text: "this picker row is not a session destination"}
418 }
419 return resolvePickerTarget(c.snapshot.Epoch, c.authorityForRefLocked(ref), ref, base, true)
420}
421
422// ResolveRef revalidates an exact selection identity a driver captured when it
423// displayed the row. It is the entry point for a selection held across a
424// broker epoch or an equivalent output boundary, where the key alone would be
425// re-bound to a newer identity.
426func (c *pickerCatalogue) ResolveRef(ref pickerSelectionRef, base pickerResolveBase) (ports.BrokerOpenStreamRequest, error) {
427 if c == nil {
428 return ports.BrokerOpenStreamRequest{}, pickerCatalogueError{Code: pickerCatalogueUnknown, Text: "no catalogue"}
429 }
430 c.mu.Lock()
431 defer c.mu.Unlock()
432 return c.resolveRefLocked(ref, base)
433}
434
435func (c *pickerCatalogue) resetLocked() {
436 c.hosts = make(map[string]*pickerHostState)
437 c.order = nil
438 c.refs = make(map[string]pickerSelectionRef)
439 c.retired = make(map[string]pickerSelectionRef)
440 c.retiredOrder = nil
441 c.lines = nil
442 c.cursor = protocol.PickerCursor{}
443 c.recent = nil
444 c.recentCursor = protocol.PickerCursor{}
445}
446
447// mergeHostsLocked folds the newest publication into the per-daemon state. A
448// daemon keeps its observed state when the incoming observation regresses, and
449// an incarnation change replaces the daemon's identity wholesale.
450func (c *pickerCatalogue) mergeHostsLocked() {
451 next := make(map[string]*pickerHostState, len(c.snapshot.Daemons))
452 for i := range c.snapshot.Daemons {
453 observation := c.snapshot.Daemons[i]
454 key := pickerDaemonKey(observation.Local, observation.Endpoint)
455 current, ok := c.hosts[key]
456 if ok && pickerSameIncarnation(current.observation, observation) && !pickerObservationNewer(observation, current.observation) {
457 // The publication is newer but this daemon's own observation is not:
458 // keep the newer observed state and still apply configured authority.
459 next[key] = &pickerHostState{
460 observation: pickerMergeAuthority(current.observation, observation),
461 }
462 continue
463 }
464 next[key] = &pickerHostState{observation: observation}
465 }
466 c.hosts = next
467 c.order = pickerStableHostOrder(c.order, c.snapshot.Daemons)
468 // Tombstones fence a retired registration even if a publication still
469 // carries it; a re-added endpoint with a fresh incarnation is never fenced.
470 for _, tombstone := range c.snapshot.Removed {
471 for key, host := range c.hosts {
472 if !host.observation.Local && host.observation.Endpoint == tombstone.Endpoint && tombstone.Fences(host.observation.Registration) {
473 delete(c.hosts, key)
474 }
475 }
476 }
477}
478
479// moveRetiredLocked keeps the last known ref of a key that left the current
480// projection so a late selection can still be classified as gone or replaced.
481func (c *pickerCatalogue) moveRetiredLocked(next map[string]pickerSelectionRef) {
482 for key, ref := range c.refs {
483 if _, ok := next[key]; ok {
484 continue
485 }
486 if _, exists := c.retired[key]; !exists {
487 c.retiredOrder = append(c.retiredOrder, key)
488 }
489 c.retired[key] = ref
490 }
491 for len(c.retiredOrder) > pickerCatalogueMaxRetiredRefs {
492 oldest := c.retiredOrder[0]
493 c.retiredOrder = c.retiredOrder[1:]
494 delete(c.retired, oldest)
495 }
496}
497
498// resolveRefLocked revalidates exactly one captured selection against the
499// latest applied publication.
500func (c *pickerCatalogue) resolveRefLocked(ref pickerSelectionRef, base pickerResolveBase) (ports.BrokerOpenStreamRequest, error) {
501 request, _, err := resolvePickerTarget(c.snapshot.Epoch, c.authorityForRefLocked(ref), ref, base, true)
502 return request, err
503}
504
505// authorityForRefLocked selects the merged observation one selection names. The
506// merged host state is used rather than the raw publication, so a slow snapshot
507// can never regress the authority a selection resolves against.
508func (c *pickerCatalogue) authorityForRefLocked(ref pickerSelectionRef) pickerResolveAuthority {
509 for _, key := range c.order {
510 host, ok := c.hosts[key]
511 if !ok {
512 continue
513 }
514 if pickerAuthorityMatches(host.observation, ref) {
515 return pickerResolveAuthority{observation: host.observation, found: true}
516 }
517 }
518 return pickerResolveAuthority{}
519}
520
521// resolveCatalogueRequest resolves one selection against a raw broker
522// publication. It is the entry point of the closed InitialNavigation union,
523// which carries its own destination fence rather than a projected row key.
524func resolveCatalogueRequest(snapshot ports.BrokerSnapshot, ref pickerSelectionRef, base pickerResolveBase) (ports.BrokerOpenStreamRequest, error) {
525 return resolvePickerRequest(snapshot.Epoch, pickerAuthorityInSnapshot(snapshot, ref), ref, base)
526}
527
528// pickerResolveAuthority is the single daemon authority one selection names in
529// one publication, merged or raw.
530type pickerResolveAuthority struct {
531 observation ports.BrokerDaemonObservation
532 found bool
533}
534
535// pickerAuthorityMatches reports whether one observation is the authority a
536// selection names: the local daemon for a local selection, or the exact remote
537// endpoint otherwise.
538func pickerAuthorityMatches(observation ports.BrokerDaemonObservation, ref pickerSelectionRef) bool {
539 if ref.local {
540 return observation.Local
541 }
542 return !observation.Local && observation.Endpoint == ref.endpoint
543}
544
545// pickerAuthorityInSnapshot selects the observation one selection names from a
546// raw broker publication.
547func pickerAuthorityInSnapshot(snapshot ports.BrokerSnapshot, ref pickerSelectionRef) pickerResolveAuthority {
548 for i := range snapshot.Daemons {
549 if pickerAuthorityMatches(snapshot.Daemons[i], ref) {
550 return pickerResolveAuthority{observation: snapshot.Daemons[i], found: true}
551 }
552 }
553 return pickerResolveAuthority{}
554}
555
556// resolvePickerRequest turns one validated selection into the exact broker
557// stream request it names. Connection and Stream come from the caller's own
558// live connection; Policy comes exclusively from the selected observation and
559// the process environment is never read.
560//
561// Refusal rules are deliberately narrow:
562//
563// - the applied publication must be valid (nonzero epoch);
564// - a captured epoch must equal the current one, never compare numerically
565// across broker processes;
566// - the selected daemon must still be in the publication, and a remote
567// selection must still match its complete registration (endpoint,
568// incarnation, and generation), not merely its incarnation, so a
569// re-registered endpoint or an advanced generation is a replacement;
570// - a known incompatibility refuses immediately, because no bounded attempt
571// can succeed against a daemon whose version or availability says so.
572//
573// Freshness is presentation information, not a lock: a stale, unavailable, or
574// not-yet-observed daemon may still be attempted explicitly, because only the
575// daemon revalidates the exact identity and only it may start a stopped target
576// under the resolved policy. The request therefore always carries the explicit
577// start-if-needed authorization; a policy that forbids launching still refuses
578// visibly at the destination.
579func resolvePickerRequest(epoch ports.BrokerEpoch, authority pickerResolveAuthority, ref pickerSelectionRef, base pickerResolveBase) (ports.BrokerOpenStreamRequest, error) {
580 request, _, err := resolvePickerTarget(epoch, authority, ref, base, false)
581 return request, err
582}
583
584// resolvePickerTarget is resolvePickerRequest plus the exact tab a tab row
585// names. refuseFailing is the picker's instant refusal: a
586// remote host the broker observed failing (unreachable, authentication, or an
587// invalid response) refuses with a notice instead of a slow attempt that
588// would fail the same way. An explicit CLI target keeps the attempt.
589func resolvePickerTarget(epoch ports.BrokerEpoch, authority pickerResolveAuthority, ref pickerSelectionRef, base pickerResolveBase, refuseFailing bool) (ports.BrokerOpenStreamRequest, attachmentTab, error) {
590 fail := func(err error) (ports.BrokerOpenStreamRequest, attachmentTab, error) {
591 return ports.BrokerOpenStreamRequest{}, attachmentTab{}, err
592 }
593 if epoch == 0 {
594 return fail(pickerCatalogueError{Code: pickerCatalogueUnavailable, Text: "broker catalogue is unavailable"})
595 }
596 if ref.epoch != 0 && ref.epoch != epoch {
597 return fail(pickerCatalogueError{Code: pickerCatalogueEpochStale, Text: "broker epoch changed"})
598 }
599 if !authority.found {
600 return fail(pickerCatalogueError{Code: pickerCatalogueGone, Text: "host is no longer in the catalogue"})
601 }
602 observation := authority.observation
603 if !ref.local && !observation.Registration.Equal(ref.registration) {
604 return fail(pickerCatalogueError{Code: pickerCatalogueReplaced, Text: "host identity was replaced"})
605 }
606 if observation.Availability == domain.RemoteAvailabilityIncompatible || pickerObservationVersionMismatch(observation) {
607 return fail(pickerCatalogueError{Code: pickerCatalogueIncompatible, Text: "host protocol is incompatible"})
608 }
609 if observation.Availability == domain.RemoteAvailabilityNoDaemon && ref.kind == pickerSelectionExact {
610 return fail(pickerCatalogueError{Code: pickerCatalogueUnavailable, Text: "remote host has no vev daemon; choose create session"})
611 }
612 if observation.Availability == domain.RemoteAvailabilityNoDaemon && ref.kind != pickerSelectionCreateNamed && ref.kind != pickerSelectionCreateEphemeral && ref.kind != pickerSelectionAttachNamed {
613 // A named attach carries StartIfNeeded and lets the started daemon
614 // restore or refuse the name, so a no-daemon observation is no refusal.
615 return fail(pickerCatalogueError{Code: pickerCatalogueUnavailable, Text: "no daemon exists on this host; choose create session"})
616 }
617 if refuseFailing && !ref.local && pickerObservationFailing(observation) && observation.Availability != domain.RemoteAvailabilityNoDaemon {
618 reason := pickerObservationReason(observation, false)
619 return fail(pickerCatalogueError{
620 Code: pickerCatalogueUnavailable,
621 Text: "host " + reason,
622 Notice: pickerUnavailableNotice(ref, observation, reason),
623 })
624 }
625 request := ports.BrokerOpenStreamRequest{
626 Epoch: epoch,
627 Purpose: ports.BrokerStreamAttachment,
628 Local: ref.local,
629 Connection: base.Connection,
630 Stream: base.Stream,
631 Registration: observation.Registration,
632 Policy: observation.Policy,
633 Env: append([]string(nil), base.Env...),
634 // An attachment or creation may need to start a stopped daemon, so it
635 // carries the explicit start-if-needed authorization; the resolved
636 // policy still decides whether launching is permitted.
637 StartMode: ports.BrokerDaemonStartIfNeeded,
638 }
639 if !ref.local {
640 request.Endpoint = observation.Endpoint
641 }
642 var tab attachmentTab
643 switch ref.kind {
644 case pickerSelectionExact:
645 session, ok := pickerFindSession(observation.Sessions, ref.lifecycle)
646 if !ok {
647 if pickerSessionReplaced(observation.Sessions, ref) {
648 return fail(pickerCatalogueError{Code: pickerCatalogueReplaced, Text: "session identity was replaced"})
649 }
650 return fail(pickerCatalogueError{Code: pickerCatalogueGone, Text: "session is no longer in the catalogue"})
651 }
652 if session.Name != ref.name {
653 // The exact identity is the lifecycle/name pair the daemon
654 // revalidates: a renamed lifecycle is a different selection.
655 return fail(pickerCatalogueError{Code: pickerCatalogueReplaced, Text: "session identity was replaced"})
656 }
657 if session.State == catalogue.RemoteCatalogSessionBroken {
658 return fail(pickerCatalogueError{Code: pickerCatalogueUnavailable, Text: "session is broken"})
659 }
660 resolved, err := pickerResolveTab(ref, session)
661 if err != nil {
662 return fail(err)
663 }
664 tab = resolved
665 request.Admission = ports.BrokerAdmissionExact
666 request.Target = protocol.ExactSessionTarget{LifecycleID: session.LifecycleID, SessionName: session.Name}
667 case pickerSelectionCreateNamed:
668 if err := domain.ValidateSessionName(ref.createName); err != nil {
669 return fail(pickerCatalogueError{Code: pickerCatalogueInvalidName, Text: "session name is not valid"})
670 }
671 request.Admission = ports.BrokerAdmissionCreateNamed
672 request.Name = ref.createName
673 case pickerSelectionAttachNamed:
674 if err := domain.ValidateSessionName(ref.name); err != nil {
675 return fail(pickerCatalogueError{Code: pickerCatalogueInvalidName, Text: "session name is not valid"})
676 }
677 request.Admission = ports.BrokerAdmissionAttachNamed
678 request.Name = ref.name
679 case pickerSelectionCreateEphemeral:
680 request.Admission = ports.BrokerAdmissionCreateEphemeral
681 default:
682 return fail(pickerCatalogueError{Code: pickerCatalogueUnknown, Text: "picker selection kind is unknown"})
683 }
684 if err := request.Validate(); err != nil {
685 return fail(pickerCatalogueError{Code: pickerCatalogueUnknown, Text: "picker stream request is not valid"})
686 }
687 return request, tab, nil
688}
689
690// freshLocked reports whether LastSuccess is inside the freshness window. An
691// observation that never succeeded is never fresh.
692func (c *pickerCatalogue) freshLocked(observation ports.BrokerDaemonObservation, now time.Time) bool {
693 if observation.LastSuccess.IsZero() {
694 return false
695 }
696 return now.Sub(observation.LastSuccess) <= c.freshness
697}
698
699func pickerDaemonKey(local bool, endpoint string) string {
700 if local {
701 return "local"
702 }
703 return "remote:" + endpoint
704}
705
706// pickerCatalogueOriginFallback is the fixed, bounded label shown when an
707// observation carries no display origin. A raw endpoint is never a display
708// label.
709const pickerCatalogueOriginFallback = "remote"
710
711func pickerOriginLabel(observation ports.BrokerDaemonObservation) string {
712 if observation.DisplayOrigin != "" {
713 return observation.DisplayOrigin
714 }
715 if observation.Local {
716 return "local"
717 }
718 return pickerCatalogueOriginFallback
719}
720
721// pickerHostRowKey is the stable, opaque key of a daemon's status row. It is
722// never a selection: a host row is focusable but not committable.
723func pickerHostRowKey(observation ports.BrokerDaemonObservation) string {
724 return pickerRowKey("h", observation.Local, observation.Endpoint, observation.Registration.Incarnation[:], observation.Registration.Generation, nil)
725}
726
727// pickerSessionRowKey is the stable, opaque key of one exact session. It is
728// derived from the complete identity (endpoint, registration incarnation and
729// generation, session lifecycle): a rename never changes it, a replaced
730// lifecycle always does, and an advanced registration generation does too. The
731// generation is part of the key precisely so that changed authority requires a
732// fresh explicit selection instead of silently re-binding a displayed row.
733func pickerSessionRowKey(ref pickerSelectionRef) string {
734 return pickerRowKey("s", ref.local, ref.endpoint, ref.registration.Incarnation[:], ref.registration.Generation, ref.lifecycle[:])
735}
736
737func pickerRowKey(prefix string, local bool, endpoint string, incarnation []byte, generation domain.RemoteGeneration, lifecycle []byte) string {
738 digest := sha256.New()
739 digest.Write([]byte("vev.picker."))
740 digest.Write([]byte(prefix))
741 digest.Write([]byte{0})
742 if local {
743 digest.Write([]byte("true"))
744 } else {
745 digest.Write([]byte("false"))
746 }
747 digest.Write([]byte{0})
748 digest.Write([]byte(endpoint))
749 digest.Write([]byte{0})
750 digest.Write(incarnation)
751 digest.Write([]byte{0})
752 var encoded [8]byte
753 binary.BigEndian.PutUint64(encoded[:], uint64(generation))
754 digest.Write(encoded[:])
755 digest.Write([]byte{0})
756 digest.Write(lifecycle)
757 return prefix + base64.RawURLEncoding.EncodeToString(digest.Sum(nil)[:16])
758}
759
760// pickerStableHostOrder keeps each host's first-seen position and appends new
761// hosts in publication order, so an unrelated update never reshuffles rows.
762func pickerStableHostOrder(previous []string, daemons []ports.BrokerDaemonObservation) []string {
763 present := make(map[string]struct{}, len(daemons))
764 published := make([]string, 0, len(daemons))
765 for i := range daemons {
766 key := pickerDaemonKey(daemons[i].Local, daemons[i].Endpoint)
767 if _, ok := present[key]; ok {
768 continue
769 }
770 present[key] = struct{}{}
771 published = append(published, key)
772 }
773 order := make([]string, 0, len(published))
774 seen := make(map[string]struct{}, len(published))
775 for _, key := range previous {
776 if _, ok := present[key]; !ok {
777 continue
778 }
779 if _, ok := seen[key]; ok {
780 continue
781 }
782 seen[key] = struct{}{}
783 order = append(order, key)
784 }
785 for _, key := range published {
786 if _, ok := seen[key]; ok {
787 continue
788 }
789 seen[key] = struct{}{}
790 order = append(order, key)
791 }
792 return order
793}
794
795// pickerObservationCompatible reports whether the observed daemon can accept an
796// attachment under its own configured policy. A version mismatch is derived by
797// comparing the observation's ProtocolVersion against its Policy.ProtocolVersion,
798// and only when the daemon was actually observed (ProtocolVersion != 0). An
799// unobserved daemon carries no version and is never compatible; reachability is
800// classified separately so an unobserved or unreachable daemon is never
801// reported as a version mismatch.
802func pickerObservationCompatible(observation ports.BrokerDaemonObservation) bool {
803 return observation.ProtocolVersion != 0 && observation.ProtocolVersion == observation.Policy.ProtocolVersion
804}
805
806// pickerObservationKnownIncompatible reports whether a daemon is known to be
807// unable to accept an attachment: either its availability was classified as
808// incompatible by the observation itself, or it was observed with a protocol
809// version that differs from its policy's required version. An unobserved or
810// unreachable daemon is not known-incompatible, so an explicit bounded attempt
811// stays possible; that distinction is what lets the picker start a stopped
812// daemon instead of blocking behind observation.
813func pickerObservationKnownIncompatible(observation ports.BrokerDaemonObservation) bool {
814 return observation.Availability == domain.RemoteAvailabilityIncompatible || pickerObservationVersionMismatch(observation)
815}
816
817// pickerObservationVersionMismatch reports whether a daemon that was actually
818// observed (ProtocolVersion != 0) reports a protocol version that differs from
819// its policy's required version. Reachability and observation are classified
820// separately, so an unobserved daemon is never a version mismatch.
821func pickerObservationVersionMismatch(observation ports.BrokerDaemonObservation) bool {
822 return observation.ProtocolVersion != 0 && observation.ProtocolVersion != observation.Policy.ProtocolVersion
823}
824
825// pickerSameIncarnation reports whether two observations describe the same
826// daemon identity, so their observed state may be compared for monotonicity.
827func pickerSameIncarnation(current, incoming ports.BrokerDaemonObservation) bool {
828 return current.Incarnation == incoming.Incarnation
829}
830
831// pickerObservationNewer reports whether an incoming observation is at least as
832// new as the current one for the same daemon identity. A different incarnation
833// is an identity replacement and always wins.
834func pickerObservationNewer(incoming, current ports.BrokerDaemonObservation) bool {
835 if incoming.Incarnation != current.Incarnation {
836 return true
837 }
838 if incoming.FailureEpisode != current.FailureEpisode {
839 return incoming.FailureEpisode > current.FailureEpisode
840 }
841 return !incoming.LastSuccess.Before(current.LastSuccess)
842}
843
844// pickerMergeAuthority keeps the newer observed state while refreshing every
845// field the registry stamps as configured authority.
846func pickerMergeAuthority(current, incoming ports.BrokerDaemonObservation) ports.BrokerDaemonObservation {
847 merged := current
848 merged.Local = incoming.Local
849 merged.Endpoint = incoming.Endpoint
850 merged.DisplayOrigin = incoming.DisplayOrigin
851 merged.Rank = incoming.Rank
852 merged.Registration = incoming.Registration
853 merged.Policy = incoming.Policy
854 return merged
855}
856
857// pickerObservationStatus classifies one observation into a row badge using
858// the shared domain.RemoteReason taxonomy: availability is decided first, so
859// an unobserved (Availability Unknown) or unreachable daemon is never reported
860// as a version mismatch. A version mismatch is derived only for a reachable
861// observation whose observed ProtocolVersion differs from policy.
862func pickerObservationStatus(observation ports.BrokerDaemonObservation, fresh bool) protocol.PickerLineStatus {
863 switch observation.Availability {
864 case domain.RemoteAvailabilityUnreachable, domain.RemoteAvailabilityAuthFailed:
865 return protocol.PickerLineStatusDown
866 case domain.RemoteAvailabilityInvalidResponse:
867 return protocol.PickerLineStatusError
868 case domain.RemoteAvailabilityIncompatible:
869 return protocol.PickerLineStatusVersion
870 case domain.RemoteAvailabilityNoDaemon:
871 return protocol.PickerLineStatusNoDaemon
872 case domain.RemoteAvailabilityReachable:
873 if pickerObservationVersionMismatch(observation) {
874 return protocol.PickerLineStatusVersion
875 }
876 if observation.ProtocolVersion == 0 || !fresh {
877 return protocol.PickerLineStatusStale
878 }
879 // A healthy host carries no dot.
880 return protocol.PickerLineStatusNone
881 default:
882 // Availability Unknown (unobserved) and any out-of-range value are
883 // never treated as compatible.
884 return protocol.PickerLineStatusStale
885 }
886}
887
888// pickerObservationReason classifies one observation into a bounded display
889// reason from the shared domain.RemoteReason taxonomy. Availability is decided
890// first; the version branch applies only to a reachable observation whose
891// observed version differs from policy. Unobserved and unknown availability
892// render as refreshing.
893func pickerObservationReason(observation ports.BrokerDaemonObservation, fresh bool) string {
894 switch observation.Availability {
895 case domain.RemoteAvailabilityUnreachable:
896 return domain.RemoteReasonHostUnreachable
897 case domain.RemoteAvailabilityAuthFailed:
898 return domain.RemoteReasonAuthFailure
899 case domain.RemoteAvailabilityInvalidResponse:
900 return domain.RemoteReasonMalformed
901 case domain.RemoteAvailabilityIncompatible:
902 return domain.RemoteReasonVersionMismatch
903 case domain.RemoteAvailabilityNoDaemon:
904 return domain.RemoteReasonNoDaemon
905 case domain.RemoteAvailabilityReachable:
906 if pickerObservationVersionMismatch(observation) {
907 return domain.RemoteReasonVersionMismatch
908 }
909 if observation.ProtocolVersion == 0 {
910 return domain.RemoteReasonRefreshing
911 }
912 if observation.Checking {
913 return domain.RemoteReasonRefreshing
914 }
915 if !fresh {
916 return domain.RemoteReasonCatalogStale
917 }
918 return ""
919 default:
920 // Availability Unknown is the unobserved state: it is still being
921 // discovered, so it renders as refreshing rather than as a failure.
922 return domain.RemoteReasonRefreshing
923 }
924}
925
926func pickerHostDetail(observation ports.BrokerDaemonObservation) string {
927 if !observation.InventoryKnown {
928 return "…"
929 }
930 // The problem dot and its toast already say why no session is listed.
931 return ""
932}
933
934// pickerSessionStatus marks only a broken session. A live session is the
935// normal case and a stopped one is already muted, so neither carries a dot.
936func pickerSessionStatus(session catalogue.RemoteCatalogSession) protocol.PickerLineStatus {
937 if session.State == catalogue.RemoteCatalogSessionBroken {
938 return protocol.PickerLineStatusError
939 }
940 return protocol.PickerLineStatusNone
941}
942
943func pickerSessionDetail(session catalogue.RemoteCatalogSession) string {
944 count := ""
945 if n := len(session.Tabs); n > 0 {
946 count = strconv.Itoa(n)
947 }
948 if session.Attached {
949 // "*" marks the attached session; ● is reserved for the problem dot.
950 return strings.TrimSpace("* " + count)
951 }
952 return count
953}
954
955// pickerCatalogueCursor hints the first actionable row, then the first
956// inspectable row, so a fresh picker always has a cursor destination when one
957// exists.
958func pickerCatalogueCursor(lines []protocol.PickerLine) protocol.PickerCursor {
959 for i, line := range lines {
960 if line.Key != "" && line.Focusable && line.Actions != 0 {
961 return protocol.PickerCursor{Key: line.Key, Index: i}
962 }
963 }
964 for i, line := range lines {
965 if line.Key != "" && line.Focusable {
966 return protocol.PickerCursor{Key: line.Key, Index: i}
967 }
968 }
969 return protocol.PickerCursor{Index: -1}
970}
971
972func pickerFindSession(sessions []catalogue.RemoteCatalogSession, lifecycle domain.SessionLifecycleID) (catalogue.RemoteCatalogSession, bool) {
973 for _, session := range sessions {
974 if session.LifecycleID == lifecycle {
975 return session, true
976 }
977 }
978 return catalogue.RemoteCatalogSession{}, false
979}
980
981// pickerSessionReplaced reports whether the exact lifecycle vanished but a
982// same-name session remains, which is a replacement rather than a removal.
983func pickerSessionReplaced(sessions []catalogue.RemoteCatalogSession, ref pickerSelectionRef) bool {
984 if ref.name == "" {
985 return false
986 }
987 for _, session := range sessions {
988 if session.LifecycleID != ref.lifecycle && session.Name == ref.name {
989 return true
990 }
991 }
992 return false
993}