1package client
2
3import (
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.
24func 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.
34func 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
44func 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
59func 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.
72func 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.
107func 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.
239func 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}
265
266// TestSupervisorSamePeerSwitchSettlesUIAction pins the driver action that
267// causes a daemon same-peer switch (palette "Switch to session" on the serving
268// daemon): the attachment is kept, so the action must follow the offer and
269// settle on the destination's first committed output — never on a source
270// publication or a source-boundary receipt — and a refused switch fails it.
271func TestSupervisorSamePeerSwitchSettlesUIAction(t *testing.T) {
272 beta := protocol.ExactSessionTarget{LifecycleID: pickerTestLifecycle(2), SessionName: "beta"}
273 tests := []struct {
274 name string
275 // settle delivers the daemon's answer to the confirmed switch.
276 settle func(*sessionTestStream, protocol.SamePeerSwitchRequest)
277 // sourceReceipt delivers a daemon receipt on the source boundary; the
278 // daemon may also lose the source fence and never answer it.
279 sourceReceipt bool
280 wantErr ports.UIErrorCode
281 }{
282 {
283 name: "destination output completes the action without a daemon receipt",
284 settle: func(stream *sessionTestStream, _ protocol.SamePeerSwitchRequest) {
285 destination := sessionTestOutput(4, "\x1b[Hbeta")
286 destination.Epoch = 2
287 destination.Context.Route.Target = beta
288 stream.deliver(destination)
289 },
290 },
291 {
292 name: "a source-boundary receipt waits for the destination output",
293 sourceReceipt: true,
294 settle: func(stream *sessionTestStream, _ protocol.SamePeerSwitchRequest) {
295 destination := sessionTestOutput(4, "\x1b[Hbeta")
296 destination.Epoch = 2
297 destination.Context.Route.Target = beta
298 stream.deliver(destination)
299 },
300 },
301 {
302 name: "a refused switch fails the action",
303 sourceReceipt: true,
304 settle: func(stream *sessionTestStream, request protocol.SamePeerSwitchRequest) {
305 stream.deliver(protocol.SamePeerSwitchFailure{RequestID: request.RequestID, Code: protocol.SamePeerSwitchStaleTarget})
306 },
307 wantErr: ports.UIErrNavigationFailed,
308 },
309 }
310 for _, tt := range tests {
311 t.Run(tt.name, func(t *testing.T) {
312 picker := newAttachTestPicker()
313 var ui *UI
314 harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) {
315 ui = NewUI(cfg.Terminal.(ports.UIState), cfg.Clock)
316 cfg.UI = ui
317 })
318 stream := attachLiveSession(t, harness, picker)
319 attached := sessionTestOutput(2, "\x1b[Hattached")
320 attached.Full, attached.Base, attached.New = false, 1, 2
321 stream.deliver(attached)
322 var snapshot ports.UISnapshot
323 require.Eventually(t, func() bool {
324 var err error
325 snapshot, err = ui.Capture(ui.Handle())
326 return err == nil && snapshot.Context.Status == ports.UIStatusAttached && snapshot.Context.ViewPublication == 2
327 }, 5*time.Second, time.Millisecond)
328
329 type actionOutcome struct {
330 result ports.UIActionResult
331 err error
332 }
333 done := make(chan actionOutcome, 1)
334 go func() {
335 result, err := ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: snapshot.Context.Generation, Keys: []string{"Enter"}})
336 done <- actionOutcome{result: result, err: err}
337 }()
338 var actionID uint64
339 require.Eventually(t, func() bool {
340 for _, message := range stream.messages() {
341 if fence, ok := message.(protocol.UIFence); ok && fence.ActionID != 0 {
342 actionID = fence.ActionID
343 return true
344 }
345 }
346 return false
347 }, 5*time.Second, time.Millisecond, "the action never reached the daemon")
348
349 target := beta
350 stream.deliver(protocol.AttachTarget{Session: "beta", Intent: protocol.IntentAttach, ExactTarget: &target, EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, SamePeer: true, CauseActionID: actionID})
351 request := awaitSent(t, stream, "SamePeerSwitchRequest", isSent[protocol.SamePeerSwitchRequest]).(protocol.SamePeerSwitchRequest)
352
353 // A source publication and a receipt on the source boundary arrive
354 // before the destination: neither may settle the navigating action.
355 source := sessionTestOutput(3, "\x1b[Hsource")
356 source.Full, source.Base, source.New = false, 2, 3
357 stream.deliver(source)
358 if tt.sourceReceipt {
359 stream.deliver(protocol.UIReceipt{ActionID: actionID, Epoch: 1, State: 3, ViewPublication: 3, Outcome: protocol.UIReceiptProcessed})
360 }
361 require.Eventually(t, func() bool {
362 ui.mu.Lock()
363 defer ui.mu.Unlock()
364 return ui.boundary.Context.ViewPublication == 3
365 }, 5*time.Second, time.Millisecond, "the source publication was never committed")
366 select {
367 case outcome := <-done:
368 t.Fatalf("the action settled before the destination: %+v %v", outcome.result, outcome.err)
369 default:
370 }
371
372 tt.settle(stream, request)
373 var outcome actionOutcome
374 select {
375 case outcome = <-done:
376 case <-time.After(5 * time.Second):
377 t.Fatal("the same-peer switch never settled the UI action")
378 }
379 if tt.wantErr != "" {
380 var uiErr *ports.UIError
381 require.ErrorAs(t, outcome.err, &uiErr)
382 require.Equal(t, tt.wantErr, uiErr.Code)
383 require.Equal(t, actionID, uiErr.ActionID)
384 return
385 }
386 require.NoError(t, outcome.err)
387 require.Equal(t, ports.UIActionProcessed, outcome.result.Status)
388 require.Equal(t, beta, outcome.result.Context.Route.Target)
389 require.Equal(t, snapshot.Context.Generation, outcome.result.Context.Generation, "an in-place switch keeps the actionable generation")
390 })
391 }
392}