detached 4895225

feat(client): publish the route ledger and settle daemon navigation

A pure client route ledger builds protocol.RecentRouteSnapshot from the
broker catalogue (every daemon, local and remote): attention is
any(tab.Attention), entries follow this client's attach recency, then
live local by LastUsedSeq, live remote in catalogue order, then stopped;
the attached lifecycle is ActiveEntry only, references stay stable per
lifecycle, and the other daemons are listed as creation hosts.

While attached the supervisor consumes the broker subscription and
publishes the snapshot after the committed initial publication, on every
attached-session change, and on every broker change. The daemon's
RouteNavigationAction, RouteCreateSessionAction and endpoint-empty
AttachTarget are settled like a picker commit (in-place tab switch or a
clean detach plus Connecting swap); refusals are answered with the typed
failures. Same-peer offers are confirmed in place.
commit
4895225c49aad735f6ef4cdc785d07b2d2562b38
author
brice · · 2026-09-22 20:39 UTC
parent
588ba1c
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
10 files · +1326 -10
A internal/usecase/client/attachment_routes.go +133 -0
@@ -0,0 +1,133 @@
1+package client
2+
3+import "github.com/bnema/vev/internal/protocol"
4+
5+// Route publication and daemon navigation seams of the attached foreground
6+// (Plan 003 C4/C5). The worker owns the stream, so the supervisor reaches the
7+// daemon only through these bounded channels: the newest route snapshot
8+// replaces an unsent one, and navigation failures are dropped rather than
9+// blocking when the worker is gone. In the other direction the worker hands
10+// every daemon navigation request to the supervisor, which resolves it
11+// through the broker catalogue exactly like a picker commit.
12+
13+// attachmentReplyCapacity bounds queued navigation failures per foreground.
14+const attachmentReplyCapacity = 4
15+
16+// demandRoutes wakes the supervisor to republish routes for this foreground.
17+func (f *attachmentForeground) demandRoutes() {
18+ if f == nil || f.host == nil || !f.authority.actionAuthorized(f) {
19+ return
20+ }
21+ select {
22+ case f.host.routeDemand <- struct{}{}:
23+ default:
24+ }
25+}
26+
27+func (f *attachmentForeground) routeSnapshots() <-chan protocol.RecentRouteSnapshot {
28+ if f == nil {
29+ return nil
30+ }
31+ return f.routes
32+}
33+
34+func (f *attachmentForeground) navigationReplies() <-chan protocol.ClientMessage {
35+ if f == nil {
36+ return nil
37+ }
38+ return f.replies
39+}
40+
41+// requestNavigation hands one daemon navigation request to the supervisor.
42+// The newest request replaces one the supervisor has not taken yet.
43+func (f *attachmentForeground) requestNavigation(message protocol.ServerMessage) {
44+ if f == nil || f.host == nil || !f.authority.actionAuthorized(f) {
45+ return
46+ }
47+ for {
48+ select {
49+ case f.host.navigations <- message:
50+ return
51+ default:
52+ }
53+ select {
54+ case <-f.host.navigations:
55+ default:
56+ }
57+ }
58+}
59+
60+// RouteDemands wakes the supervisor when the attached session changed.
61+func (h *attachmentHost) RouteDemands() <-chan struct{} {
62+ if h == nil {
63+ return nil
64+ }
65+ return h.routeDemand
66+}
67+
68+// Navigations carries the daemon's navigation requests to the supervisor.
69+func (h *attachmentHost) Navigations() <-chan protocol.ServerMessage {
70+ if h == nil {
71+ return nil
72+ }
73+ return h.navigations
74+}
75+
76+// takeNavigation returns a navigation request left behind by a worker that
77+// already settled, if any.
78+func (h *attachmentHost) takeNavigation() (protocol.ServerMessage, bool) {
79+ if h == nil {
80+ return nil, false
81+ }
82+ select {
83+ case message := <-h.navigations:
84+ return message, true
85+ default:
86+ return nil, false
87+ }
88+}
89+
90+// publishRoutes hands the newest route snapshot to the current foreground's
91+// worker. It reports false when token no longer owns the foreground.
92+func (h *attachmentHost) publishRoutes(token AttachmentToken, snapshot protocol.RecentRouteSnapshot) bool {
93+ fg := h.currentForeground(token)
94+ if fg == nil {
95+ return false
96+ }
97+ for {
98+ select {
99+ case fg.routes <- snapshot:
100+ return true
101+ default:
102+ }
103+ select {
104+ case <-fg.routes:
105+ default:
106+ }
107+ }
108+}
109+
110+// replyNavigation queues one navigation failure for the current worker.
111+func (h *attachmentHost) replyNavigation(token AttachmentToken, message protocol.ClientMessage) bool {
112+ fg := h.currentForeground(token)
113+ if fg == nil {
114+ return false
115+ }
116+ select {
117+ case fg.replies <- message:
118+ return true
119+ default:
120+ return false
121+ }
122+}
123+
124+func (h *attachmentHost) currentForeground(token AttachmentToken) *attachmentForeground {
125+ if h == nil {
126+ return nil
127+ }
128+ fg := h.authority.foreground()
129+ if fg == nil || fg.token != token {
130+ return nil
131+ }
132+ return fg
133+}
@@ -0,0 +1,133 @@
1+package client
2+
3+import "github.com/bnema/vev/internal/protocol"
4+
5+// Route publication and daemon navigation seams of the attached foreground
6+// (Plan 003 C4/C5). The worker owns the stream, so the supervisor reaches the
7+// daemon only through these bounded channels: the newest route snapshot
8+// replaces an unsent one, and navigation failures are dropped rather than
9+// blocking when the worker is gone. In the other direction the worker hands
10+// every daemon navigation request to the supervisor, which resolves it
11+// through the broker catalogue exactly like a picker commit.
12+
13+// attachmentReplyCapacity bounds queued navigation failures per foreground.
14+const attachmentReplyCapacity = 4
15+
16+// demandRoutes wakes the supervisor to republish routes for this foreground.
17+func (f *attachmentForeground) demandRoutes() {
18+ if f == nil || f.host == nil || !f.authority.actionAuthorized(f) {
19+ return
20+ }
21+ select {
22+ case f.host.routeDemand <- struct{}{}:
23+ default:
24+ }
25+}
26+
27+func (f *attachmentForeground) routeSnapshots() <-chan protocol.RecentRouteSnapshot {
28+ if f == nil {
29+ return nil
30+ }
31+ return f.routes
32+}
33+
34+func (f *attachmentForeground) navigationReplies() <-chan protocol.ClientMessage {
35+ if f == nil {
36+ return nil
37+ }
38+ return f.replies
39+}
40+
41+// requestNavigation hands one daemon navigation request to the supervisor.
42+// The newest request replaces one the supervisor has not taken yet.
43+func (f *attachmentForeground) requestNavigation(message protocol.ServerMessage) {
44+ if f == nil || f.host == nil || !f.authority.actionAuthorized(f) {
45+ return
46+ }
47+ for {
48+ select {
49+ case f.host.navigations <- message:
50+ return
51+ default:
52+ }
53+ select {
54+ case <-f.host.navigations:
55+ default:
56+ }
57+ }
58+}
59+
60+// RouteDemands wakes the supervisor when the attached session changed.
61+func (h *attachmentHost) RouteDemands() <-chan struct{} {
62+ if h == nil {
63+ return nil
64+ }
65+ return h.routeDemand
66+}
67+
68+// Navigations carries the daemon's navigation requests to the supervisor.
69+func (h *attachmentHost) Navigations() <-chan protocol.ServerMessage {
70+ if h == nil {
71+ return nil
72+ }
73+ return h.navigations
74+}
75+
76+// takeNavigation returns a navigation request left behind by a worker that
77+// already settled, if any.
78+func (h *attachmentHost) takeNavigation() (protocol.ServerMessage, bool) {
79+ if h == nil {
80+ return nil, false
81+ }
82+ select {
83+ case message := <-h.navigations:
84+ return message, true
85+ default:
86+ return nil, false
87+ }
88+}
89+
90+// publishRoutes hands the newest route snapshot to the current foreground's
91+// worker. It reports false when token no longer owns the foreground.
92+func (h *attachmentHost) publishRoutes(token AttachmentToken, snapshot protocol.RecentRouteSnapshot) bool {
93+ fg := h.currentForeground(token)
94+ if fg == nil {
95+ return false
96+ }
97+ for {
98+ select {
99+ case fg.routes <- snapshot:
100+ return true
101+ default:
102+ }
103+ select {
104+ case <-fg.routes:
105+ default:
106+ }
107+ }
108+}
109+
110+// replyNavigation queues one navigation failure for the current worker.
111+func (h *attachmentHost) replyNavigation(token AttachmentToken, message protocol.ClientMessage) bool {
112+ fg := h.currentForeground(token)
113+ if fg == nil {
114+ return false
115+ }
116+ select {
117+ case fg.replies <- message:
118+ return true
119+ default:
120+ return false
121+ }
122+}
123+
124+func (h *attachmentHost) currentForeground(token AttachmentToken) *attachmentForeground {
125+ if h == nil {
126+ return nil
127+ }
128+ fg := h.authority.foreground()
129+ if fg == nil || fg.token != token {
130+ return nil
131+ }
132+ return fg
133+}
M internal/usecase/client/attachment_session.go +38 -0
@@ -291,10 +291,15 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
291291 overlay := attachmentOverlay(fg)
292292 var repaint <-chan struct{}
293293 var tabSelections <-chan domain.TabStableID
294+ var routeUpdates <-chan protocol.RecentRouteSnapshot
295+ var navigationReplies <-chan protocol.ClientMessage
294296 if overlay != nil {
295297 repaint = overlay.overlayRepaint()
296298 tabSelections = overlay.tabSelections()
299+ routeUpdates = overlay.routeSnapshots()
300+ navigationReplies = overlay.navigationReplies()
297301 }
302+ var samePeerRequests uint64
298303 picker := &attachmentMovePicker{worker: w, fg: fg, overlay: overlay, stream: stream, size: w.cfg.Geometry.Size, move: newMovePickerOverlay(w.cfg.TrueColor)}
299304 defer picker.stopEscape()
300305 input := newAttachmentInput(w, fg, stream, picker)
@@ -334,6 +339,16 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
334339 if err := w.send(ctx, fg, stream, protocol.SelectTab{TabID: tab}); err != nil {
335340 return w.settle(ctx, fg, stream, token, err)
336341 }
342+ case snapshot := <-routeUpdates:
343+ // The supervisor's route ledger changed: the serving daemon renders
344+ // the status-bar MRU, bells, and palette routes from it.
345+ if err := w.send(ctx, fg, stream, snapshot); err != nil {
346+ return w.settle(ctx, fg, stream, token, err)
347+ }
348+ case reply := <-navigationReplies:
349+ if err := w.send(ctx, fg, stream, reply); err != nil {
350+ return w.settle(ctx, fg, stream, token, err)
351+ }
337352 case <-picker.escape():
338353 picker.escapeFired()
339354 op, changed := picker.move.flush()
@@ -387,6 +402,29 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
387402 }
388403 case protocol.PickerClosed:
389404 picker.closed(typed.InteractionID)
405+ case protocol.AttachTarget:
406+ if typed.SamePeer && typed.ExactTarget != nil {
407+ // An endpoint-empty offer on this very connection: confirm
408+ // it and the daemon moves this attachment in place.
409+ samePeerRequests++
410+ request := protocol.SamePeerSwitchRequest{RequestID: samePeerRequests, Target: *typed.ExactTarget, PreferredTabID: typed.PreferredTabID}
411+ if request.Validate() != nil {
412+ continue
413+ }
414+ if err := w.send(ctx, fg, stream, request); err != nil {
415+ return w.settle(ctx, fg, stream, token, err)
416+ }
417+ continue
418+ }
419+ if overlay != nil {
420+ overlay.requestNavigation(typed)
421+ }
422+ case protocol.RouteNavigationAction, protocol.RouteCreateSessionAction:
423+ // The daemon asks the client to navigate: the supervisor
424+ // resolves it through the broker catalogue.
425+ if overlay != nil {
426+ overlay.requestNavigation(typed)
427+ }
390428 case protocol.ErrorMsg:
391429 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: &ProtocolError{Code: typed.Code, Text: typed.Text}}
392430 case protocol.Detached:
@@ -291,10 +291,15 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
291 overlay := attachmentOverlay(fg)291 overlay := attachmentOverlay(fg)
292 var repaint <-chan struct{}292 var repaint <-chan struct{}
293 var tabSelections <-chan domain.TabStableID293 var tabSelections <-chan domain.TabStableID
294+ var routeUpdates <-chan protocol.RecentRouteSnapshot
295+ var navigationReplies <-chan protocol.ClientMessage
294 if overlay != nil {296 if overlay != nil {
295 repaint = overlay.overlayRepaint()297 repaint = overlay.overlayRepaint()
296 tabSelections = overlay.tabSelections()298 tabSelections = overlay.tabSelections()
299+ routeUpdates = overlay.routeSnapshots()
300+ navigationReplies = overlay.navigationReplies()
297 }301 }
302+ var samePeerRequests uint64
298 picker := &attachmentMovePicker{worker: w, fg: fg, overlay: overlay, stream: stream, size: w.cfg.Geometry.Size, move: newMovePickerOverlay(w.cfg.TrueColor)}303 picker := &attachmentMovePicker{worker: w, fg: fg, overlay: overlay, stream: stream, size: w.cfg.Geometry.Size, move: newMovePickerOverlay(w.cfg.TrueColor)}
299 defer picker.stopEscape()304 defer picker.stopEscape()
300 input := newAttachmentInput(w, fg, stream, picker)305 input := newAttachmentInput(w, fg, stream, picker)
@@ -334,6 +339,16 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
334 if err := w.send(ctx, fg, stream, protocol.SelectTab{TabID: tab}); err != nil {339 if err := w.send(ctx, fg, stream, protocol.SelectTab{TabID: tab}); err != nil {
335 return w.settle(ctx, fg, stream, token, err)340 return w.settle(ctx, fg, stream, token, err)
336 }341 }
342+ case snapshot := <-routeUpdates:
343+ // The supervisor's route ledger changed: the serving daemon renders
344+ // the status-bar MRU, bells, and palette routes from it.
345+ if err := w.send(ctx, fg, stream, snapshot); err != nil {
346+ return w.settle(ctx, fg, stream, token, err)
347+ }
348+ case reply := <-navigationReplies:
349+ if err := w.send(ctx, fg, stream, reply); err != nil {
350+ return w.settle(ctx, fg, stream, token, err)
351+ }
337 case <-picker.escape():352 case <-picker.escape():
338 picker.escapeFired()353 picker.escapeFired()
339 op, changed := picker.move.flush()354 op, changed := picker.move.flush()
@@ -387,6 +402,29 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
387 }402 }
388 case protocol.PickerClosed:403 case protocol.PickerClosed:
389 picker.closed(typed.InteractionID)404 picker.closed(typed.InteractionID)
405+ case protocol.AttachTarget:
406+ if typed.SamePeer && typed.ExactTarget != nil {
407+ // An endpoint-empty offer on this very connection: confirm
408+ // it and the daemon moves this attachment in place.
409+ samePeerRequests++
410+ request := protocol.SamePeerSwitchRequest{RequestID: samePeerRequests, Target: *typed.ExactTarget, PreferredTabID: typed.PreferredTabID}
411+ if request.Validate() != nil {
412+ continue
413+ }
414+ if err := w.send(ctx, fg, stream, request); err != nil {
415+ return w.settle(ctx, fg, stream, token, err)
416+ }
417+ continue
418+ }
419+ if overlay != nil {
420+ overlay.requestNavigation(typed)
421+ }
422+ case protocol.RouteNavigationAction, protocol.RouteCreateSessionAction:
423+ // The daemon asks the client to navigate: the supervisor
424+ // resolves it through the broker catalogue.
425+ if overlay != nil {
426+ overlay.requestNavigation(typed)
427+ }
390 case protocol.ErrorMsg:428 case protocol.ErrorMsg:
391 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: &ProtocolError{Code: typed.Code, Text: typed.Text}}429 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: &ProtocolError{Code: typed.Code, Text: typed.Text}}
392 case protocol.Detached:430 case protocol.Detached:
M internal/usecase/client/attachment_worker.go +22 -0
@@ -400,6 +400,11 @@ type attachmentHost struct {
400400 // internalActions carries supervisor-originated lifecycle decisions for the
401401 // current foreground, such as the detach that precedes an attachment swap.
402402 internalActions chan AttachmentLifecycleAction
403+ // routeDemand wakes the supervisor to publish routes when the committed
404+ // session changes; navigations carries the daemon's navigation requests
405+ // (Plan 003 C4/C5). Only the supervisor consumes them.
406+ routeDemand chan struct{}
407+ navigations chan protocol.ServerMessage
403408 }
404409
405410 // newAttachmentHost builds the reusable foreground host. A nil clock falls back
@@ -439,6 +444,8 @@ func newAttachmentHost(cfg attachmentHostConfig) *attachmentHost {
439444 presentationUpdate: make(chan struct{}, 1),
440445 navigationRequests: make(chan struct{}, 1),
441446 internalActions: make(chan AttachmentLifecycleAction, 1),
447+ routeDemand: make(chan struct{}, 1),
448+ navigations: make(chan protocol.ServerMessage, 1),
442449 }
443450 }
444451
@@ -610,6 +617,8 @@ func (h *attachmentHost) newForeground(token AttachmentToken, stream ports.Broke
610617 onAttached: h.onAttach,
611618 repaint: make(chan struct{}, 1),
612619 tabSelect: make(chan domain.TabStableID, 1),
620+ routes: make(chan protocol.RecentRouteSnapshot, 1),
621+ replies: make(chan protocol.ClientMessage, attachmentReplyCapacity),
613622 }
614623 }
615624
@@ -620,6 +629,8 @@ func (h *attachmentHost) drainForegroundSignals() {
620629 select {
621630 case <-h.navigationRequests:
622631 case <-h.internalActions:
632+ case <-h.routeDemand:
633+ case <-h.navigations:
623634 default:
624635 return
625636 }
@@ -754,6 +765,9 @@ type attachmentOverlayForeground interface {
754765 overlayOutput(uiContext ports.UIContext, data []byte) error
755766 noteCommitted(target protocol.ExactSessionTarget, tab domain.TabStableID)
756767 tabSelections() <-chan domain.TabStableID
768+ routeSnapshots() <-chan protocol.RecentRouteSnapshot
769+ navigationReplies() <-chan protocol.ClientMessage
770+ requestNavigation(message protocol.ServerMessage)
757771 }
758772
759773 var _ attachmentOverlayForeground = (*attachmentForeground)(nil)
@@ -870,10 +884,14 @@ func (f *attachmentForeground) noteCommitted(target protocol.ExactSessionTarget,
870884 return
871885 }
872886 f.overlayMu.Lock()
887+ changed := !f.committedKnown || f.committed != target
873888 f.committed = target
874889 f.committedTab = tab
875890 f.committedKnown = true
876891 f.overlayMu.Unlock()
892+ if changed {
893+ f.demandRoutes()
894+ }
877895 }
878896
879897 // tabSelections carries the supervisor's in-place tab switches to the worker.
@@ -1074,6 +1092,10 @@ type attachmentForeground struct {
10741092 repaint chan struct{}
10751093 // tabSelect carries one pending in-place tab switch to the worker.
10761094 tabSelect chan domain.TabStableID
1095+ // routes carries the newest client route snapshot to the worker; replies
1096+ // carries the supervisor's navigation failures (Plan 003 C4/C5).
1097+ routes chan protocol.RecentRouteSnapshot
1098+ replies chan protocol.ClientMessage
10771099 }
10781100
10791101 // Token is the generation/attempt identity of this grant.
@@ -400,6 +400,11 @@ type attachmentHost struct {
400 // internalActions carries supervisor-originated lifecycle decisions for the400 // internalActions carries supervisor-originated lifecycle decisions for the
401 // current foreground, such as the detach that precedes an attachment swap.401 // current foreground, such as the detach that precedes an attachment swap.
402 internalActions chan AttachmentLifecycleAction402 internalActions chan AttachmentLifecycleAction
403+ // routeDemand wakes the supervisor to publish routes when the committed
404+ // session changes; navigations carries the daemon's navigation requests
405+ // (Plan 003 C4/C5). Only the supervisor consumes them.
406+ routeDemand chan struct{}
407+ navigations chan protocol.ServerMessage
403 }408 }
404 409
405 // newAttachmentHost builds the reusable foreground host. A nil clock falls back410 // newAttachmentHost builds the reusable foreground host. A nil clock falls back
@@ -439,6 +444,8 @@ func newAttachmentHost(cfg attachmentHostConfig) *attachmentHost {
439 presentationUpdate: make(chan struct{}, 1),444 presentationUpdate: make(chan struct{}, 1),
440 navigationRequests: make(chan struct{}, 1),445 navigationRequests: make(chan struct{}, 1),
441 internalActions: make(chan AttachmentLifecycleAction, 1),446 internalActions: make(chan AttachmentLifecycleAction, 1),
447+ routeDemand: make(chan struct{}, 1),
448+ navigations: make(chan protocol.ServerMessage, 1),
442 }449 }
443 }450 }
444 451
@@ -610,6 +617,8 @@ func (h *attachmentHost) newForeground(token AttachmentToken, stream ports.Broke
610 onAttached: h.onAttach,617 onAttached: h.onAttach,
611 repaint: make(chan struct{}, 1),618 repaint: make(chan struct{}, 1),
612 tabSelect: make(chan domain.TabStableID, 1),619 tabSelect: make(chan domain.TabStableID, 1),
620+ routes: make(chan protocol.RecentRouteSnapshot, 1),
621+ replies: make(chan protocol.ClientMessage, attachmentReplyCapacity),
613 }622 }
614 }623 }
615 624
@@ -620,6 +629,8 @@ func (h *attachmentHost) drainForegroundSignals() {
620 select {629 select {
621 case <-h.navigationRequests:630 case <-h.navigationRequests:
622 case <-h.internalActions:631 case <-h.internalActions:
632+ case <-h.routeDemand:
633+ case <-h.navigations:
623 default:634 default:
624 return635 return
625 }636 }
@@ -754,6 +765,9 @@ type attachmentOverlayForeground interface {
754 overlayOutput(uiContext ports.UIContext, data []byte) error765 overlayOutput(uiContext ports.UIContext, data []byte) error
755 noteCommitted(target protocol.ExactSessionTarget, tab domain.TabStableID)766 noteCommitted(target protocol.ExactSessionTarget, tab domain.TabStableID)
756 tabSelections() <-chan domain.TabStableID767 tabSelections() <-chan domain.TabStableID
768+ routeSnapshots() <-chan protocol.RecentRouteSnapshot
769+ navigationReplies() <-chan protocol.ClientMessage
770+ requestNavigation(message protocol.ServerMessage)
757 }771 }
758 772
759 var _ attachmentOverlayForeground = (*attachmentForeground)(nil)773 var _ attachmentOverlayForeground = (*attachmentForeground)(nil)
@@ -870,10 +884,14 @@ func (f *attachmentForeground) noteCommitted(target protocol.ExactSessionTarget,
870 return884 return
871 }885 }
872 f.overlayMu.Lock()886 f.overlayMu.Lock()
887+ changed := !f.committedKnown || f.committed != target
873 f.committed = target888 f.committed = target
874 f.committedTab = tab889 f.committedTab = tab
875 f.committedKnown = true890 f.committedKnown = true
876 f.overlayMu.Unlock()891 f.overlayMu.Unlock()
892+ if changed {
893+ f.demandRoutes()
894+ }
877 }895 }
878 896
879 // tabSelections carries the supervisor's in-place tab switches to the worker.897 // tabSelections carries the supervisor's in-place tab switches to the worker.
@@ -1074,6 +1092,10 @@ type attachmentForeground struct {
1074 repaint chan struct{}1092 repaint chan struct{}
1075 // tabSelect carries one pending in-place tab switch to the worker.1093 // tabSelect carries one pending in-place tab switch to the worker.
1076 tabSelect chan domain.TabStableID1094 tabSelect chan domain.TabStableID
1095+ // routes carries the newest client route snapshot to the worker; replies
1096+ // carries the supervisor's navigation failures (Plan 003 C4/C5).
1097+ routes chan protocol.RecentRouteSnapshot
1098+ replies chan protocol.ClientMessage
1077 }1099 }
1078 1100
1079 // Token is the generation/attempt identity of this grant.1101 // Token is the generation/attempt identity of this grant.
A internal/usecase/client/route_ledger.go +367 -0
@@ -0,0 +1,367 @@
1+package client
2+
3+import (
4+ "slices"
5+ "sort"
6+
7+ "github.com/bnema/vev/internal/domain"
8+ "github.com/bnema/vev/internal/ports"
9+ "github.com/bnema/vev/internal/protocol"
10+ "github.com/bnema/vev/internal/protocol/catalogue"
11+)
12+
13+// Client-owned route ledger (Plan 003 C4/E3).
14+//
15+// The serving daemon renders the status-bar MRU, the other-session bells, the
16+// jump-recent hints, and the palette's remote, recent-route, and
17+// create-on-host rows from one bounded protocol.RecentRouteSnapshot the client
18+// publishes. The ledger builds that snapshot from the broker catalogue, which
19+// already observes every daemon (local and remote), so no daemon monitors
20+// another one.
21+//
22+// It is pure: no clock, no I/O, no lock. The supervisor's run goroutine owns
23+// it and feeds it each broker publication together with the attachment's
24+// committed target.
25+//
26+// - Every session lifecycle, and every daemon as a creation host, gets a
27+// RouteRef that stays stable while it remains in the catalogue. A new
28+// lifecycle of the same name is a new reference.
29+// - Entries are ordered by this client's attach recency, then live local
30+// sessions by catalogue LastUsedSeq, then live remote sessions in
31+// catalogue order, then stopped sessions.
32+// - The attached lifecycle is metadata only (ActiveEntry), never an entry.
33+// - Attention is any(tab.Attention). AttentionSeq orders onsets as the
34+// client observed them, across daemons, so the serving daemon can jump to
35+// the oldest one.
36+// - Generation advances only when the published content changes.
37+
38+// routeAuthority names one daemon in the broker catalogue.
39+type routeAuthority struct {
40+ local bool
41+ endpoint string
42+}
43+
44+func routeAuthorityOf(observation ports.BrokerDaemonObservation) routeAuthority {
45+ if observation.Local {
46+ return routeAuthority{local: true}
47+ }
48+ return routeAuthority{endpoint: observation.Endpoint}
49+}
50+
51+// routeID names one session lifecycle on one daemon. A zero lifecycle names
52+// the daemon itself as a creation host.
53+type routeID struct {
54+ authority routeAuthority
55+ lifecycle domain.SessionLifecycleID
56+}
57+
58+// routeLedgerTarget is what one published reference resolves to.
59+type routeLedgerTarget struct {
60+ authority routeAuthority
61+ // target is zero for a creation host.
62+ target protocol.ExactSessionTarget
63+ host bool
64+}
65+
66+// routeActive is the attachment the snapshot is published over.
67+type routeActive struct {
68+ known bool
69+ authority routeAuthority
70+ target protocol.ExactSessionTarget
71+}
72+
73+type routeLedger struct {
74+ nextKey uint64
75+ keys map[routeID]uint64
76+ attachSeq uint64
77+ attached map[routeID]uint64
78+ attentionSeq uint64
79+ attention map[routeID]uint64
80+ lastActive routeID
81+ generation uint64
82+ published protocol.RecentRouteSnapshot
83+ targets map[protocol.RouteRef]routeLedgerTarget
84+}
85+
86+func newRouteLedger() *routeLedger {
87+ return &routeLedger{
88+ keys: make(map[routeID]uint64),
89+ attached: make(map[routeID]uint64),
90+ attention: make(map[routeID]uint64),
91+ targets: make(map[protocol.RouteRef]routeLedgerTarget),
92+ }
93+}
94+
95+// routeCandidate is one catalogue session before ordering.
96+type routeCandidate struct {
97+ id routeID
98+ entry protocol.RecentRouteEntry
99+ tier int
100+ lastUsed uint64
101+ daemon int
102+ session int
103+}
104+
105+const (
106+ routeTierLiveLocal = iota
107+ routeTierLiveRemote
108+ routeTierStopped
109+)
110+
111+// build folds one broker publication and the attachment's committed target
112+// into the next snapshot. It reports false when nothing visible changed, in
113+// which case the previously published snapshot is returned unchanged.
114+func (l *routeLedger) build(snapshot ports.BrokerSnapshot, active routeActive) (protocol.RecentRouteSnapshot, bool) {
115+ activeID := routeID{}
116+ if active.known {
117+ activeID = routeID{authority: active.authority, lifecycle: active.target.LifecycleID}
118+ if activeID != l.lastActive {
119+ l.attachSeq++
120+ l.attached[activeID] = l.attachSeq
121+ l.lastActive = activeID
122+ }
123+ }
124+
125+ present := make(map[routeID]struct{})
126+ candidates := make([]routeCandidate, 0)
127+ attentive := make(map[routeID]struct{})
128+ var activeObservation *ports.BrokerDaemonObservation
129+ for i := range snapshot.Daemons {
130+ daemon := snapshot.Daemons[i]
131+ authority := routeAuthorityOf(daemon)
132+ if active.known && authority == active.authority {
133+ activeObservation = &snapshot.Daemons[i]
134+ }
135+ for j, session := range daemon.Sessions {
136+ if session.State == catalogue.RemoteCatalogSessionBroken {
137+ continue
138+ }
139+ id := routeID{authority: authority, lifecycle: session.LifecycleID}
140+ if _, dup := present[id]; dup {
141+ continue
142+ }
143+ entry := routeEntryFor(daemon, session)
144+ if validateRouteEntryShape(entry) != nil {
145+ continue
146+ }
147+ present[id] = struct{}{}
148+ if entry.Attention {
149+ attentive[id] = struct{}{}
150+ }
151+ tier := routeTierStopped
152+ if session.State == catalogue.RemoteCatalogSessionUp {
153+ tier = routeTierLiveRemote
154+ if daemon.Local {
155+ tier = routeTierLiveLocal
156+ }
157+ }
158+ candidates = append(candidates, routeCandidate{id: id, entry: entry, tier: tier, lastUsed: session.LastUsedSeq, daemon: i, session: j})
159+ }
160+ }
161+
162+ // Attention onsets keep their order until the attention clears.
163+ for id := range l.attention {
164+ if _, ok := attentive[id]; !ok {
165+ delete(l.attention, id)
166+ }
167+ }
168+ for _, candidate := range candidates {
169+ if _, ok := attentive[candidate.id]; ok && l.attention[candidate.id] == 0 {
170+ l.attentionSeq++
171+ l.attention[candidate.id] = l.attentionSeq
172+ }
173+ }
174+
175+ sort.SliceStable(candidates, func(a, b int) bool {
176+ left, right := candidates[a], candidates[b]
177+ if la, ra := l.attached[left.id], l.attached[right.id]; la != ra {
178+ return la > ra
179+ }
180+ if left.tier != right.tier {
181+ return left.tier < right.tier
182+ }
183+ leftLocal, rightLocal := left.id.authority.local, right.id.authority.local
184+ if leftLocal != rightLocal {
185+ return leftLocal
186+ }
187+ if leftLocal && left.lastUsed != right.lastUsed {
188+ return left.lastUsed > right.lastUsed
189+ }
190+ if left.daemon != right.daemon {
191+ return left.daemon < right.daemon
192+ }
193+ return left.session < right.session
194+ })
195+
196+ live := make(map[routeID]struct{}, len(present)+len(snapshot.Daemons)+1)
197+ next := protocol.RecentRouteSnapshot{}
198+ targets := make(map[protocol.RouteRef]routeLedgerTarget)
199+ if active.known {
200+ live[activeID] = struct{}{}
201+ entry := routeActiveEntry(candidates, activeID, active, activeObservation)
202+ entry.Key, entry.Generation = l.keyFor(activeID), 1
203+ entry.AttentionSeq = l.attention[activeID]
204+ if validateRouteEntryShape(entry) == nil {
205+ next.ActiveEntry = entry
206+ next.Active = protocol.RouteRef{Key: entry.Key, Generation: entry.Generation}
207+ targets[next.Active] = routeLedgerTarget{authority: active.authority, target: entry.Target}
208+ if active.authority.local {
209+ next.Home = next.Active
210+ }
211+ }
212+ }
213+ for _, candidate := range candidates {
214+ if candidate.id == activeID && active.known {
215+ continue
216+ }
217+ if len(next.Entries) == protocol.RouteSnapshotMaxEntries {
218+ break
219+ }
220+ entry := candidate.entry
221+ entry.Key, entry.Generation = l.keyFor(candidate.id), 1
222+ entry.AttentionSeq = l.attention[candidate.id]
223+ ref := protocol.RouteRef{Key: entry.Key, Generation: entry.Generation}
224+ if next.Previous.IsZero() && l.attached[candidate.id] != 0 {
225+ next.Previous = ref
226+ }
227+ next.Entries = append(next.Entries, entry)
228+ targets[ref] = routeLedgerTarget{authority: candidate.id.authority, target: entry.Target}
229+ live[candidate.id] = struct{}{}
230+ }
231+ for _, daemon := range snapshot.Daemons {
232+ authority := routeAuthorityOf(daemon)
233+ if active.known && authority == active.authority {
234+ // The serving daemon creates in place; it is never a host row.
235+ continue
236+ }
237+ if len(next.Hosts) == protocol.RouteSnapshotMaxHosts {
238+ break
239+ }
240+ if daemon.Availability == domain.RemoteAvailabilityIncompatible || pickerObservationVersionMismatch(daemon) {
241+ continue
242+ }
243+ kind := protocol.RouteKindRemote
244+ if daemon.Local {
245+ kind = protocol.RouteKindLocal
246+ }
247+ label := pickerOriginLabel(daemon)
248+ if protocol.ValidateRouteLabel(label, false) != nil {
249+ continue
250+ }
251+ id := routeID{authority: authority}
252+ host := protocol.RouteHost{Key: l.keyFor(id), Generation: 1, Label: label, Kind: kind}
253+ next.Hosts = append(next.Hosts, host)
254+ targets[protocol.RouteRef{Key: host.Key, Generation: host.Generation}] = routeLedgerTarget{authority: authority, host: true}
255+ live[id] = struct{}{}
256+ }
257+ if len(next.Entries) == 0 {
258+ next.Entries = nil
259+ }
260+
261+ // Identities that left the catalogue lose their reference and recency.
262+ for id := range l.keys {
263+ if _, ok := live[id]; !ok {
264+ delete(l.keys, id)
265+ }
266+ }
267+ for id := range l.attached {
268+ if _, ok := live[id]; !ok {
269+ delete(l.attached, id)
270+ }
271+ }
272+
273+ if l.generation != 0 && routeSnapshotContentEqual(next, l.published) {
274+ return l.published, false
275+ }
276+ next.Generation = l.generation + 1
277+ if next.Validate() != nil {
278+ return l.published, false
279+ }
280+ l.generation = next.Generation
281+ l.published = next
282+ l.targets = targets
283+ return next, true
284+}
285+
286+// resolve returns what one reference of the latest published snapshot names.
287+func (l *routeLedger) resolve(ref protocol.RouteRef) (routeLedgerTarget, bool) {
288+ target, ok := l.targets[ref]
289+ return target, ok
290+}
291+
292+func (l *routeLedger) keyFor(id routeID) uint64 {
293+ if key, ok := l.keys[id]; ok {
294+ return key
295+ }
296+ l.nextKey++
297+ l.keys[id] = l.nextKey
298+ return l.nextKey
299+}
300+
301+// routeEntryFor projects one catalogue session. Identity fields are filled by
302+// the caller.
303+func routeEntryFor(daemon ports.BrokerDaemonObservation, session catalogue.RemoteCatalogSession) protocol.RecentRouteEntry {
304+ entry := protocol.RecentRouteEntry{
305+ Target: protocol.ExactSessionTarget{LifecycleID: session.LifecycleID, SessionName: session.Name},
306+ Name: session.Name,
307+ Kind: protocol.RouteKindLocal,
308+ Ephemeral: session.Ephemeral,
309+ Reachability: routeReachability(daemon),
310+ }
311+ if !daemon.Local {
312+ entry.Kind = protocol.RouteKindRemote
313+ entry.HostLabel = pickerOriginLabel(daemon)
314+ }
315+ for _, tab := range session.Tabs {
316+ if tab.Attention {
317+ entry.Attention = true
318+ break
319+ }
320+ }
321+ return entry
322+}
323+
324+// routeActiveEntry is the attached lifecycle's presentation: its catalogue
325+// row when the broker already observed it, otherwise the committed target.
326+func routeActiveEntry(candidates []routeCandidate, id routeID, active routeActive, observation *ports.BrokerDaemonObservation) protocol.RecentRouteEntry {
327+ for _, candidate := range candidates {
328+ if candidate.id == id {
329+ return candidate.entry
330+ }
331+ }
332+ entry := protocol.RecentRouteEntry{
333+ Target: active.target,
334+ Name: active.target.SessionName,
335+ Kind: protocol.RouteKindLocal,
336+ }
337+ if !active.authority.local {
338+ entry.Kind = protocol.RouteKindRemote
339+ if observation != nil {
340+ entry.HostLabel = pickerOriginLabel(*observation)
341+ }
342+ }
343+ return entry
344+}
345+
346+func routeReachability(daemon ports.BrokerDaemonObservation) protocol.RouteReachability {
347+ switch {
348+ case daemon.Availability == domain.RemoteAvailabilityReachable:
349+ return protocol.RouteReachabilityReachable
350+ case pickerObservationFailing(daemon) || daemon.Availability == domain.RemoteAvailabilityIncompatible:
351+ return protocol.RouteReachabilityUnavailable
352+ default:
353+ return protocol.RouteReachabilityUnknown
354+ }
355+}
356+
357+// validateRouteEntryShape checks the fields a catalogue row contributes, with
358+// a placeholder identity.
359+func validateRouteEntryShape(entry protocol.RecentRouteEntry) error {
360+ entry.Key, entry.Generation = 1, 1
361+ return protocol.RecentRouteSnapshot{Generation: 1, Entries: []protocol.RecentRouteEntry{entry}}.Validate()
362+}
363+
364+func routeSnapshotContentEqual(a, b protocol.RecentRouteSnapshot) bool {
365+ return a.Active == b.Active && a.ActiveEntry == b.ActiveEntry && a.Previous == b.Previous && a.Home == b.Home &&
366+ slices.Equal(a.Entries, b.Entries) && slices.Equal(a.Hosts, b.Hosts)
367+}
@@ -0,0 +1,367 @@
1+package client
2+
3+import (
4+ "slices"
5+ "sort"
6+
7+ "github.com/bnema/vev/internal/domain"
8+ "github.com/bnema/vev/internal/ports"
9+ "github.com/bnema/vev/internal/protocol"
10+ "github.com/bnema/vev/internal/protocol/catalogue"
11+)
12+
13+// Client-owned route ledger (Plan 003 C4/E3).
14+//
15+// The serving daemon renders the status-bar MRU, the other-session bells, the
16+// jump-recent hints, and the palette's remote, recent-route, and
17+// create-on-host rows from one bounded protocol.RecentRouteSnapshot the client
18+// publishes. The ledger builds that snapshot from the broker catalogue, which
19+// already observes every daemon (local and remote), so no daemon monitors
20+// another one.
21+//
22+// It is pure: no clock, no I/O, no lock. The supervisor's run goroutine owns
23+// it and feeds it each broker publication together with the attachment's
24+// committed target.
25+//
26+// - Every session lifecycle, and every daemon as a creation host, gets a
27+// RouteRef that stays stable while it remains in the catalogue. A new
28+// lifecycle of the same name is a new reference.
29+// - Entries are ordered by this client's attach recency, then live local
30+// sessions by catalogue LastUsedSeq, then live remote sessions in
31+// catalogue order, then stopped sessions.
32+// - The attached lifecycle is metadata only (ActiveEntry), never an entry.
33+// - Attention is any(tab.Attention). AttentionSeq orders onsets as the
34+// client observed them, across daemons, so the serving daemon can jump to
35+// the oldest one.
36+// - Generation advances only when the published content changes.
37+
38+// routeAuthority names one daemon in the broker catalogue.
39+type routeAuthority struct {
40+ local bool
41+ endpoint string
42+}
43+
44+func routeAuthorityOf(observation ports.BrokerDaemonObservation) routeAuthority {
45+ if observation.Local {
46+ return routeAuthority{local: true}
47+ }
48+ return routeAuthority{endpoint: observation.Endpoint}
49+}
50+
51+// routeID names one session lifecycle on one daemon. A zero lifecycle names
52+// the daemon itself as a creation host.
53+type routeID struct {
54+ authority routeAuthority
55+ lifecycle domain.SessionLifecycleID
56+}
57+
58+// routeLedgerTarget is what one published reference resolves to.
59+type routeLedgerTarget struct {
60+ authority routeAuthority
61+ // target is zero for a creation host.
62+ target protocol.ExactSessionTarget
63+ host bool
64+}
65+
66+// routeActive is the attachment the snapshot is published over.
67+type routeActive struct {
68+ known bool
69+ authority routeAuthority
70+ target protocol.ExactSessionTarget
71+}
72+
73+type routeLedger struct {
74+ nextKey uint64
75+ keys map[routeID]uint64
76+ attachSeq uint64
77+ attached map[routeID]uint64
78+ attentionSeq uint64
79+ attention map[routeID]uint64
80+ lastActive routeID
81+ generation uint64
82+ published protocol.RecentRouteSnapshot
83+ targets map[protocol.RouteRef]routeLedgerTarget
84+}
85+
86+func newRouteLedger() *routeLedger {
87+ return &routeLedger{
88+ keys: make(map[routeID]uint64),
89+ attached: make(map[routeID]uint64),
90+ attention: make(map[routeID]uint64),
91+ targets: make(map[protocol.RouteRef]routeLedgerTarget),
92+ }
93+}
94+
95+// routeCandidate is one catalogue session before ordering.
96+type routeCandidate struct {
97+ id routeID
98+ entry protocol.RecentRouteEntry
99+ tier int
100+ lastUsed uint64
101+ daemon int
102+ session int
103+}
104+
105+const (
106+ routeTierLiveLocal = iota
107+ routeTierLiveRemote
108+ routeTierStopped
109+)
110+
111+// build folds one broker publication and the attachment's committed target
112+// into the next snapshot. It reports false when nothing visible changed, in
113+// which case the previously published snapshot is returned unchanged.
114+func (l *routeLedger) build(snapshot ports.BrokerSnapshot, active routeActive) (protocol.RecentRouteSnapshot, bool) {
115+ activeID := routeID{}
116+ if active.known {
117+ activeID = routeID{authority: active.authority, lifecycle: active.target.LifecycleID}
118+ if activeID != l.lastActive {
119+ l.attachSeq++
120+ l.attached[activeID] = l.attachSeq
121+ l.lastActive = activeID
122+ }
123+ }
124+
125+ present := make(map[routeID]struct{})
126+ candidates := make([]routeCandidate, 0)
127+ attentive := make(map[routeID]struct{})
128+ var activeObservation *ports.BrokerDaemonObservation
129+ for i := range snapshot.Daemons {
130+ daemon := snapshot.Daemons[i]
131+ authority := routeAuthorityOf(daemon)
132+ if active.known && authority == active.authority {
133+ activeObservation = &snapshot.Daemons[i]
134+ }
135+ for j, session := range daemon.Sessions {
136+ if session.State == catalogue.RemoteCatalogSessionBroken {
137+ continue
138+ }
139+ id := routeID{authority: authority, lifecycle: session.LifecycleID}
140+ if _, dup := present[id]; dup {
141+ continue
142+ }
143+ entry := routeEntryFor(daemon, session)
144+ if validateRouteEntryShape(entry) != nil {
145+ continue
146+ }
147+ present[id] = struct{}{}
148+ if entry.Attention {
149+ attentive[id] = struct{}{}
150+ }
151+ tier := routeTierStopped
152+ if session.State == catalogue.RemoteCatalogSessionUp {
153+ tier = routeTierLiveRemote
154+ if daemon.Local {
155+ tier = routeTierLiveLocal
156+ }
157+ }
158+ candidates = append(candidates, routeCandidate{id: id, entry: entry, tier: tier, lastUsed: session.LastUsedSeq, daemon: i, session: j})
159+ }
160+ }
161+
162+ // Attention onsets keep their order until the attention clears.
163+ for id := range l.attention {
164+ if _, ok := attentive[id]; !ok {
165+ delete(l.attention, id)
166+ }
167+ }
168+ for _, candidate := range candidates {
169+ if _, ok := attentive[candidate.id]; ok && l.attention[candidate.id] == 0 {
170+ l.attentionSeq++
171+ l.attention[candidate.id] = l.attentionSeq
172+ }
173+ }
174+
175+ sort.SliceStable(candidates, func(a, b int) bool {
176+ left, right := candidates[a], candidates[b]
177+ if la, ra := l.attached[left.id], l.attached[right.id]; la != ra {
178+ return la > ra
179+ }
180+ if left.tier != right.tier {
181+ return left.tier < right.tier
182+ }
183+ leftLocal, rightLocal := left.id.authority.local, right.id.authority.local
184+ if leftLocal != rightLocal {
185+ return leftLocal
186+ }
187+ if leftLocal && left.lastUsed != right.lastUsed {
188+ return left.lastUsed > right.lastUsed
189+ }
190+ if left.daemon != right.daemon {
191+ return left.daemon < right.daemon
192+ }
193+ return left.session < right.session
194+ })
195+
196+ live := make(map[routeID]struct{}, len(present)+len(snapshot.Daemons)+1)
197+ next := protocol.RecentRouteSnapshot{}
198+ targets := make(map[protocol.RouteRef]routeLedgerTarget)
199+ if active.known {
200+ live[activeID] = struct{}{}
201+ entry := routeActiveEntry(candidates, activeID, active, activeObservation)
202+ entry.Key, entry.Generation = l.keyFor(activeID), 1
203+ entry.AttentionSeq = l.attention[activeID]
204+ if validateRouteEntryShape(entry) == nil {
205+ next.ActiveEntry = entry
206+ next.Active = protocol.RouteRef{Key: entry.Key, Generation: entry.Generation}
207+ targets[next.Active] = routeLedgerTarget{authority: active.authority, target: entry.Target}
208+ if active.authority.local {
209+ next.Home = next.Active
210+ }
211+ }
212+ }
213+ for _, candidate := range candidates {
214+ if candidate.id == activeID && active.known {
215+ continue
216+ }
217+ if len(next.Entries) == protocol.RouteSnapshotMaxEntries {
218+ break
219+ }
220+ entry := candidate.entry
221+ entry.Key, entry.Generation = l.keyFor(candidate.id), 1
222+ entry.AttentionSeq = l.attention[candidate.id]
223+ ref := protocol.RouteRef{Key: entry.Key, Generation: entry.Generation}
224+ if next.Previous.IsZero() && l.attached[candidate.id] != 0 {
225+ next.Previous = ref
226+ }
227+ next.Entries = append(next.Entries, entry)
228+ targets[ref] = routeLedgerTarget{authority: candidate.id.authority, target: entry.Target}
229+ live[candidate.id] = struct{}{}
230+ }
231+ for _, daemon := range snapshot.Daemons {
232+ authority := routeAuthorityOf(daemon)
233+ if active.known && authority == active.authority {
234+ // The serving daemon creates in place; it is never a host row.
235+ continue
236+ }
237+ if len(next.Hosts) == protocol.RouteSnapshotMaxHosts {
238+ break
239+ }
240+ if daemon.Availability == domain.RemoteAvailabilityIncompatible || pickerObservationVersionMismatch(daemon) {
241+ continue
242+ }
243+ kind := protocol.RouteKindRemote
244+ if daemon.Local {
245+ kind = protocol.RouteKindLocal
246+ }
247+ label := pickerOriginLabel(daemon)
248+ if protocol.ValidateRouteLabel(label, false) != nil {
249+ continue
250+ }
251+ id := routeID{authority: authority}
252+ host := protocol.RouteHost{Key: l.keyFor(id), Generation: 1, Label: label, Kind: kind}
253+ next.Hosts = append(next.Hosts, host)
254+ targets[protocol.RouteRef{Key: host.Key, Generation: host.Generation}] = routeLedgerTarget{authority: authority, host: true}
255+ live[id] = struct{}{}
256+ }
257+ if len(next.Entries) == 0 {
258+ next.Entries = nil
259+ }
260+
261+ // Identities that left the catalogue lose their reference and recency.
262+ for id := range l.keys {
263+ if _, ok := live[id]; !ok {
264+ delete(l.keys, id)
265+ }
266+ }
267+ for id := range l.attached {
268+ if _, ok := live[id]; !ok {
269+ delete(l.attached, id)
270+ }
271+ }
272+
273+ if l.generation != 0 && routeSnapshotContentEqual(next, l.published) {
274+ return l.published, false
275+ }
276+ next.Generation = l.generation + 1
277+ if next.Validate() != nil {
278+ return l.published, false
279+ }
280+ l.generation = next.Generation
281+ l.published = next
282+ l.targets = targets
283+ return next, true
284+}
285+
286+// resolve returns what one reference of the latest published snapshot names.
287+func (l *routeLedger) resolve(ref protocol.RouteRef) (routeLedgerTarget, bool) {
288+ target, ok := l.targets[ref]
289+ return target, ok
290+}
291+
292+func (l *routeLedger) keyFor(id routeID) uint64 {
293+ if key, ok := l.keys[id]; ok {
294+ return key
295+ }
296+ l.nextKey++
297+ l.keys[id] = l.nextKey
298+ return l.nextKey
299+}
300+
301+// routeEntryFor projects one catalogue session. Identity fields are filled by
302+// the caller.
303+func routeEntryFor(daemon ports.BrokerDaemonObservation, session catalogue.RemoteCatalogSession) protocol.RecentRouteEntry {
304+ entry := protocol.RecentRouteEntry{
305+ Target: protocol.ExactSessionTarget{LifecycleID: session.LifecycleID, SessionName: session.Name},
306+ Name: session.Name,
307+ Kind: protocol.RouteKindLocal,
308+ Ephemeral: session.Ephemeral,
309+ Reachability: routeReachability(daemon),
310+ }
311+ if !daemon.Local {
312+ entry.Kind = protocol.RouteKindRemote
313+ entry.HostLabel = pickerOriginLabel(daemon)
314+ }
315+ for _, tab := range session.Tabs {
316+ if tab.Attention {
317+ entry.Attention = true
318+ break
319+ }
320+ }
321+ return entry
322+}
323+
324+// routeActiveEntry is the attached lifecycle's presentation: its catalogue
325+// row when the broker already observed it, otherwise the committed target.
326+func routeActiveEntry(candidates []routeCandidate, id routeID, active routeActive, observation *ports.BrokerDaemonObservation) protocol.RecentRouteEntry {
327+ for _, candidate := range candidates {
328+ if candidate.id == id {
329+ return candidate.entry
330+ }
331+ }
332+ entry := protocol.RecentRouteEntry{
333+ Target: active.target,
334+ Name: active.target.SessionName,
335+ Kind: protocol.RouteKindLocal,
336+ }
337+ if !active.authority.local {
338+ entry.Kind = protocol.RouteKindRemote
339+ if observation != nil {
340+ entry.HostLabel = pickerOriginLabel(*observation)
341+ }
342+ }
343+ return entry
344+}
345+
346+func routeReachability(daemon ports.BrokerDaemonObservation) protocol.RouteReachability {
347+ switch {
348+ case daemon.Availability == domain.RemoteAvailabilityReachable:
349+ return protocol.RouteReachabilityReachable
350+ case pickerObservationFailing(daemon) || daemon.Availability == domain.RemoteAvailabilityIncompatible:
351+ return protocol.RouteReachabilityUnavailable
352+ default:
353+ return protocol.RouteReachabilityUnknown
354+ }
355+}
356+
357+// validateRouteEntryShape checks the fields a catalogue row contributes, with
358+// a placeholder identity.
359+func validateRouteEntryShape(entry protocol.RecentRouteEntry) error {
360+ entry.Key, entry.Generation = 1, 1
361+ return protocol.RecentRouteSnapshot{Generation: 1, Entries: []protocol.RecentRouteEntry{entry}}.Validate()
362+}
363+
364+func routeSnapshotContentEqual(a, b protocol.RecentRouteSnapshot) bool {
365+ return a.Active == b.Active && a.ActiveEntry == b.ActiveEntry && a.Previous == b.Previous && a.Home == b.Home &&
366+ slices.Equal(a.Entries, b.Entries) && slices.Equal(a.Hosts, b.Hosts)
367+}
A internal/usecase/client/route_ledger_test.go +264 -0
@@ -0,0 +1,264 @@
1+package client
2+
3+import (
4+ "fmt"
5+ "testing"
6+ "time"
7+
8+ "github.com/stretchr/testify/require"
9+
10+ "github.com/bnema/vev/internal/domain"
11+ "github.com/bnema/vev/internal/ports"
12+ "github.com/bnema/vev/internal/protocol"
13+ "github.com/bnema/vev/internal/protocol/catalogue"
14+)
15+
16+// Route ledger (Plan 003 C4/E3), ported from main's routes_test observation
17+// and ordering cases onto the broker catalogue.
18+
19+const routeTestRemote = "vev@box"
20+
21+func routeTestSnapshot(daemons ...ports.BrokerDaemonObservation) ports.BrokerSnapshot {
22+ return ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: daemons}
23+}
24+
25+func routeTestLocal(sessions ...catalogue.RemoteCatalogSession) ports.BrokerDaemonObservation {
26+ return pickerTestLocalObservation(time.Unix(1000, 0), sessions...)
27+}
28+
29+func routeTestRemoteHost(sessions ...catalogue.RemoteCatalogSession) ports.BrokerDaemonObservation {
30+ return pickerTestRemoteObservation(routeTestRemote, 1, 2, time.Unix(1000, 0), sessions...)
31+}
32+
33+func routeTestLive(name string, seed byte, lastUsed uint64, attention ...bool) catalogue.RemoteCatalogSession {
34+ tabs := make([]catalogue.RemoteCatalogTab, 0, len(attention))
35+ for i, bell := range attention {
36+ tabs = append(tabs, pickerTestTab(fmt.Sprintf("t%d", i), fmt.Sprintf("tab%d", i), "", bell))
37+ }
38+ return pickerTestTabbed(name, seed, catalogue.RemoteCatalogSessionUp, lastUsed, "", tabs...)
39+}
40+
41+func routeTestStopped(name string, seed byte, lastUsed uint64) catalogue.RemoteCatalogSession {
42+ return pickerTestTabbed(name, seed, catalogue.RemoteCatalogSessionDown, lastUsed, "")
43+}
44+
45+func routeTestActive(local bool, name string, seed byte) routeActive {
46+ authority := routeAuthority{local: true}
47+ if !local {
48+ authority = routeAuthority{endpoint: routeTestRemote}
49+ }
50+ return routeActive{known: true, authority: authority, target: protocol.ExactSessionTarget{LifecycleID: pickerTestLifecycle(seed), SessionName: name}}
51+}
52+
53+func routeEntryNames(snapshot protocol.RecentRouteSnapshot) []string {
54+ names := make([]string, 0, len(snapshot.Entries))
55+ for _, entry := range snapshot.Entries {
56+ names = append(names, entry.Name)
57+ }
58+ return names
59+}
60+
61+func TestRouteLedgerOrdering(t *testing.T) {
62+ tests := []struct {
63+ name string
64+ visits []routeActive
65+ build ports.BrokerSnapshot
66+ active routeActive
67+ want []string
68+ }{
69+ {
70+ name: "live local by recency, then remote in catalogue order, then stopped",
71+ build: routeTestSnapshot(
72+ routeTestLocal(routeTestLive("old", 1, 1), routeTestStopped("gone", 2, 9), routeTestLive("new", 3, 5), routeTestLive("here", 4, 7)),
73+ routeTestRemoteHost(routeTestLive("r2", 5, 0), routeTestLive("r1", 6, 0), routeTestStopped("rs", 7, 0)),
74+ ),
75+ active: routeTestActive(true, "here", 4),
76+ want: []string{"new", "old", "r2", "r1", "gone", "rs"},
77+ },
78+ {
79+ name: "this client's attach recency comes first",
80+ visits: []routeActive{routeTestActive(false, "r1", 6), routeTestActive(true, "old", 1)},
81+ build: routeTestSnapshot(
82+ routeTestLocal(routeTestLive("old", 1, 1), routeTestLive("new", 3, 5), routeTestLive("here", 4, 7)),
83+ routeTestRemoteHost(routeTestLive("r2", 5, 0), routeTestLive("r1", 6, 0)),
84+ ),
85+ active: routeTestActive(true, "here", 4),
86+ want: []string{"old", "r1", "new", "r2"},
87+ },
88+ {
89+ name: "broken sessions are never routes",
90+ build: routeTestSnapshot(routeTestLocal(routeTestLive("ok", 1, 1), pickerTestTabbed("bad", 2, catalogue.RemoteCatalogSessionBroken, 5, ""))),
91+ active: routeTestActive(true, "here", 4),
92+ want: []string{"ok"},
93+ },
94+ }
95+ for _, tt := range tests {
96+ t.Run(tt.name, func(t *testing.T) {
97+ ledger := newRouteLedger()
98+ for _, visit := range tt.visits {
99+ ledger.build(tt.build, visit)
100+ }
101+ snapshot, _ := ledger.build(tt.build, tt.active)
102+ require.NoError(t, snapshot.Validate())
103+ require.Equal(t, tt.want, routeEntryNames(snapshot))
104+ require.Equal(t, tt.active.target, snapshot.ActiveEntry.Target, "the attached lifecycle is metadata only")
105+ })
106+ }
107+}
108+
109+func TestRouteLedgerActivePreviousAndHome(t *testing.T) {
110+ build := routeTestSnapshot(
111+ routeTestLocal(routeTestLive("a", 1, 2), routeTestLive("b", 2, 1)),
112+ routeTestRemoteHost(routeTestLive("r", 3, 0)),
113+ )
114+ tests := []struct {
115+ name string
116+ active routeActive
117+ wantKind protocol.RouteKind
118+ wantLabel string
119+ wantHome bool
120+ wantPrevious string
121+ wantHosts []protocol.RouteKind
122+ }{
123+ {name: "local attachment is home", active: routeTestActive(true, "a", 1), wantKind: protocol.RouteKindLocal, wantHome: true, wantPrevious: "r", wantHosts: []protocol.RouteKind{protocol.RouteKindRemote}},
124+ {name: "remote attachment offers the local host", active: routeTestActive(false, "r", 3), wantKind: protocol.RouteKindRemote, wantLabel: routeTestRemote, wantPrevious: "a", wantHosts: []protocol.RouteKind{protocol.RouteKindLocal}},
125+ {name: "an unobserved attachment is synthesized", active: routeTestActive(true, "fresh", 9), wantKind: protocol.RouteKindLocal, wantHome: true, wantPrevious: "r", wantHosts: []protocol.RouteKind{protocol.RouteKindRemote}},
126+ }
127+ for _, tt := range tests {
128+ t.Run(tt.name, func(t *testing.T) {
129+ ledger := newRouteLedger()
130+ // Visit the other kind first so Previous names it.
131+ if tt.active.authority.local {
132+ ledger.build(build, routeTestActive(false, "r", 3))
133+ } else {
134+ ledger.build(build, routeTestActive(true, "a", 1))
135+ }
136+ snapshot, changed := ledger.build(build, tt.active)
137+ require.True(t, changed)
138+ require.NoError(t, snapshot.Validate())
139+ require.Equal(t, tt.wantKind, snapshot.ActiveEntry.Kind)
140+ require.Equal(t, tt.wantLabel, snapshot.ActiveEntry.HostLabel)
141+ require.Equal(t, snapshot.Active, protocol.RouteRef{Key: snapshot.ActiveEntry.Key, Generation: snapshot.ActiveEntry.Generation})
142+ if tt.wantHome {
143+ require.Equal(t, snapshot.Active, snapshot.Home)
144+ } else {
145+ require.True(t, snapshot.Home.IsZero())
146+ }
147+ previous, ok := ledger.resolve(snapshot.Previous)
148+ require.True(t, ok)
149+ require.Equal(t, tt.wantPrevious, previous.target.SessionName)
150+ kinds := make([]protocol.RouteKind, 0, len(snapshot.Hosts))
151+ for _, host := range snapshot.Hosts {
152+ kinds = append(kinds, host.Kind)
153+ resolved, ok := ledger.resolve(protocol.RouteRef{Key: host.Key, Generation: host.Generation})
154+ require.True(t, ok)
155+ require.True(t, resolved.host)
156+ require.NotEqual(t, tt.active.authority, resolved.authority, "the serving daemon is never a host row")
157+ }
158+ require.Equal(t, tt.wantHosts, kinds)
159+ })
160+ }
161+}
162+
163+func TestRouteLedgerStableReferencesAndGeneration(t *testing.T) {
164+ ledger := newRouteLedger()
165+ active := routeTestActive(true, "here", 4)
166+ first, changed := ledger.build(routeTestSnapshot(routeTestLocal(routeTestLive("a", 1, 1), routeTestLive("here", 4, 2))), active)
167+ require.True(t, changed)
168+ require.Equal(t, uint64(1), first.Generation)
169+
170+ same, changed := ledger.build(routeTestSnapshot(routeTestLocal(routeTestLive("a", 1, 1), routeTestLive("here", 4, 2))), active)
171+ require.False(t, changed, "an unchanged publication never advances the generation")
172+ require.Equal(t, first, same)
173+
174+ grown, changed := ledger.build(routeTestSnapshot(routeTestLocal(routeTestLive("a", 1, 1), routeTestLive("b", 2, 3), routeTestLive("here", 4, 2))), active)
175+ require.True(t, changed)
176+ require.Equal(t, uint64(2), grown.Generation)
177+ require.Equal(t, first.Active, grown.Active, "a lifecycle keeps its reference")
178+ require.Equal(t, first.Entries[0].Key, grown.Entries[1].Key)
179+
180+ replaced, _ := ledger.build(routeTestSnapshot(routeTestLocal(routeTestLive("a", 5, 1), routeTestLive("here", 4, 2))), active)
181+ require.NotEqual(t, first.Entries[0].Key, replaced.Entries[0].Key, "a new lifecycle of the same name is a new reference")
182+ _, ok := ledger.resolve(protocol.RouteRef{Key: first.Entries[0].Key, Generation: 1})
183+ require.False(t, ok, "a retired lifecycle no longer resolves")
184+}
185+
186+// TestRouteLedgerAttention ports main's local and remote attention
187+// observations: a session carries attention while any of its tabs does, and
188+// onsets are ordered as the client observed them across daemons.
189+func TestRouteLedgerAttention(t *testing.T) {
190+ active := routeTestActive(true, "here", 4)
191+ steps := []struct {
192+ name string
193+ build ports.BrokerSnapshot
194+ want map[string]bool
195+ order []string
196+ }{
197+ {
198+ name: "quiet",
199+ build: routeTestSnapshot(routeTestLocal(routeTestLive("here", 4, 9), routeTestLive("loc", 1, 1, false, false)), routeTestRemoteHost(routeTestLive("rem", 2, 0, false))),
200+ want: map[string]bool{"loc": false, "rem": false},
201+ },
202+ {
203+ name: "remote bell first",
204+ build: routeTestSnapshot(routeTestLocal(routeTestLive("here", 4, 9), routeTestLive("loc", 1, 1, false, false)), routeTestRemoteHost(routeTestLive("rem", 2, 0, true))),
205+ want: map[string]bool{"loc": false, "rem": true},
206+ order: []string{"rem"},
207+ },
208+ {
209+ name: "then a local tab bell",
210+ build: routeTestSnapshot(routeTestLocal(routeTestLive("here", 4, 9), routeTestLive("loc", 1, 1, false, true)), routeTestRemoteHost(routeTestLive("rem", 2, 0, true))),
211+ want: map[string]bool{"loc": true, "rem": true},
212+ order: []string{"rem", "loc"},
213+ },
214+ {
215+ name: "remote cleared, then rings again later",
216+ build: routeTestSnapshot(routeTestLocal(routeTestLive("here", 4, 9), routeTestLive("loc", 1, 1, false, true)), routeTestRemoteHost(routeTestLive("rem", 2, 0, false))),
217+ want: map[string]bool{"loc": true, "rem": false},
218+ order: []string{"loc"},
219+ },
220+ {
221+ name: "rings again",
222+ build: routeTestSnapshot(routeTestLocal(routeTestLive("here", 4, 9), routeTestLive("loc", 1, 1, false, true)), routeTestRemoteHost(routeTestLive("rem", 2, 0, true))),
223+ want: map[string]bool{"loc": true, "rem": true},
224+ order: []string{"loc", "rem"},
225+ },
226+ }
227+ ledger := newRouteLedger()
228+ for _, step := range steps {
229+ t.Run(step.name, func(t *testing.T) {
230+ snapshot, _ := ledger.build(step.build, active)
231+ require.NoError(t, snapshot.Validate())
232+ got := map[string]bool{}
233+ seq := map[string]uint64{}
234+ for _, entry := range snapshot.Entries {
235+ got[entry.Name] = entry.Attention
236+ if entry.Attention {
237+ require.NotZero(t, entry.AttentionSeq)
238+ seq[entry.Name] = entry.AttentionSeq
239+ } else {
240+ require.Zero(t, entry.AttentionSeq)
241+ }
242+ }
243+ require.Equal(t, step.want, got)
244+ for i := 1; i < len(step.order); i++ {
245+ require.Less(t, seq[step.order[i-1]], seq[step.order[i]], "%s rang before %s", step.order[i-1], step.order[i])
246+ }
247+ })
248+ }
249+}
250+
251+func TestRouteLedgerBounds(t *testing.T) {
252+ sessions := make([]catalogue.RemoteCatalogSession, 0, protocol.RouteSnapshotMaxEntries+5)
253+ for i := 0; i < protocol.RouteSnapshotMaxEntries+5; i++ {
254+ sessions = append(sessions, routeTestLive(fmt.Sprintf("s%02d", i), byte(i+10), uint64(i)))
255+ }
256+ incompatible := pickerTestRemoteObservation("old@box", 3, 4, time.Unix(1000, 0))
257+ incompatible.Availability = domain.RemoteAvailabilityIncompatible
258+ ledger := newRouteLedger()
259+ snapshot, changed := ledger.build(routeTestSnapshot(routeTestLocal(sessions...), incompatible), routeTestActive(true, "s00", 10))
260+ require.True(t, changed)
261+ require.NoError(t, snapshot.Validate())
262+ require.Len(t, snapshot.Entries, protocol.RouteSnapshotMaxEntries)
263+ require.Empty(t, snapshot.Hosts, "an incompatible daemon is never a creation host")
264+}
@@ -0,0 +1,264 @@
1+package client
2+
3+import (
4+ "fmt"
5+ "testing"
6+ "time"
7+
8+ "github.com/stretchr/testify/require"
9+
10+ "github.com/bnema/vev/internal/domain"
11+ "github.com/bnema/vev/internal/ports"
12+ "github.com/bnema/vev/internal/protocol"
13+ "github.com/bnema/vev/internal/protocol/catalogue"
14+)
15+
16+// Route ledger (Plan 003 C4/E3), ported from main's routes_test observation
17+// and ordering cases onto the broker catalogue.
18+
19+const routeTestRemote = "vev@box"
20+
21+func routeTestSnapshot(daemons ...ports.BrokerDaemonObservation) ports.BrokerSnapshot {
22+ return ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: daemons}
23+}
24+
25+func routeTestLocal(sessions ...catalogue.RemoteCatalogSession) ports.BrokerDaemonObservation {
26+ return pickerTestLocalObservation(time.Unix(1000, 0), sessions...)
27+}
28+
29+func routeTestRemoteHost(sessions ...catalogue.RemoteCatalogSession) ports.BrokerDaemonObservation {
30+ return pickerTestRemoteObservation(routeTestRemote, 1, 2, time.Unix(1000, 0), sessions...)
31+}
32+
33+func routeTestLive(name string, seed byte, lastUsed uint64, attention ...bool) catalogue.RemoteCatalogSession {
34+ tabs := make([]catalogue.RemoteCatalogTab, 0, len(attention))
35+ for i, bell := range attention {
36+ tabs = append(tabs, pickerTestTab(fmt.Sprintf("t%d", i), fmt.Sprintf("tab%d", i), "", bell))
37+ }
38+ return pickerTestTabbed(name, seed, catalogue.RemoteCatalogSessionUp, lastUsed, "", tabs...)
39+}
40+
41+func routeTestStopped(name string, seed byte, lastUsed uint64) catalogue.RemoteCatalogSession {
42+ return pickerTestTabbed(name, seed, catalogue.RemoteCatalogSessionDown, lastUsed, "")
43+}
44+
45+func routeTestActive(local bool, name string, seed byte) routeActive {
46+ authority := routeAuthority{local: true}
47+ if !local {
48+ authority = routeAuthority{endpoint: routeTestRemote}
49+ }
50+ return routeActive{known: true, authority: authority, target: protocol.ExactSessionTarget{LifecycleID: pickerTestLifecycle(seed), SessionName: name}}
51+}
52+
53+func routeEntryNames(snapshot protocol.RecentRouteSnapshot) []string {
54+ names := make([]string, 0, len(snapshot.Entries))
55+ for _, entry := range snapshot.Entries {
56+ names = append(names, entry.Name)
57+ }
58+ return names
59+}
60+
61+func TestRouteLedgerOrdering(t *testing.T) {
62+ tests := []struct {
63+ name string
64+ visits []routeActive
65+ build ports.BrokerSnapshot
66+ active routeActive
67+ want []string
68+ }{
69+ {
70+ name: "live local by recency, then remote in catalogue order, then stopped",
71+ build: routeTestSnapshot(
72+ routeTestLocal(routeTestLive("old", 1, 1), routeTestStopped("gone", 2, 9), routeTestLive("new", 3, 5), routeTestLive("here", 4, 7)),
73+ routeTestRemoteHost(routeTestLive("r2", 5, 0), routeTestLive("r1", 6, 0), routeTestStopped("rs", 7, 0)),
74+ ),
75+ active: routeTestActive(true, "here", 4),
76+ want: []string{"new", "old", "r2", "r1", "gone", "rs"},
77+ },
78+ {
79+ name: "this client's attach recency comes first",
80+ visits: []routeActive{routeTestActive(false, "r1", 6), routeTestActive(true, "old", 1)},
81+ build: routeTestSnapshot(
82+ routeTestLocal(routeTestLive("old", 1, 1), routeTestLive("new", 3, 5), routeTestLive("here", 4, 7)),
83+ routeTestRemoteHost(routeTestLive("r2", 5, 0), routeTestLive("r1", 6, 0)),
84+ ),
85+ active: routeTestActive(true, "here", 4),
86+ want: []string{"old", "r1", "new", "r2"},
87+ },
88+ {
89+ name: "broken sessions are never routes",
90+ build: routeTestSnapshot(routeTestLocal(routeTestLive("ok", 1, 1), pickerTestTabbed("bad", 2, catalogue.RemoteCatalogSessionBroken, 5, ""))),
91+ active: routeTestActive(true, "here", 4),
92+ want: []string{"ok"},
93+ },
94+ }
95+ for _, tt := range tests {
96+ t.Run(tt.name, func(t *testing.T) {
97+ ledger := newRouteLedger()
98+ for _, visit := range tt.visits {
99+ ledger.build(tt.build, visit)
100+ }
101+ snapshot, _ := ledger.build(tt.build, tt.active)
102+ require.NoError(t, snapshot.Validate())
103+ require.Equal(t, tt.want, routeEntryNames(snapshot))
104+ require.Equal(t, tt.active.target, snapshot.ActiveEntry.Target, "the attached lifecycle is metadata only")
105+ })
106+ }
107+}
108+
109+func TestRouteLedgerActivePreviousAndHome(t *testing.T) {
110+ build := routeTestSnapshot(
111+ routeTestLocal(routeTestLive("a", 1, 2), routeTestLive("b", 2, 1)),
112+ routeTestRemoteHost(routeTestLive("r", 3, 0)),
113+ )
114+ tests := []struct {
115+ name string
116+ active routeActive
117+ wantKind protocol.RouteKind
118+ wantLabel string
119+ wantHome bool
120+ wantPrevious string
121+ wantHosts []protocol.RouteKind
122+ }{
123+ {name: "local attachment is home", active: routeTestActive(true, "a", 1), wantKind: protocol.RouteKindLocal, wantHome: true, wantPrevious: "r", wantHosts: []protocol.RouteKind{protocol.RouteKindRemote}},
124+ {name: "remote attachment offers the local host", active: routeTestActive(false, "r", 3), wantKind: protocol.RouteKindRemote, wantLabel: routeTestRemote, wantPrevious: "a", wantHosts: []protocol.RouteKind{protocol.RouteKindLocal}},
125+ {name: "an unobserved attachment is synthesized", active: routeTestActive(true, "fresh", 9), wantKind: protocol.RouteKindLocal, wantHome: true, wantPrevious: "r", wantHosts: []protocol.RouteKind{protocol.RouteKindRemote}},
126+ }
127+ for _, tt := range tests {
128+ t.Run(tt.name, func(t *testing.T) {
129+ ledger := newRouteLedger()
130+ // Visit the other kind first so Previous names it.
131+ if tt.active.authority.local {
132+ ledger.build(build, routeTestActive(false, "r", 3))
133+ } else {
134+ ledger.build(build, routeTestActive(true, "a", 1))
135+ }
136+ snapshot, changed := ledger.build(build, tt.active)
137+ require.True(t, changed)
138+ require.NoError(t, snapshot.Validate())
139+ require.Equal(t, tt.wantKind, snapshot.ActiveEntry.Kind)
140+ require.Equal(t, tt.wantLabel, snapshot.ActiveEntry.HostLabel)
141+ require.Equal(t, snapshot.Active, protocol.RouteRef{Key: snapshot.ActiveEntry.Key, Generation: snapshot.ActiveEntry.Generation})
142+ if tt.wantHome {
143+ require.Equal(t, snapshot.Active, snapshot.Home)
144+ } else {
145+ require.True(t, snapshot.Home.IsZero())
146+ }
147+ previous, ok := ledger.resolve(snapshot.Previous)
148+ require.True(t, ok)
149+ require.Equal(t, tt.wantPrevious, previous.target.SessionName)
150+ kinds := make([]protocol.RouteKind, 0, len(snapshot.Hosts))
151+ for _, host := range snapshot.Hosts {
152+ kinds = append(kinds, host.Kind)
153+ resolved, ok := ledger.resolve(protocol.RouteRef{Key: host.Key, Generation: host.Generation})
154+ require.True(t, ok)
155+ require.True(t, resolved.host)
156+ require.NotEqual(t, tt.active.authority, resolved.authority, "the serving daemon is never a host row")
157+ }
158+ require.Equal(t, tt.wantHosts, kinds)
159+ })
160+ }
161+}
162+
163+func TestRouteLedgerStableReferencesAndGeneration(t *testing.T) {
164+ ledger := newRouteLedger()
165+ active := routeTestActive(true, "here", 4)
166+ first, changed := ledger.build(routeTestSnapshot(routeTestLocal(routeTestLive("a", 1, 1), routeTestLive("here", 4, 2))), active)
167+ require.True(t, changed)
168+ require.Equal(t, uint64(1), first.Generation)
169+
170+ same, changed := ledger.build(routeTestSnapshot(routeTestLocal(routeTestLive("a", 1, 1), routeTestLive("here", 4, 2))), active)
171+ require.False(t, changed, "an unchanged publication never advances the generation")
172+ require.Equal(t, first, same)
173+
174+ grown, changed := ledger.build(routeTestSnapshot(routeTestLocal(routeTestLive("a", 1, 1), routeTestLive("b", 2, 3), routeTestLive("here", 4, 2))), active)
175+ require.True(t, changed)
176+ require.Equal(t, uint64(2), grown.Generation)
177+ require.Equal(t, first.Active, grown.Active, "a lifecycle keeps its reference")
178+ require.Equal(t, first.Entries[0].Key, grown.Entries[1].Key)
179+
180+ replaced, _ := ledger.build(routeTestSnapshot(routeTestLocal(routeTestLive("a", 5, 1), routeTestLive("here", 4, 2))), active)
181+ require.NotEqual(t, first.Entries[0].Key, replaced.Entries[0].Key, "a new lifecycle of the same name is a new reference")
182+ _, ok := ledger.resolve(protocol.RouteRef{Key: first.Entries[0].Key, Generation: 1})
183+ require.False(t, ok, "a retired lifecycle no longer resolves")
184+}
185+
186+// TestRouteLedgerAttention ports main's local and remote attention
187+// observations: a session carries attention while any of its tabs does, and
188+// onsets are ordered as the client observed them across daemons.
189+func TestRouteLedgerAttention(t *testing.T) {
190+ active := routeTestActive(true, "here", 4)
191+ steps := []struct {
192+ name string
193+ build ports.BrokerSnapshot
194+ want map[string]bool
195+ order []string
196+ }{
197+ {
198+ name: "quiet",
199+ build: routeTestSnapshot(routeTestLocal(routeTestLive("here", 4, 9), routeTestLive("loc", 1, 1, false, false)), routeTestRemoteHost(routeTestLive("rem", 2, 0, false))),
200+ want: map[string]bool{"loc": false, "rem": false},
201+ },
202+ {
203+ name: "remote bell first",
204+ build: routeTestSnapshot(routeTestLocal(routeTestLive("here", 4, 9), routeTestLive("loc", 1, 1, false, false)), routeTestRemoteHost(routeTestLive("rem", 2, 0, true))),
205+ want: map[string]bool{"loc": false, "rem": true},
206+ order: []string{"rem"},
207+ },
208+ {
209+ name: "then a local tab bell",
210+ build: routeTestSnapshot(routeTestLocal(routeTestLive("here", 4, 9), routeTestLive("loc", 1, 1, false, true)), routeTestRemoteHost(routeTestLive("rem", 2, 0, true))),
211+ want: map[string]bool{"loc": true, "rem": true},
212+ order: []string{"rem", "loc"},
213+ },
214+ {
215+ name: "remote cleared, then rings again later",
216+ build: routeTestSnapshot(routeTestLocal(routeTestLive("here", 4, 9), routeTestLive("loc", 1, 1, false, true)), routeTestRemoteHost(routeTestLive("rem", 2, 0, false))),
217+ want: map[string]bool{"loc": true, "rem": false},
218+ order: []string{"loc"},
219+ },
220+ {
221+ name: "rings again",
222+ build: routeTestSnapshot(routeTestLocal(routeTestLive("here", 4, 9), routeTestLive("loc", 1, 1, false, true)), routeTestRemoteHost(routeTestLive("rem", 2, 0, true))),
223+ want: map[string]bool{"loc": true, "rem": true},
224+ order: []string{"loc", "rem"},
225+ },
226+ }
227+ ledger := newRouteLedger()
228+ for _, step := range steps {
229+ t.Run(step.name, func(t *testing.T) {
230+ snapshot, _ := ledger.build(step.build, active)
231+ require.NoError(t, snapshot.Validate())
232+ got := map[string]bool{}
233+ seq := map[string]uint64{}
234+ for _, entry := range snapshot.Entries {
235+ got[entry.Name] = entry.Attention
236+ if entry.Attention {
237+ require.NotZero(t, entry.AttentionSeq)
238+ seq[entry.Name] = entry.AttentionSeq
239+ } else {
240+ require.Zero(t, entry.AttentionSeq)
241+ }
242+ }
243+ require.Equal(t, step.want, got)
244+ for i := 1; i < len(step.order); i++ {
245+ require.Less(t, seq[step.order[i-1]], seq[step.order[i]], "%s rang before %s", step.order[i-1], step.order[i])
246+ }
247+ })
248+ }
249+}
250+
251+func TestRouteLedgerBounds(t *testing.T) {
252+ sessions := make([]catalogue.RemoteCatalogSession, 0, protocol.RouteSnapshotMaxEntries+5)
253+ for i := 0; i < protocol.RouteSnapshotMaxEntries+5; i++ {
254+ sessions = append(sessions, routeTestLive(fmt.Sprintf("s%02d", i), byte(i+10), uint64(i)))
255+ }
256+ incompatible := pickerTestRemoteObservation("old@box", 3, 4, time.Unix(1000, 0))
257+ incompatible.Availability = domain.RemoteAvailabilityIncompatible
258+ ledger := newRouteLedger()
259+ snapshot, changed := ledger.build(routeTestSnapshot(routeTestLocal(sessions...), incompatible), routeTestActive(true, "s00", 10))
260+ require.True(t, changed)
261+ require.NoError(t, snapshot.Validate())
262+ require.Len(t, snapshot.Entries, protocol.RouteSnapshotMaxEntries)
263+ require.Empty(t, snapshot.Hosts, "an incompatible daemon is never a creation host")
264+}
M internal/usecase/client/supervisor.go +5 -0
@@ -412,6 +412,11 @@ type Supervisor struct {
412412 pendingSwap *pickerAttachmentTarget
413413 // kills runs the picker's `x` operations off the run goroutine.
414414 kills pickerKills
415+ // routes is the client route ledger published to the serving daemon;
416+ // routesSent is the attachment that received its latest snapshot. Both are
417+ // only touched from the run goroutine.
418+ routes *routeLedger
419+ routesSent AttachmentToken
415420 }
416421
417422 // NewSupervisor validates the required dependencies and returns a supervisor
@@ -412,6 +412,11 @@ type Supervisor struct {
412 pendingSwap *pickerAttachmentTarget412 pendingSwap *pickerAttachmentTarget
413 // kills runs the picker's `x` operations off the run goroutine.413 // kills runs the picker's `x` operations off the run goroutine.
414 kills pickerKills414 kills pickerKills
415+ // routes is the client route ledger published to the serving daemon;
416+ // routesSent is the attachment that received its latest snapshot. Both are
417+ // only touched from the run goroutine.
418+ routes *routeLedger
419+ routesSent AttachmentToken
415 }420 }
416 421
417 // NewSupervisor validates the required dependencies and returns a supervisor422 // NewSupervisor validates the required dependencies and returns a supervisor
M internal/usecase/client/supervisor_attach.go +17 -2
@@ -436,6 +436,9 @@ func (s *Supervisor) runResolvedAttachment(ctx context.Context, input *terminalI
436436 s.invalidatePickerPresentation()
437437 // Back on the plain picker there is no attachment to start on.
438438 s.setPickerCurrent(pickerCurrent{})
439+ // Publications folded into routes while attached never reached the
440+ // picker catalogue.
441+ picker.ApplySnapshot(service.Snapshot())
439442 picker.SetOwnsInput(true)
440443 input.acquirePicker()
441444 }()
@@ -555,8 +558,17 @@ settlement:
555558 break settlement
556559 case <-s.attachments.NavigationRequests():
557560 overlay.enter()
558- case <-overlay.changed():
559- overlay.applyPublication()
561+ case <-s.brokerChanged():
562+ if overlay.active {
563+ overlay.applyPublication()
564+ }
565+ s.publishRoutes(service, run, request)
566+ case <-s.attachments.RouteDemands():
567+ s.publishRoutes(service, run, request)
568+ case message := <-s.attachments.Navigations():
569+ if !overlay.swapping {
570+ s.settleDaemonNavigation(service, overlay, message)
571+ }
560572 case <-overlay.previewChanged():
561573 overlay.publishPreview()
562574 case <-overlay.ops():
@@ -587,6 +599,9 @@ settlement:
587599
588600 // The attachment settled, so its foreground (and overlay slot) is gone.
589601 overlay.drop()
602+ if !terminated && !brokerLost {
603+ s.takeSettledNavigation(service, run.token, request)
604+ }
590605 if terminated {
591606 s.pendingSwap = nil
592607 return true, termErr
@@ -436,6 +436,9 @@ func (s *Supervisor) runResolvedAttachment(ctx context.Context, input *terminalI
436 s.invalidatePickerPresentation()436 s.invalidatePickerPresentation()
437 // Back on the plain picker there is no attachment to start on.437 // Back on the plain picker there is no attachment to start on.
438 s.setPickerCurrent(pickerCurrent{})438 s.setPickerCurrent(pickerCurrent{})
439+ // Publications folded into routes while attached never reached the
440+ // picker catalogue.
441+ picker.ApplySnapshot(service.Snapshot())
439 picker.SetOwnsInput(true)442 picker.SetOwnsInput(true)
440 input.acquirePicker()443 input.acquirePicker()
441 }()444 }()
@@ -555,8 +558,17 @@ settlement:
555 break settlement558 break settlement
556 case <-s.attachments.NavigationRequests():559 case <-s.attachments.NavigationRequests():
557 overlay.enter()560 overlay.enter()
558- case <-overlay.changed():561+ case <-s.brokerChanged():
559- overlay.applyPublication()562+ if overlay.active {
563+ overlay.applyPublication()
564+ }
565+ s.publishRoutes(service, run, request)
566+ case <-s.attachments.RouteDemands():
567+ s.publishRoutes(service, run, request)
568+ case message := <-s.attachments.Navigations():
569+ if !overlay.swapping {
570+ s.settleDaemonNavigation(service, overlay, message)
571+ }
560 case <-overlay.previewChanged():572 case <-overlay.previewChanged():
561 overlay.publishPreview()573 overlay.publishPreview()
562 case <-overlay.ops():574 case <-overlay.ops():
@@ -587,6 +599,9 @@ settlement:
587 599
588 // The attachment settled, so its foreground (and overlay slot) is gone.600 // The attachment settled, so its foreground (and overlay slot) is gone.
589 overlay.drop()601 overlay.drop()
602+ if !terminated && !brokerLost {
603+ s.takeSettledNavigation(service, run.token, request)
604+ }
590 if terminated {605 if terminated {
591 s.pendingSwap = nil606 s.pendingSwap = nil
592 return true, termErr607 return true, termErr
M internal/usecase/client/supervisor_overlay.go +0 -8
@@ -39,14 +39,6 @@ type attachmentPickerOverlay struct {
3939
4040 func (o *attachmentPickerOverlay) picker() pickerHost { return o.sup.cfg.Picker }
4141
42-// changed is the broker publication wake while the overlay is presented.
43-func (o *attachmentPickerOverlay) changed() <-chan struct{} {
44- if !o.active || supervisorNil(o.sup.readySub) {
45- return nil
46- }
47- return o.sup.readySub.Changed()
48-}
49-
5042 func (o *attachmentPickerOverlay) previewChanged() <-chan struct{} {
5143 if !o.active {
5244 return nil
@@ -39,14 +39,6 @@ type attachmentPickerOverlay struct {
39 39
40 func (o *attachmentPickerOverlay) picker() pickerHost { return o.sup.cfg.Picker }40 func (o *attachmentPickerOverlay) picker() pickerHost { return o.sup.cfg.Picker }
41 41
42-// changed is the broker publication wake while the overlay is presented.
43-func (o *attachmentPickerOverlay) changed() <-chan struct{} {
44- if !o.active || supervisorNil(o.sup.readySub) {
45- return nil
46- }
47- return o.sup.readySub.Changed()
48-}
49-
50 func (o *attachmentPickerOverlay) previewChanged() <-chan struct{} {42 func (o *attachmentPickerOverlay) previewChanged() <-chan struct{} {
51 if !o.active {43 if !o.active {
52 return nil44 return nil
A internal/usecase/client/supervisor_routes.go +216 -0
@@ -0,0 +1,216 @@
1+package client
2+
3+import (
4+ "errors"
5+
6+ "github.com/bnema/vev/internal/ports"
7+ "github.com/bnema/vev/internal/protocol"
8+)
9+
10+// Routes and daemon navigation while attached (Plan 003 C4, C5, E3).
11+//
12+// The supervisor publishes the client route ledger to the serving daemon after
13+// the committed initial publication, whenever the attached session changes,
14+// and on every broker publication. The daemon renders the status-bar MRU,
15+// other-session bells, and the palette's remote, recent-route, and
16+// create-on-host rows from it; it never monitors another daemon.
17+//
18+// The daemon never dials either. Its navigation requests reach the
19+// supervisor, which settles them exactly like a picker commit: another tab of
20+// the attached session switches in place with SelectTab, anything else detaches
21+// cleanly and swaps through Connecting.
22+//
23+// - RouteNavigationAction and RouteCreateSessionAction name a reference of
24+// the published ledger; the supervisor resolves it against the latest
25+// broker catalogue and answers a refusal with the typed failure the daemon
26+// turns into a toast.
27+// - An endpoint-empty AttachTarget names a lifecycle on the serving daemon
28+// itself (a stopped session or a close-and-dial handoff); the same-peer
29+// offer is confirmed by the worker in place.
30+//
31+// Every method runs on the supervisor's run goroutine.
32+
33+// brokerChanged is the adopted connection's publication wake, or nil.
34+func (s *Supervisor) brokerChanged() <-chan struct{} {
35+ if supervisorNil(s.readySub) {
36+ return nil
37+ }
38+ return s.readySub.Changed()
39+}
40+
41+// publishRoutes rebuilds the ledger over the live attachment and hands a
42+// changed snapshot, or the first one of this attachment, to its worker.
43+func (s *Supervisor) publishRoutes(service ports.BrokerService, run *attachmentRun, request ports.BrokerOpenStreamRequest) {
44+ target, _, known := s.attachments.committedView()
45+ if !known {
46+ return
47+ }
48+ if s.routes == nil {
49+ s.routes = newRouteLedger()
50+ }
51+ snapshot, changed := s.routes.build(service.Snapshot(), routeActive{known: true, authority: requestAuthority(request), target: target})
52+ if snapshot.Generation == 0 || (!changed && s.routesSent == run.token) {
53+ return
54+ }
55+ if s.attachments.publishRoutes(run.token, snapshot) {
56+ s.routesSent = run.token
57+ }
58+}
59+
60+func requestAuthority(request ports.BrokerOpenStreamRequest) routeAuthority {
61+ if request.Local {
62+ return routeAuthority{local: true}
63+ }
64+ return routeAuthority{endpoint: request.Endpoint}
65+}
66+
67+// settleDaemonNavigation carries out one daemon navigation request over the
68+// live attachment.
69+func (s *Supervisor) settleDaemonNavigation(service ports.BrokerService, overlay *attachmentPickerOverlay, message protocol.ServerMessage) {
70+ target, ok := s.resolveDaemonNavigation(service, overlay.run.token, overlay.request, message)
71+ if !ok {
72+ return
73+ }
74+ if sameAttachmentTarget(overlay.request, s.attachments.committedTargetOrZero(), target.request) && target.tab.stopped == nil {
75+ _, current, _ := s.attachments.committedView()
76+ if target.tab.preferred != "" && target.tab.preferred != current {
77+ s.attachments.requestTabSelection(overlay.run.token, target.tab.preferred)
78+ }
79+ return
80+ }
81+ if overlay.active {
82+ overlay.exit()
83+ }
84+ s.pendingSwap = &target
85+ overlay.swapping = true
86+ s.attachments.requestDetach(overlay.run.token)
87+}
88+
89+// takeSettledNavigation adopts a navigation request the worker handed over
90+// just before its attachment ended, such as a close-and-dial handoff whose
91+// daemon detached the source right after sending it.
92+func (s *Supervisor) takeSettledNavigation(service ports.BrokerService, token AttachmentToken, request ports.BrokerOpenStreamRequest) {
93+ message, ok := s.attachments.takeNavigation()
94+ if !ok || s.pendingSwap != nil {
95+ return
96+ }
97+ if target, ok := s.resolveDaemonNavigation(service, token, request, message); ok {
98+ s.pendingSwap = &target
99+ }
100+}
101+
102+// resolveDaemonNavigation turns one daemon request into an exact attachment
103+// target. A refusal is answered on the attachment with the typed failure.
104+func (s *Supervisor) resolveDaemonNavigation(service ports.BrokerService, token AttachmentToken, request ports.BrokerOpenStreamRequest, message protocol.ServerMessage) (pickerAttachmentTarget, bool) {
105+ switch typed := message.(type) {
106+ case protocol.RouteNavigationAction:
107+ fail := func(code protocol.RouteFailureCode) (pickerAttachmentTarget, bool) {
108+ s.attachments.replyNavigation(token, protocol.RouteNavigationFailure{Key: typed.Key, Generation: typed.Generation, Code: code})
109+ return pickerAttachmentTarget{}, false
110+ }
111+ routed, ok := s.resolveRoute(protocol.RouteRef{Key: typed.Key, Generation: typed.Generation})
112+ if !ok || routed.host {
113+ return fail(protocol.RouteFailureStaleSelection)
114+ }
115+ target, err := s.resolveRouteSelection(service, pickerSelectionRef{
116+ kind: pickerSelectionExact, local: routed.authority.local, endpoint: routed.authority.endpoint,
117+ lifecycle: routed.target.LifecycleID, name: routed.target.SessionName,
118+ })
119+ if err != nil {
120+ return fail(routeNavigationFailureCode(err))
121+ }
122+ return target, true
123+ case protocol.RouteCreateSessionAction:
124+ fail := func(code protocol.RouteFailureCode) (pickerAttachmentTarget, bool) {
125+ s.attachments.replyNavigation(token, protocol.SessionCreationFailure{RequestID: typed.RequestID, Code: code})
126+ return pickerAttachmentTarget{}, false
127+ }
128+ routed, ok := s.resolveRoute(protocol.RouteRef{Key: typed.Key, Generation: typed.Generation})
129+ if !ok {
130+ return fail(protocol.RouteFailureStaleSelection)
131+ }
132+ target, err := s.resolveRouteSelection(service, pickerSelectionRef{
133+ kind: pickerSelectionCreateNamed, local: routed.authority.local, endpoint: routed.authority.endpoint,
134+ createName: typed.SessionName,
135+ })
136+ if err != nil {
137+ return fail(routeNavigationFailureCode(err))
138+ }
139+ return target, true
140+ case protocol.AttachTarget:
141+ return s.resolveServingHandoff(service, request, typed)
142+ default:
143+ return pickerAttachmentTarget{}, false
144+ }
145+}
146+
147+// resolveServingHandoff resolves an endpoint-empty handoff: the serving
148+// daemon itself named one of its own lifecycles, so the attachment's own
149+// authority is kept and only the exact target and tab change. The daemon
150+// revalidates the exact identity on Hello.
151+func (s *Supervisor) resolveServingHandoff(service ports.BrokerService, request ports.BrokerOpenStreamRequest, handoff protocol.AttachTarget) (pickerAttachmentTarget, bool) {
152+ if handoff.Endpoint != "" || handoff.SessionTarget != nil || handoff.ExactTarget == nil || handoff.Intent != protocol.IntentAttach {
153+ s.offerPickerNotice("daemon-handoff", "the daemon asked for a destination this client cannot open")
154+ return pickerAttachmentTarget{}, false
155+ }
156+ stream, err := service.NextStreamID()
157+ if err != nil {
158+ return pickerAttachmentTarget{}, false
159+ }
160+ next := request
161+ next.Stream = stream
162+ next.Connection = service.ConnectionID()
163+ next.Admission = ports.BrokerAdmissionExact
164+ next.Target = *handoff.ExactTarget
165+ next.Name = ""
166+ next.Purpose = ports.BrokerStreamAttachment
167+ next.StartMode = ports.BrokerDaemonStartIfNeeded
168+ if next.Validate() != nil {
169+ return pickerAttachmentTarget{}, false
170+ }
171+ return pickerAttachmentTarget{request: next, tab: attachmentTab{preferred: handoff.PreferredTabID}}, true
172+}
173+
174+func (s *Supervisor) resolveRoute(ref protocol.RouteRef) (routeLedgerTarget, bool) {
175+ if s.routes == nil || ref.IsZero() {
176+ return routeLedgerTarget{}, false
177+ }
178+ return s.routes.resolve(ref)
179+}
180+
181+// resolveRouteSelection resolves one route through the same catalogue rules
182+// as a picker commit, against the latest broker publication.
183+func (s *Supervisor) resolveRouteSelection(service ports.BrokerService, selection pickerSelectionRef) (pickerAttachmentTarget, error) {
184+ stream, err := service.NextStreamID()
185+ if err != nil {
186+ return pickerAttachmentTarget{}, err
187+ }
188+ snapshot := service.Snapshot()
189+ authority := pickerAuthorityInSnapshot(snapshot, selection)
190+ if authority.found {
191+ // The ledger keys a host by endpoint; the registration is the
192+ // catalogue's current one.
193+ selection.registration = authority.observation.Registration
194+ }
195+ base := pickerResolveBase{Connection: service.ConnectionID(), Stream: stream}
196+ request, tab, err := resolvePickerTarget(snapshot.Epoch, authority, selection, base, true)
197+ if err != nil {
198+ return pickerAttachmentTarget{}, err
199+ }
200+ return pickerAttachmentTarget{request: request, tab: tab}, nil
201+}
202+
203+func routeNavigationFailureCode(err error) protocol.RouteFailureCode {
204+ var catalogueErr pickerCatalogueError
205+ if errors.As(err, &catalogueErr) {
206+ switch catalogueErr.Code {
207+ case pickerCatalogueGone:
208+ return protocol.RouteFailureNoSuchRoute
209+ case pickerCatalogueReplaced:
210+ return protocol.RouteFailureTargetChanged
211+ case pickerCatalogueEpochStale:
212+ return protocol.RouteFailureStaleSelection
213+ }
214+ }
215+ return protocol.RouteFailureUnavailable
216+}
@@ -0,0 +1,216 @@
1+package client
2+
3+import (
4+ "errors"
5+
6+ "github.com/bnema/vev/internal/ports"
7+ "github.com/bnema/vev/internal/protocol"
8+)
9+
10+// Routes and daemon navigation while attached (Plan 003 C4, C5, E3).
11+//
12+// The supervisor publishes the client route ledger to the serving daemon after
13+// the committed initial publication, whenever the attached session changes,
14+// and on every broker publication. The daemon renders the status-bar MRU,
15+// other-session bells, and the palette's remote, recent-route, and
16+// create-on-host rows from it; it never monitors another daemon.
17+//
18+// The daemon never dials either. Its navigation requests reach the
19+// supervisor, which settles them exactly like a picker commit: another tab of
20+// the attached session switches in place with SelectTab, anything else detaches
21+// cleanly and swaps through Connecting.
22+//
23+// - RouteNavigationAction and RouteCreateSessionAction name a reference of
24+// the published ledger; the supervisor resolves it against the latest
25+// broker catalogue and answers a refusal with the typed failure the daemon
26+// turns into a toast.
27+// - An endpoint-empty AttachTarget names a lifecycle on the serving daemon
28+// itself (a stopped session or a close-and-dial handoff); the same-peer
29+// offer is confirmed by the worker in place.
30+//
31+// Every method runs on the supervisor's run goroutine.
32+
33+// brokerChanged is the adopted connection's publication wake, or nil.
34+func (s *Supervisor) brokerChanged() <-chan struct{} {
35+ if supervisorNil(s.readySub) {
36+ return nil
37+ }
38+ return s.readySub.Changed()
39+}
40+
41+// publishRoutes rebuilds the ledger over the live attachment and hands a
42+// changed snapshot, or the first one of this attachment, to its worker.
43+func (s *Supervisor) publishRoutes(service ports.BrokerService, run *attachmentRun, request ports.BrokerOpenStreamRequest) {
44+ target, _, known := s.attachments.committedView()
45+ if !known {
46+ return
47+ }
48+ if s.routes == nil {
49+ s.routes = newRouteLedger()
50+ }
51+ snapshot, changed := s.routes.build(service.Snapshot(), routeActive{known: true, authority: requestAuthority(request), target: target})
52+ if snapshot.Generation == 0 || (!changed && s.routesSent == run.token) {
53+ return
54+ }
55+ if s.attachments.publishRoutes(run.token, snapshot) {
56+ s.routesSent = run.token
57+ }
58+}
59+
60+func requestAuthority(request ports.BrokerOpenStreamRequest) routeAuthority {
61+ if request.Local {
62+ return routeAuthority{local: true}
63+ }
64+ return routeAuthority{endpoint: request.Endpoint}
65+}
66+
67+// settleDaemonNavigation carries out one daemon navigation request over the
68+// live attachment.
69+func (s *Supervisor) settleDaemonNavigation(service ports.BrokerService, overlay *attachmentPickerOverlay, message protocol.ServerMessage) {
70+ target, ok := s.resolveDaemonNavigation(service, overlay.run.token, overlay.request, message)
71+ if !ok {
72+ return
73+ }
74+ if sameAttachmentTarget(overlay.request, s.attachments.committedTargetOrZero(), target.request) && target.tab.stopped == nil {
75+ _, current, _ := s.attachments.committedView()
76+ if target.tab.preferred != "" && target.tab.preferred != current {
77+ s.attachments.requestTabSelection(overlay.run.token, target.tab.preferred)
78+ }
79+ return
80+ }
81+ if overlay.active {
82+ overlay.exit()
83+ }
84+ s.pendingSwap = &target
85+ overlay.swapping = true
86+ s.attachments.requestDetach(overlay.run.token)
87+}
88+
89+// takeSettledNavigation adopts a navigation request the worker handed over
90+// just before its attachment ended, such as a close-and-dial handoff whose
91+// daemon detached the source right after sending it.
92+func (s *Supervisor) takeSettledNavigation(service ports.BrokerService, token AttachmentToken, request ports.BrokerOpenStreamRequest) {
93+ message, ok := s.attachments.takeNavigation()
94+ if !ok || s.pendingSwap != nil {
95+ return
96+ }
97+ if target, ok := s.resolveDaemonNavigation(service, token, request, message); ok {
98+ s.pendingSwap = &target
99+ }
100+}
101+
102+// resolveDaemonNavigation turns one daemon request into an exact attachment
103+// target. A refusal is answered on the attachment with the typed failure.
104+func (s *Supervisor) resolveDaemonNavigation(service ports.BrokerService, token AttachmentToken, request ports.BrokerOpenStreamRequest, message protocol.ServerMessage) (pickerAttachmentTarget, bool) {
105+ switch typed := message.(type) {
106+ case protocol.RouteNavigationAction:
107+ fail := func(code protocol.RouteFailureCode) (pickerAttachmentTarget, bool) {
108+ s.attachments.replyNavigation(token, protocol.RouteNavigationFailure{Key: typed.Key, Generation: typed.Generation, Code: code})
109+ return pickerAttachmentTarget{}, false
110+ }
111+ routed, ok := s.resolveRoute(protocol.RouteRef{Key: typed.Key, Generation: typed.Generation})
112+ if !ok || routed.host {
113+ return fail(protocol.RouteFailureStaleSelection)
114+ }
115+ target, err := s.resolveRouteSelection(service, pickerSelectionRef{
116+ kind: pickerSelectionExact, local: routed.authority.local, endpoint: routed.authority.endpoint,
117+ lifecycle: routed.target.LifecycleID, name: routed.target.SessionName,
118+ })
119+ if err != nil {
120+ return fail(routeNavigationFailureCode(err))
121+ }
122+ return target, true
123+ case protocol.RouteCreateSessionAction:
124+ fail := func(code protocol.RouteFailureCode) (pickerAttachmentTarget, bool) {
125+ s.attachments.replyNavigation(token, protocol.SessionCreationFailure{RequestID: typed.RequestID, Code: code})
126+ return pickerAttachmentTarget{}, false
127+ }
128+ routed, ok := s.resolveRoute(protocol.RouteRef{Key: typed.Key, Generation: typed.Generation})
129+ if !ok {
130+ return fail(protocol.RouteFailureStaleSelection)
131+ }
132+ target, err := s.resolveRouteSelection(service, pickerSelectionRef{
133+ kind: pickerSelectionCreateNamed, local: routed.authority.local, endpoint: routed.authority.endpoint,
134+ createName: typed.SessionName,
135+ })
136+ if err != nil {
137+ return fail(routeNavigationFailureCode(err))
138+ }
139+ return target, true
140+ case protocol.AttachTarget:
141+ return s.resolveServingHandoff(service, request, typed)
142+ default:
143+ return pickerAttachmentTarget{}, false
144+ }
145+}
146+
147+// resolveServingHandoff resolves an endpoint-empty handoff: the serving
148+// daemon itself named one of its own lifecycles, so the attachment's own
149+// authority is kept and only the exact target and tab change. The daemon
150+// revalidates the exact identity on Hello.
151+func (s *Supervisor) resolveServingHandoff(service ports.BrokerService, request ports.BrokerOpenStreamRequest, handoff protocol.AttachTarget) (pickerAttachmentTarget, bool) {
152+ if handoff.Endpoint != "" || handoff.SessionTarget != nil || handoff.ExactTarget == nil || handoff.Intent != protocol.IntentAttach {
153+ s.offerPickerNotice("daemon-handoff", "the daemon asked for a destination this client cannot open")
154+ return pickerAttachmentTarget{}, false
155+ }
156+ stream, err := service.NextStreamID()
157+ if err != nil {
158+ return pickerAttachmentTarget{}, false
159+ }
160+ next := request
161+ next.Stream = stream
162+ next.Connection = service.ConnectionID()
163+ next.Admission = ports.BrokerAdmissionExact
164+ next.Target = *handoff.ExactTarget
165+ next.Name = ""
166+ next.Purpose = ports.BrokerStreamAttachment
167+ next.StartMode = ports.BrokerDaemonStartIfNeeded
168+ if next.Validate() != nil {
169+ return pickerAttachmentTarget{}, false
170+ }
171+ return pickerAttachmentTarget{request: next, tab: attachmentTab{preferred: handoff.PreferredTabID}}, true
172+}
173+
174+func (s *Supervisor) resolveRoute(ref protocol.RouteRef) (routeLedgerTarget, bool) {
175+ if s.routes == nil || ref.IsZero() {
176+ return routeLedgerTarget{}, false
177+ }
178+ return s.routes.resolve(ref)
179+}
180+
181+// resolveRouteSelection resolves one route through the same catalogue rules
182+// as a picker commit, against the latest broker publication.
183+func (s *Supervisor) resolveRouteSelection(service ports.BrokerService, selection pickerSelectionRef) (pickerAttachmentTarget, error) {
184+ stream, err := service.NextStreamID()
185+ if err != nil {
186+ return pickerAttachmentTarget{}, err
187+ }
188+ snapshot := service.Snapshot()
189+ authority := pickerAuthorityInSnapshot(snapshot, selection)
190+ if authority.found {
191+ // The ledger keys a host by endpoint; the registration is the
192+ // catalogue's current one.
193+ selection.registration = authority.observation.Registration
194+ }
195+ base := pickerResolveBase{Connection: service.ConnectionID(), Stream: stream}
196+ request, tab, err := resolvePickerTarget(snapshot.Epoch, authority, selection, base, true)
197+ if err != nil {
198+ return pickerAttachmentTarget{}, err
199+ }
200+ return pickerAttachmentTarget{request: request, tab: tab}, nil
201+}
202+
203+func routeNavigationFailureCode(err error) protocol.RouteFailureCode {
204+ var catalogueErr pickerCatalogueError
205+ if errors.As(err, &catalogueErr) {
206+ switch catalogueErr.Code {
207+ case pickerCatalogueGone:
208+ return protocol.RouteFailureNoSuchRoute
209+ case pickerCatalogueReplaced:
210+ return protocol.RouteFailureTargetChanged
211+ case pickerCatalogueEpochStale:
212+ return protocol.RouteFailureStaleSelection
213+ }
214+ }
215+ return protocol.RouteFailureUnavailable
216+}
A internal/usecase/client/supervisor_routes_test.go +264 -0
@@ -0,0 +1,264 @@
1+package client
2+
3+import (
4+ "context"
5+ "sync"
6+ "testing"
7+ "time"
8+
9+ "github.com/stretchr/testify/require"
10+
11+ "github.com/bnema/vev/internal/domain"
12+ "github.com/bnema/vev/internal/ports"
13+ "github.com/bnema/vev/internal/protocol"
14+ "github.com/bnema/vev/internal/protocol/catalogue"
15+)
16+
17+// Supervisor halves of Plan 003 C4/C5/E3: the route snapshot published to the
18+// serving daemon while attached, and the daemon's navigation requests settled
19+// through the broker like a picker commit. They drive the real supervisor,
20+// host, foreground, and worker with bounded waits and no sleeps.
21+
22+// routeTestCatalogue is a local daemon serving the attached "alpha" plus one
23+// remote daemon with "rem", whose bell is remoteBell.
24+func routeTestCatalogue(revision ports.BrokerRevision, remoteBell bool) ports.BrokerSnapshot {
25+ snapshot := routeTestSnapshot(
26+ routeTestLocal(routeTestLive("alpha", 1, 2), routeTestLive("beta", 2, 1)),
27+ routeTestRemoteHost(routeTestLive("rem", 3, 0, remoteBell)),
28+ )
29+ snapshot.Revision = revision
30+ return snapshot
31+}
32+
33+// sentRouteSnapshots returns every route snapshot the attachment published.
34+func sentRouteSnapshots(stream *sessionTestStream) []protocol.RecentRouteSnapshot {
35+ var out []protocol.RecentRouteSnapshot
36+ for _, message := range stream.messages() {
37+ if snapshot, ok := message.(protocol.RecentRouteSnapshot); ok {
38+ out = append(out, snapshot)
39+ }
40+ }
41+ return out
42+}
43+
44+func awaitRouteSnapshot(t *testing.T, stream *sessionTestStream, what string, match func(protocol.RecentRouteSnapshot) bool) protocol.RecentRouteSnapshot {
45+ t.Helper()
46+ var found protocol.RecentRouteSnapshot
47+ require.Eventually(t, func() bool {
48+ for _, snapshot := range sentRouteSnapshots(stream) {
49+ if match(snapshot) {
50+ found = snapshot
51+ return true
52+ }
53+ }
54+ return false
55+ }, 5*time.Second, time.Millisecond, "client never published %s", what)
56+ return found
57+}
58+
59+func routeEntryNamed(snapshot protocol.RecentRouteSnapshot, name string) (protocol.RecentRouteEntry, bool) {
60+ for _, entry := range snapshot.Entries {
61+ if entry.Name == name {
62+ return entry, true
63+ }
64+ }
65+ return protocol.RecentRouteEntry{}, false
66+}
67+
68+// TestSupervisorPublishesRoutesWhileAttached ports main's local/remote
69+// attention publication: the first snapshot follows the committed initial
70+// publication, and a broker publication while attached republishes it with
71+// the other daemon's bell.
72+func TestSupervisorPublishesRoutesWhileAttached(t *testing.T) {
73+ picker := newAttachTestPicker()
74+ harness := startAttachHarness(t, picker)
75+ harness.service.publishSnapshot(routeTestCatalogue(2, false))
76+ stream := attachLiveSession(t, harness, picker)
77+
78+ first := awaitRouteSnapshot(t, stream, "the first route snapshot", func(s protocol.RecentRouteSnapshot) bool {
79+ _, ok := routeEntryNamed(s, "rem")
80+ return ok
81+ })
82+ require.NoError(t, first.Validate())
83+ require.Equal(t, "alpha", first.ActiveEntry.Name, "the attached lifecycle is metadata only")
84+ _, listed := routeEntryNamed(first, "alpha")
85+ require.False(t, listed)
86+ remote, _ := routeEntryNamed(first, "rem")
87+ require.False(t, remote.Attention)
88+ require.Equal(t, protocol.RouteKindRemote, remote.Kind)
89+
90+ harness.service.publishSnapshot(routeTestCatalogue(3, true))
91+ rung := awaitRouteSnapshot(t, stream, "the remote bell", func(s protocol.RecentRouteSnapshot) bool {
92+ entry, ok := routeEntryNamed(s, "rem")
93+ return ok && entry.Attention
94+ })
95+ require.Greater(t, rung.Generation, first.Generation)
96+ require.Equal(t, first.Active, rung.Active, "the attachment keeps its reference")
97+
98+ harness.service.publishSnapshot(routeTestCatalogue(4, true))
99+ require.Never(t, func() bool { return len(sentRouteSnapshots(stream)) > 2 }, 50*time.Millisecond, time.Millisecond,
100+ "an unchanged catalogue publishes nothing")
101+ require.Len(t, harness.service.openedRequests(), 1)
102+}
103+
104+// TestSupervisorSettlesDaemonNavigation pins C5 and the E3 palette commits:
105+// the daemon's requests resolve through the broker catalogue and open through
106+// the broker exactly like a picker commit; a refusal is answered typed.
107+func TestSupervisorSettlesDaemonNavigation(t *testing.T) {
108+ beta := protocol.ExactSessionTarget{LifecycleID: pickerTestLifecycle(2), SessionName: "beta"}
109+ tests := []struct {
110+ name string
111+ // request builds the daemon message from the first route snapshot.
112+ request func(protocol.RecentRouteSnapshot) protocol.ServerMessage
113+ detached bool
114+ wantOpen func(*testing.T, ports.BrokerOpenStreamRequest)
115+ wantReply func(*testing.T, *sessionTestStream)
116+ }{
117+ {
118+ name: "recent route to another daemon swaps",
119+ request: func(s protocol.RecentRouteSnapshot) protocol.ServerMessage {
120+ entry, _ := routeEntryNamed(s, "rem")
121+ return protocol.RouteNavigationAction{SnapshotGeneration: s.Generation, Key: entry.Key, Generation: entry.Generation}
122+ },
123+ wantOpen: func(t *testing.T, request ports.BrokerOpenStreamRequest) {
124+ require.False(t, request.Local)
125+ require.Equal(t, routeTestRemote, request.Endpoint)
126+ require.Equal(t, ports.BrokerAdmissionExact, request.Admission)
127+ require.Equal(t, "rem", request.Target.SessionName)
128+ },
129+ },
130+ {
131+ name: "a stale reference is refused typed",
132+ request: func(s protocol.RecentRouteSnapshot) protocol.ServerMessage {
133+ return protocol.RouteNavigationAction{SnapshotGeneration: s.Generation, Key: 999, Generation: 1}
134+ },
135+ wantReply: func(t *testing.T, stream *sessionTestStream) {
136+ failure := awaitSent(t, stream, "RouteNavigationFailure", isSent[protocol.RouteNavigationFailure]).(protocol.RouteNavigationFailure)
137+ require.Equal(t, protocol.RouteNavigationFailure{Key: 999, Generation: 1, Code: protocol.RouteFailureStaleSelection}, failure)
138+ },
139+ },
140+ {
141+ name: "create on another host",
142+ request: func(s protocol.RecentRouteSnapshot) protocol.ServerMessage {
143+ host := s.Hosts[0]
144+ return protocol.RouteCreateSessionAction{RequestID: 4, SnapshotGeneration: s.Generation, Key: host.Key, Generation: host.Generation, SessionName: "fresh"}
145+ },
146+ wantOpen: func(t *testing.T, request ports.BrokerOpenStreamRequest) {
147+ require.Equal(t, routeTestRemote, request.Endpoint)
148+ require.Equal(t, ports.BrokerAdmissionCreateNamed, request.Admission)
149+ require.Equal(t, "fresh", request.Name)
150+ },
151+ },
152+ {
153+ name: "create with a stale host is refused typed",
154+ request: func(s protocol.RecentRouteSnapshot) protocol.ServerMessage {
155+ return protocol.RouteCreateSessionAction{RequestID: 4, SnapshotGeneration: s.Generation, Key: 999, Generation: 1, SessionName: "fresh"}
156+ },
157+ wantReply: func(t *testing.T, stream *sessionTestStream) {
158+ failure := awaitSent(t, stream, "SessionCreationFailure", isSent[protocol.SessionCreationFailure]).(protocol.SessionCreationFailure)
159+ require.Equal(t, protocol.SessionCreationFailure{RequestID: 4, Code: protocol.RouteFailureStaleSelection}, failure)
160+ },
161+ },
162+ {
163+ name: "a same-peer offer is confirmed in place",
164+ request: func(protocol.RecentRouteSnapshot) protocol.ServerMessage {
165+ target := beta
166+ return protocol.AttachTarget{Session: "beta", Intent: protocol.IntentAttach, ExactTarget: &target, EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, SamePeer: true}
167+ },
168+ wantReply: func(t *testing.T, stream *sessionTestStream) {
169+ request := awaitSent(t, stream, "SamePeerSwitchRequest", isSent[protocol.SamePeerSwitchRequest]).(protocol.SamePeerSwitchRequest)
170+ require.Equal(t, beta, request.Target)
171+ require.NotZero(t, request.RequestID)
172+ },
173+ },
174+ {
175+ name: "a close-and-dial handoff reopens on the serving daemon",
176+ request: func(protocol.RecentRouteSnapshot) protocol.ServerMessage {
177+ target := beta
178+ return protocol.AttachTarget{Session: "beta", Intent: protocol.IntentAttach, ExactTarget: &target, EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, PreferredTabID: "t_x"}
179+ },
180+ detached: true,
181+ wantOpen: func(t *testing.T, request ports.BrokerOpenStreamRequest) {
182+ require.True(t, request.Local)
183+ require.Equal(t, ports.BrokerAdmissionExact, request.Admission)
184+ require.Equal(t, beta, request.Target)
185+ },
186+ },
187+ }
188+ for _, tt := range tests {
189+ t.Run(tt.name, func(t *testing.T) {
190+ picker := newAttachTestPicker()
191+ harness := startAttachHarness(t, picker)
192+ harness.service.publishSnapshot(routeTestCatalogue(2, false))
193+ stream := attachLiveSession(t, harness, picker)
194+ snapshot := awaitRouteSnapshot(t, stream, "the route snapshot", func(s protocol.RecentRouteSnapshot) bool {
195+ _, ok := routeEntryNamed(s, "rem")
196+ return ok && len(s.Hosts) == 1
197+ })
198+
199+ var mu sync.Mutex
200+ var second *sessionTestStream
201+ harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
202+ mu.Lock()
203+ defer mu.Unlock()
204+ second = newSessionTestStream()
205+ return second, nil
206+ })
207+ stream.deliver(tt.request(snapshot))
208+ if tt.detached {
209+ // The daemon ends the source right after a close-and-dial
210+ // handoff; the supervisor still adopts the handoff.
211+ stream.deliver(protocol.Detached{Reason: protocol.ReasonDetach})
212+ }
213+
214+ if tt.wantOpen != nil {
215+ require.Eventually(t, func() bool { return len(harness.service.openedRequests()) == 2 }, 5*time.Second, time.Millisecond, "the navigation never opened a stream")
216+ opened := harness.service.openedRequests()
217+ tt.wantOpen(t, opened[1])
218+ awaitStreamHello(t, &mu, &second)
219+ if tt.detached {
220+ mu.Lock()
221+ hello := second.messages()[0].(protocol.Hello)
222+ mu.Unlock()
223+ require.Equal(t, domain.TabStableID("t_x"), hello.PreferredTabID)
224+ } else {
225+ awaitSent(t, stream, "Detach", isSent[protocol.Detach])
226+ }
227+ return
228+ }
229+ tt.wantReply(t, stream)
230+ require.Never(t, func() bool { return len(harness.service.openedRequests()) > 1 }, 50*time.Millisecond, time.Millisecond, "a refusal or an in-place switch never reconnects")
231+ require.Zero(t, countSent[protocol.Detach](stream))
232+ })
233+ }
234+}
235+
236+// TestSupervisorRouteSnapshotExcludesServingHost pins E3's destinations: a
237+// remote attachment offers the local daemon as a creation host and never
238+// itself.
239+func TestSupervisorRouteSnapshotExcludesServingHost(t *testing.T) {
240+ picker := newAttachTestPicker()
241+ harness := startAttachHarness(t, picker)
242+ remote := routeTestRemoteHost(catalogue.RemoteCatalogSession{LifecycleID: domain.SessionLifecycleID{1}, Name: "alpha", State: catalogue.RemoteCatalogSessionUp})
243+ remote.Registration = sessionTestRequest(false).Registration
244+ remote.Endpoint = sessionTestRequest(false).Endpoint
245+ remote.DisplayOrigin = "example"
246+ snapshot := routeTestSnapshot(routeTestLocal(routeTestLive("loc", 4, 1)), remote)
247+ snapshot.Revision = 2
248+ harness.service.publishSnapshot(snapshot)
249+
250+ stream := newSessionTestStream()
251+ harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
252+ return stream, nil
253+ })
254+ picker.commit(sessionTestRequest(false))
255+ awaitStreamHello(t, &sync.Mutex{}, &stream)
256+ deliverReadyStream(t, stream)
257+ awaitAttachedState(t, harness.sup)
258+
259+ published := awaitRouteSnapshot(t, stream, "the route snapshot", func(s protocol.RecentRouteSnapshot) bool { return len(s.Hosts) > 0 })
260+ require.Equal(t, []protocol.RouteHost{{Key: published.Hosts[0].Key, Generation: 1, Label: "local", Kind: protocol.RouteKindLocal}}, published.Hosts)
261+ require.Equal(t, protocol.RouteKindRemote, published.ActiveEntry.Kind)
262+ require.Equal(t, "example", published.ActiveEntry.HostLabel)
263+ require.True(t, published.Home.IsZero(), "only a local attachment is home")
264+}
@@ -0,0 +1,264 @@
1+package client
2+
3+import (
4+ "context"
5+ "sync"
6+ "testing"
7+ "time"
8+
9+ "github.com/stretchr/testify/require"
10+
11+ "github.com/bnema/vev/internal/domain"
12+ "github.com/bnema/vev/internal/ports"
13+ "github.com/bnema/vev/internal/protocol"
14+ "github.com/bnema/vev/internal/protocol/catalogue"
15+)
16+
17+// Supervisor halves of Plan 003 C4/C5/E3: the route snapshot published to the
18+// serving daemon while attached, and the daemon's navigation requests settled
19+// through the broker like a picker commit. They drive the real supervisor,
20+// host, foreground, and worker with bounded waits and no sleeps.
21+
22+// routeTestCatalogue is a local daemon serving the attached "alpha" plus one
23+// remote daemon with "rem", whose bell is remoteBell.
24+func routeTestCatalogue(revision ports.BrokerRevision, remoteBell bool) ports.BrokerSnapshot {
25+ snapshot := routeTestSnapshot(
26+ routeTestLocal(routeTestLive("alpha", 1, 2), routeTestLive("beta", 2, 1)),
27+ routeTestRemoteHost(routeTestLive("rem", 3, 0, remoteBell)),
28+ )
29+ snapshot.Revision = revision
30+ return snapshot
31+}
32+
33+// sentRouteSnapshots returns every route snapshot the attachment published.
34+func sentRouteSnapshots(stream *sessionTestStream) []protocol.RecentRouteSnapshot {
35+ var out []protocol.RecentRouteSnapshot
36+ for _, message := range stream.messages() {
37+ if snapshot, ok := message.(protocol.RecentRouteSnapshot); ok {
38+ out = append(out, snapshot)
39+ }
40+ }
41+ return out
42+}
43+
44+func awaitRouteSnapshot(t *testing.T, stream *sessionTestStream, what string, match func(protocol.RecentRouteSnapshot) bool) protocol.RecentRouteSnapshot {
45+ t.Helper()
46+ var found protocol.RecentRouteSnapshot
47+ require.Eventually(t, func() bool {
48+ for _, snapshot := range sentRouteSnapshots(stream) {
49+ if match(snapshot) {
50+ found = snapshot
51+ return true
52+ }
53+ }
54+ return false
55+ }, 5*time.Second, time.Millisecond, "client never published %s", what)
56+ return found
57+}
58+
59+func routeEntryNamed(snapshot protocol.RecentRouteSnapshot, name string) (protocol.RecentRouteEntry, bool) {
60+ for _, entry := range snapshot.Entries {
61+ if entry.Name == name {
62+ return entry, true
63+ }
64+ }
65+ return protocol.RecentRouteEntry{}, false
66+}
67+
68+// TestSupervisorPublishesRoutesWhileAttached ports main's local/remote
69+// attention publication: the first snapshot follows the committed initial
70+// publication, and a broker publication while attached republishes it with
71+// the other daemon's bell.
72+func TestSupervisorPublishesRoutesWhileAttached(t *testing.T) {
73+ picker := newAttachTestPicker()
74+ harness := startAttachHarness(t, picker)
75+ harness.service.publishSnapshot(routeTestCatalogue(2, false))
76+ stream := attachLiveSession(t, harness, picker)
77+
78+ first := awaitRouteSnapshot(t, stream, "the first route snapshot", func(s protocol.RecentRouteSnapshot) bool {
79+ _, ok := routeEntryNamed(s, "rem")
80+ return ok
81+ })
82+ require.NoError(t, first.Validate())
83+ require.Equal(t, "alpha", first.ActiveEntry.Name, "the attached lifecycle is metadata only")
84+ _, listed := routeEntryNamed(first, "alpha")
85+ require.False(t, listed)
86+ remote, _ := routeEntryNamed(first, "rem")
87+ require.False(t, remote.Attention)
88+ require.Equal(t, protocol.RouteKindRemote, remote.Kind)
89+
90+ harness.service.publishSnapshot(routeTestCatalogue(3, true))
91+ rung := awaitRouteSnapshot(t, stream, "the remote bell", func(s protocol.RecentRouteSnapshot) bool {
92+ entry, ok := routeEntryNamed(s, "rem")
93+ return ok && entry.Attention
94+ })
95+ require.Greater(t, rung.Generation, first.Generation)
96+ require.Equal(t, first.Active, rung.Active, "the attachment keeps its reference")
97+
98+ harness.service.publishSnapshot(routeTestCatalogue(4, true))
99+ require.Never(t, func() bool { return len(sentRouteSnapshots(stream)) > 2 }, 50*time.Millisecond, time.Millisecond,
100+ "an unchanged catalogue publishes nothing")
101+ require.Len(t, harness.service.openedRequests(), 1)
102+}
103+
104+// TestSupervisorSettlesDaemonNavigation pins C5 and the E3 palette commits:
105+// the daemon's requests resolve through the broker catalogue and open through
106+// the broker exactly like a picker commit; a refusal is answered typed.
107+func TestSupervisorSettlesDaemonNavigation(t *testing.T) {
108+ beta := protocol.ExactSessionTarget{LifecycleID: pickerTestLifecycle(2), SessionName: "beta"}
109+ tests := []struct {
110+ name string
111+ // request builds the daemon message from the first route snapshot.
112+ request func(protocol.RecentRouteSnapshot) protocol.ServerMessage
113+ detached bool
114+ wantOpen func(*testing.T, ports.BrokerOpenStreamRequest)
115+ wantReply func(*testing.T, *sessionTestStream)
116+ }{
117+ {
118+ name: "recent route to another daemon swaps",
119+ request: func(s protocol.RecentRouteSnapshot) protocol.ServerMessage {
120+ entry, _ := routeEntryNamed(s, "rem")
121+ return protocol.RouteNavigationAction{SnapshotGeneration: s.Generation, Key: entry.Key, Generation: entry.Generation}
122+ },
123+ wantOpen: func(t *testing.T, request ports.BrokerOpenStreamRequest) {
124+ require.False(t, request.Local)
125+ require.Equal(t, routeTestRemote, request.Endpoint)
126+ require.Equal(t, ports.BrokerAdmissionExact, request.Admission)
127+ require.Equal(t, "rem", request.Target.SessionName)
128+ },
129+ },
130+ {
131+ name: "a stale reference is refused typed",
132+ request: func(s protocol.RecentRouteSnapshot) protocol.ServerMessage {
133+ return protocol.RouteNavigationAction{SnapshotGeneration: s.Generation, Key: 999, Generation: 1}
134+ },
135+ wantReply: func(t *testing.T, stream *sessionTestStream) {
136+ failure := awaitSent(t, stream, "RouteNavigationFailure", isSent[protocol.RouteNavigationFailure]).(protocol.RouteNavigationFailure)
137+ require.Equal(t, protocol.RouteNavigationFailure{Key: 999, Generation: 1, Code: protocol.RouteFailureStaleSelection}, failure)
138+ },
139+ },
140+ {
141+ name: "create on another host",
142+ request: func(s protocol.RecentRouteSnapshot) protocol.ServerMessage {
143+ host := s.Hosts[0]
144+ return protocol.RouteCreateSessionAction{RequestID: 4, SnapshotGeneration: s.Generation, Key: host.Key, Generation: host.Generation, SessionName: "fresh"}
145+ },
146+ wantOpen: func(t *testing.T, request ports.BrokerOpenStreamRequest) {
147+ require.Equal(t, routeTestRemote, request.Endpoint)
148+ require.Equal(t, ports.BrokerAdmissionCreateNamed, request.Admission)
149+ require.Equal(t, "fresh", request.Name)
150+ },
151+ },
152+ {
153+ name: "create with a stale host is refused typed",
154+ request: func(s protocol.RecentRouteSnapshot) protocol.ServerMessage {
155+ return protocol.RouteCreateSessionAction{RequestID: 4, SnapshotGeneration: s.Generation, Key: 999, Generation: 1, SessionName: "fresh"}
156+ },
157+ wantReply: func(t *testing.T, stream *sessionTestStream) {
158+ failure := awaitSent(t, stream, "SessionCreationFailure", isSent[protocol.SessionCreationFailure]).(protocol.SessionCreationFailure)
159+ require.Equal(t, protocol.SessionCreationFailure{RequestID: 4, Code: protocol.RouteFailureStaleSelection}, failure)
160+ },
161+ },
162+ {
163+ name: "a same-peer offer is confirmed in place",
164+ request: func(protocol.RecentRouteSnapshot) protocol.ServerMessage {
165+ target := beta
166+ return protocol.AttachTarget{Session: "beta", Intent: protocol.IntentAttach, ExactTarget: &target, EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, SamePeer: true}
167+ },
168+ wantReply: func(t *testing.T, stream *sessionTestStream) {
169+ request := awaitSent(t, stream, "SamePeerSwitchRequest", isSent[protocol.SamePeerSwitchRequest]).(protocol.SamePeerSwitchRequest)
170+ require.Equal(t, beta, request.Target)
171+ require.NotZero(t, request.RequestID)
172+ },
173+ },
174+ {
175+ name: "a close-and-dial handoff reopens on the serving daemon",
176+ request: func(protocol.RecentRouteSnapshot) protocol.ServerMessage {
177+ target := beta
178+ return protocol.AttachTarget{Session: "beta", Intent: protocol.IntentAttach, ExactTarget: &target, EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, PreferredTabID: "t_x"}
179+ },
180+ detached: true,
181+ wantOpen: func(t *testing.T, request ports.BrokerOpenStreamRequest) {
182+ require.True(t, request.Local)
183+ require.Equal(t, ports.BrokerAdmissionExact, request.Admission)
184+ require.Equal(t, beta, request.Target)
185+ },
186+ },
187+ }
188+ for _, tt := range tests {
189+ t.Run(tt.name, func(t *testing.T) {
190+ picker := newAttachTestPicker()
191+ harness := startAttachHarness(t, picker)
192+ harness.service.publishSnapshot(routeTestCatalogue(2, false))
193+ stream := attachLiveSession(t, harness, picker)
194+ snapshot := awaitRouteSnapshot(t, stream, "the route snapshot", func(s protocol.RecentRouteSnapshot) bool {
195+ _, ok := routeEntryNamed(s, "rem")
196+ return ok && len(s.Hosts) == 1
197+ })
198+
199+ var mu sync.Mutex
200+ var second *sessionTestStream
201+ harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
202+ mu.Lock()
203+ defer mu.Unlock()
204+ second = newSessionTestStream()
205+ return second, nil
206+ })
207+ stream.deliver(tt.request(snapshot))
208+ if tt.detached {
209+ // The daemon ends the source right after a close-and-dial
210+ // handoff; the supervisor still adopts the handoff.
211+ stream.deliver(protocol.Detached{Reason: protocol.ReasonDetach})
212+ }
213+
214+ if tt.wantOpen != nil {
215+ require.Eventually(t, func() bool { return len(harness.service.openedRequests()) == 2 }, 5*time.Second, time.Millisecond, "the navigation never opened a stream")
216+ opened := harness.service.openedRequests()
217+ tt.wantOpen(t, opened[1])
218+ awaitStreamHello(t, &mu, &second)
219+ if tt.detached {
220+ mu.Lock()
221+ hello := second.messages()[0].(protocol.Hello)
222+ mu.Unlock()
223+ require.Equal(t, domain.TabStableID("t_x"), hello.PreferredTabID)
224+ } else {
225+ awaitSent(t, stream, "Detach", isSent[protocol.Detach])
226+ }
227+ return
228+ }
229+ tt.wantReply(t, stream)
230+ require.Never(t, func() bool { return len(harness.service.openedRequests()) > 1 }, 50*time.Millisecond, time.Millisecond, "a refusal or an in-place switch never reconnects")
231+ require.Zero(t, countSent[protocol.Detach](stream))
232+ })
233+ }
234+}
235+
236+// TestSupervisorRouteSnapshotExcludesServingHost pins E3's destinations: a
237+// remote attachment offers the local daemon as a creation host and never
238+// itself.
239+func TestSupervisorRouteSnapshotExcludesServingHost(t *testing.T) {
240+ picker := newAttachTestPicker()
241+ harness := startAttachHarness(t, picker)
242+ remote := routeTestRemoteHost(catalogue.RemoteCatalogSession{LifecycleID: domain.SessionLifecycleID{1}, Name: "alpha", State: catalogue.RemoteCatalogSessionUp})
243+ remote.Registration = sessionTestRequest(false).Registration
244+ remote.Endpoint = sessionTestRequest(false).Endpoint
245+ remote.DisplayOrigin = "example"
246+ snapshot := routeTestSnapshot(routeTestLocal(routeTestLive("loc", 4, 1)), remote)
247+ snapshot.Revision = 2
248+ harness.service.publishSnapshot(snapshot)
249+
250+ stream := newSessionTestStream()
251+ harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
252+ return stream, nil
253+ })
254+ picker.commit(sessionTestRequest(false))
255+ awaitStreamHello(t, &sync.Mutex{}, &stream)
256+ deliverReadyStream(t, stream)
257+ awaitAttachedState(t, harness.sup)
258+
259+ published := awaitRouteSnapshot(t, stream, "the route snapshot", func(s protocol.RecentRouteSnapshot) bool { return len(s.Hosts) > 0 })
260+ require.Equal(t, []protocol.RouteHost{{Key: published.Hosts[0].Key, Generation: 1, Label: "local", Kind: protocol.RouteKindLocal}}, published.Hosts)
261+ require.Equal(t, protocol.RouteKindRemote, published.ActiveEntry.Kind)
262+ require.Equal(t, "example", published.ActiveEntry.HostLabel)
263+ require.True(t, published.Home.IsZero(), "only a local attachment is home")
264+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close