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 // refuseInPlace answers the in-place switch with a daemon refusal.
115 refuseInPlace bool
116 wantOpen func(*testing.T, ports.BrokerOpenStreamRequest)
117 wantReply func(*testing.T, *sessionTestStream)
118 }{
119 {
120 name: "recent route to another daemon swaps",
121 request: func(s protocol.RecentRouteSnapshot) protocol.ServerMessage {
122 entry, _ := routeEntryNamed(s, "rem")
123 return protocol.RouteNavigationAction{SnapshotGeneration: s.Generation, Key: entry.Key, Generation: entry.Generation}
124 },
125 wantOpen: func(t *testing.T, request ports.BrokerOpenStreamRequest) {
126 require.False(t, request.Local)
127 require.Equal(t, routeTestRemote, request.Endpoint)
128 require.Equal(t, ports.BrokerAdmissionExact, request.Admission)
129 require.Equal(t, "rem", request.Target.SessionName)
130 },
131 },
132 {
133 name: "recent route on the serving daemon switches in place",
134 request: func(s protocol.RecentRouteSnapshot) protocol.ServerMessage {
135 entry, _ := routeEntryNamed(s, "beta")
136 return protocol.RouteNavigationAction{SnapshotGeneration: s.Generation, Key: entry.Key, Generation: entry.Generation}
137 },
138 wantReply: func(t *testing.T, stream *sessionTestStream) {
139 request := awaitSent(t, stream, "SamePeerSwitchRequest", isSent[protocol.SamePeerSwitchRequest]).(protocol.SamePeerSwitchRequest)
140 require.Equal(t, beta, request.Target)
141 },
142 },
143 {
144 name: "a refused in-place route reconnects",
145 request: func(s protocol.RecentRouteSnapshot) protocol.ServerMessage {
146 entry, _ := routeEntryNamed(s, "beta")
147 return protocol.RouteNavigationAction{SnapshotGeneration: s.Generation, Key: entry.Key, Generation: entry.Generation}
148 },
149 refuseInPlace: true,
150 wantOpen: func(t *testing.T, request ports.BrokerOpenStreamRequest) {
151 require.True(t, request.Local)
152 require.Equal(t, beta, request.Target)
153 },
154 },
155 {
156 name: "a stale reference is refused typed",
157 request: func(s protocol.RecentRouteSnapshot) protocol.ServerMessage {
158 return protocol.RouteNavigationAction{SnapshotGeneration: s.Generation, Key: 999, Generation: 1}
159 },
160 wantReply: func(t *testing.T, stream *sessionTestStream) {
161 failure := awaitSent(t, stream, "RouteNavigationFailure", isSent[protocol.RouteNavigationFailure]).(protocol.RouteNavigationFailure)
162 require.Equal(t, protocol.RouteNavigationFailure{Key: 999, Generation: 1, Code: protocol.RouteFailureStaleSelection}, failure)
163 },
164 },
165 {
166 name: "create on another host",
167 request: func(s protocol.RecentRouteSnapshot) protocol.ServerMessage {
168 host := s.Hosts[0]
169 return protocol.RouteCreateSessionAction{RequestID: 4, SnapshotGeneration: s.Generation, Key: host.Key, Generation: host.Generation, SessionName: "fresh"}
170 },
171 wantOpen: func(t *testing.T, request ports.BrokerOpenStreamRequest) {
172 require.Equal(t, routeTestRemote, request.Endpoint)
173 require.Equal(t, ports.BrokerAdmissionCreateNamed, request.Admission)
174 require.Equal(t, "fresh", request.Name)
175 },
176 },
177 {
178 name: "create with a stale host is refused typed",
179 request: func(s protocol.RecentRouteSnapshot) protocol.ServerMessage {
180 return protocol.RouteCreateSessionAction{RequestID: 4, SnapshotGeneration: s.Generation, Key: 999, Generation: 1, SessionName: "fresh"}
181 },
182 wantReply: func(t *testing.T, stream *sessionTestStream) {
183 failure := awaitSent(t, stream, "SessionCreationFailure", isSent[protocol.SessionCreationFailure]).(protocol.SessionCreationFailure)
184 require.Equal(t, protocol.SessionCreationFailure{RequestID: 4, Code: protocol.RouteFailureStaleSelection}, failure)
185 },
186 },
187 {
188 name: "a same-peer offer is confirmed in place",
189 request: func(protocol.RecentRouteSnapshot) protocol.ServerMessage {
190 target := beta
191 return protocol.AttachTarget{Session: "beta", Intent: protocol.IntentAttach, ExactTarget: &target, EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, SamePeer: true}
192 },
193 wantReply: func(t *testing.T, stream *sessionTestStream) {
194 request := awaitSent(t, stream, "SamePeerSwitchRequest", isSent[protocol.SamePeerSwitchRequest]).(protocol.SamePeerSwitchRequest)
195 require.Equal(t, beta, request.Target)
196 require.NotZero(t, request.RequestID)
197 },
198 },
199 {
200 name: "a close-and-dial handoff reopens on the serving daemon",
201 request: func(protocol.RecentRouteSnapshot) protocol.ServerMessage {
202 target := beta
203 return protocol.AttachTarget{Session: "beta", Intent: protocol.IntentAttach, ExactTarget: &target, EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, PreferredTabID: "t_x"}
204 },
205 detached: true,
206 wantOpen: func(t *testing.T, request ports.BrokerOpenStreamRequest) {
207 require.True(t, request.Local)
208 require.Equal(t, ports.BrokerAdmissionExact, request.Admission)
209 require.Equal(t, beta, request.Target)
210 },
211 },
212 }
213 for _, tt := range tests {
214 t.Run(tt.name, func(t *testing.T) {
215 picker := newAttachTestPicker()
216 harness := startAttachHarness(t, picker)
217 harness.service.publishSnapshot(routeTestCatalogue(2, false))
218 stream := attachLiveSession(t, harness, picker)
219 snapshot := awaitRouteSnapshot(t, stream, "the route snapshot", func(s protocol.RecentRouteSnapshot) bool {
220 _, ok := routeEntryNamed(s, "rem")
221 return ok && len(s.Hosts) == 1
222 })
223
224 var mu sync.Mutex
225 var second *sessionTestStream
226 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
227 mu.Lock()
228 defer mu.Unlock()
229 second = newSessionTestStream()
230 return second, nil
231 })
232 stream.deliver(tt.request(snapshot))
233 if tt.refuseInPlace {
234 request := awaitSent(t, stream, "SamePeerSwitchRequest", isSent[protocol.SamePeerSwitchRequest]).(protocol.SamePeerSwitchRequest)
235 stream.deliver(protocol.SamePeerSwitchFailure{RequestID: request.RequestID, Code: protocol.SamePeerSwitchStaleTarget})
236 }
237 if tt.detached {
238 // The daemon ends the source right after a close-and-dial
239 // handoff; the supervisor still adopts the handoff.
240 stream.deliver(protocol.Detached{Reason: protocol.ReasonDetach})
241 }
242
243 if tt.wantOpen != nil {
244 require.Eventually(t, func() bool { return len(harness.service.openedRequests()) == 2 }, 5*time.Second, time.Millisecond, "the navigation never opened a stream")
245 opened := harness.service.openedRequests()
246 tt.wantOpen(t, opened[1])
247 awaitStreamHello(t, &mu, &second)
248 if tt.detached {
249 mu.Lock()
250 hello := second.messages()[0].(protocol.Hello)
251 mu.Unlock()
252 require.Equal(t, domain.TabStableID("t_x"), hello.PreferredTabID)
253 } else {
254 awaitSent(t, stream, "Detach", isSent[protocol.Detach])
255 }
256 return
257 }
258 tt.wantReply(t, stream)
259 // The supervisor chose before the worker sent the reply above, so
260 // no reconnect can still be pending.
261 require.Len(t, harness.service.openedRequests(), 1, "a refusal or an in-place switch never reconnects")
262 require.Zero(t, countSent[protocol.Detach](stream))
263 })
264 }
265}
266
267// TestSupervisorRouteSnapshotExcludesServingHost pins E3's destinations: a
268// remote attachment offers the local daemon as a creation host and never
269// itself.
270func TestSupervisorRouteSnapshotExcludesServingHost(t *testing.T) {
271 picker := newAttachTestPicker()
272 harness := startAttachHarness(t, picker)
273 remote := routeTestRemoteHost(catalogue.RemoteCatalogSession{LifecycleID: domain.SessionLifecycleID{1}, Name: "alpha", State: catalogue.RemoteCatalogSessionUp})
274 remote.Registration = sessionTestRequest(false).Registration
275 remote.Endpoint = sessionTestRequest(false).Endpoint
276 remote.DisplayOrigin = "example"
277 snapshot := routeTestSnapshot(routeTestLocal(routeTestLive("loc", 4, 1)), remote)
278 snapshot.Revision = 2
279 harness.service.publishSnapshot(snapshot)
280
281 stream := newSessionTestStream()
282 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
283 return stream, nil
284 })
285 picker.commit(sessionTestRequest(false))
286 awaitStreamHello(t, &sync.Mutex{}, &stream)
287 deliverReadyStream(t, stream)
288 awaitAttachedState(t, harness.sup)
289
290 published := awaitRouteSnapshot(t, stream, "the route snapshot", func(s protocol.RecentRouteSnapshot) bool { return len(s.Hosts) > 0 })
291 require.Equal(t, []protocol.RouteHost{{Key: published.Hosts[0].Key, Generation: 1, Label: "local", Kind: protocol.RouteKindLocal}}, published.Hosts)
292 require.Equal(t, protocol.RouteKindRemote, published.ActiveEntry.Kind)
293 require.Equal(t, "example", published.ActiveEntry.HostLabel)
294 require.True(t, published.Home.IsZero(), "only a local attachment is home")
295}
296
297// TestSupervisorSamePeerSwitchSettlesUIAction pins the driver action that
298// causes a daemon same-peer switch (palette "Switch to session" on the serving
299// daemon): the attachment is kept, so the action must follow the offer and
300// settle on the destination's first committed output — never on a source
301// publication or a source-boundary receipt — and a refused switch fails it.
302func TestSupervisorSamePeerSwitchSettlesUIAction(t *testing.T) {
303 beta := protocol.ExactSessionTarget{LifecycleID: pickerTestLifecycle(2), SessionName: "beta"}
304 tests := []struct {
305 name string
306 // settle delivers the daemon's answer to the confirmed switch.
307 settle func(*sessionTestStream, protocol.SamePeerSwitchRequest)
308 // sourceReceipt delivers a daemon receipt on the source boundary; the
309 // daemon may also lose the source fence and never answer it.
310 sourceReceipt bool
311 wantErr ports.UIErrorCode
312 }{
313 {
314 name: "destination output completes the action without a daemon receipt",
315 settle: func(stream *sessionTestStream, _ protocol.SamePeerSwitchRequest) {
316 destination := sessionTestOutput(4, "\x1b[Hbeta")
317 destination.Epoch = 2
318 destination.Context.Route.Target = beta
319 stream.deliver(destination)
320 },
321 },
322 {
323 name: "a source-boundary receipt waits for the destination output",
324 sourceReceipt: true,
325 settle: func(stream *sessionTestStream, _ protocol.SamePeerSwitchRequest) {
326 destination := sessionTestOutput(4, "\x1b[Hbeta")
327 destination.Epoch = 2
328 destination.Context.Route.Target = beta
329 stream.deliver(destination)
330 },
331 },
332 {
333 name: "a refused switch fails the action",
334 sourceReceipt: true,
335 settle: func(stream *sessionTestStream, request protocol.SamePeerSwitchRequest) {
336 stream.deliver(protocol.SamePeerSwitchFailure{RequestID: request.RequestID, Code: protocol.SamePeerSwitchStaleTarget})
337 },
338 wantErr: ports.UIErrNavigationFailed,
339 },
340 }
341 for _, tt := range tests {
342 t.Run(tt.name, func(t *testing.T) {
343 picker := newAttachTestPicker()
344 var ui *UI
345 harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) {
346 ui = NewUI(cfg.Terminal.(ports.UIState), cfg.Clock)
347 cfg.UI = ui
348 })
349 stream := attachLiveSession(t, harness, picker)
350 attached := sessionTestOutput(2, "\x1b[Hattached")
351 attached.Full, attached.Base, attached.New = false, 1, 2
352 stream.deliver(attached)
353 var snapshot ports.UISnapshot
354 require.Eventually(t, func() bool {
355 var err error
356 snapshot, err = ui.Capture(ui.Handle())
357 return err == nil && snapshot.Context.Status == ports.UIStatusAttached && snapshot.Context.ViewPublication == 2
358 }, 5*time.Second, time.Millisecond)
359
360 type actionOutcome struct {
361 result ports.UIActionResult
362 err error
363 }
364 done := make(chan actionOutcome, 1)
365 go func() {
366 result, err := ui.Action(t.Context(), ports.UIActionRequest{Attachment: ui.Handle(), Generation: snapshot.Context.Generation, Keys: []string{"Enter"}})
367 done <- actionOutcome{result: result, err: err}
368 }()
369 var actionID uint64
370 require.Eventually(t, func() bool {
371 for _, message := range stream.messages() {
372 if fence, ok := message.(protocol.UIFence); ok && fence.ActionID != 0 {
373 actionID = fence.ActionID
374 return true
375 }
376 }
377 return false
378 }, 5*time.Second, time.Millisecond, "the action never reached the daemon")
379
380 target := beta
381 stream.deliver(protocol.AttachTarget{Session: "beta", Intent: protocol.IntentAttach, ExactTarget: &target, EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, SamePeer: true, CauseActionID: actionID})
382 request := awaitSent(t, stream, "SamePeerSwitchRequest", isSent[protocol.SamePeerSwitchRequest]).(protocol.SamePeerSwitchRequest)
383
384 // A source publication and a receipt on the source boundary arrive
385 // before the destination: neither may settle the navigating action.
386 source := sessionTestOutput(3, "\x1b[Hsource")
387 source.Full, source.Base, source.New = false, 2, 3
388 stream.deliver(source)
389 if tt.sourceReceipt {
390 stream.deliver(protocol.UIReceipt{ActionID: actionID, Epoch: 1, State: 3, ViewPublication: 3, Outcome: protocol.UIReceiptProcessed})
391 }
392 require.Eventually(t, func() bool {
393 ui.mu.Lock()
394 defer ui.mu.Unlock()
395 return ui.boundary.Context.ViewPublication == 3
396 }, 5*time.Second, time.Millisecond, "the source publication was never committed")
397 select {
398 case outcome := <-done:
399 t.Fatalf("the action settled before the destination: %+v %v", outcome.result, outcome.err)
400 default:
401 }
402
403 tt.settle(stream, request)
404 var outcome actionOutcome
405 select {
406 case outcome = <-done:
407 case <-time.After(5 * time.Second):
408 t.Fatal("the same-peer switch never settled the UI action")
409 }
410 if tt.wantErr != "" {
411 var uiErr *ports.UIError
412 require.ErrorAs(t, outcome.err, &uiErr)
413 require.Equal(t, tt.wantErr, uiErr.Code)
414 require.Equal(t, actionID, uiErr.ActionID)
415 return
416 }
417 require.NoError(t, outcome.err)
418 require.Equal(t, ports.UIActionProcessed, outcome.result.Status)
419 require.Equal(t, beta, outcome.result.Context.Route.Target)
420 require.Equal(t, snapshot.Context.Generation, outcome.result.Context.Generation, "an in-place switch keeps the actionable generation")
421 })
422 }
423}