1package app
2
3// List/kill/stop broker-only caller contract (Plan 001 P7.4f).
4//
5// runList, runKill, and requestDaemonStop reach the sole per-user broker through
6// the private connectBroker composition seam. These tests substitute a scripted
7// broker connection there, so every case drives the real caller code without a
8// production broker process, a daemon dialer, or a real signal. They pin the
9// explicit outcome table: success is printed only for an explicit KillSucceeded,
10// a definite failure keeps its bounded error, and a lost or uncorrelated reply
11// is an exit-3 unknown outcome that is never read as success.
12
13import (
14 "bytes"
15 "context"
16 "errors"
17 "io"
18 "strings"
19 "sync"
20 "testing"
21
22 "github.com/stretchr/testify/require"
23
24 "github.com/bnema/vev/internal/domain"
25 "github.com/bnema/vev/internal/ports"
26 "github.com/bnema/vev/internal/protocol"
27 "github.com/bnema/vev/internal/protocol/catalogue"
28)
29
30const seamBrokerTestEpoch = ports.BrokerEpoch(0x3c)
31
32// seamBrokerTestPolicy is the exact local policy the scripted snapshot carries.
33func seamBrokerTestPolicy() ports.BrokerPolicy {
34 return ports.BrokerPolicy{
35 ProtocolVersion: protocol.Version,
36 CatalogSchemaVersion: catalogue.RemoteCatalogSchemaVersion,
37 EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned,
38 Transport: "unix-mux",
39 Trust: "same-user",
40 Launch: "explicit",
41 Isolation: "per-user",
42 }
43}
44
45// seamBrokerTestSnapshot publishes one reachable local daemon at the scripted
46// epoch, so a route built from it validates against the served snapshot.
47func seamBrokerTestSnapshot() ports.BrokerSnapshot {
48 return ports.BrokerSnapshot{
49 Epoch: seamBrokerTestEpoch,
50 Revision: 1,
51 Daemons: []ports.BrokerDaemonObservation{{
52 Local: true,
53 DisplayOrigin: "local",
54 Policy: seamBrokerTestPolicy(),
55 Identity: ports.BrokerDaemonIdentity("local-daemon"),
56 Incarnation: ports.BrokerDaemonIncarnation{9},
57 ProtocolVersion: protocol.Version,
58 Availability: domain.RemoteAvailabilityReachable,
59 }},
60 }
61}
62
63// seamBrokerStream is one scripted broker logical stream. It records every
64// typed client message and serves a fixed reply queue: an exhausted queue is an
65// EOF, which is exactly the lost-reply condition the caller must classify as
66// unknown rather than success.
67type seamBrokerStream struct {
68 mu sync.Mutex
69 sent []protocol.ClientMessage
70 replies []protocol.ServerMessage
71 closed bool
72 done chan struct{}
73}
74
75func newSeamBrokerStream(replies ...protocol.ServerMessage) *seamBrokerStream {
76 return &seamBrokerStream{replies: replies, done: make(chan struct{})}
77}
78
79func (s *seamBrokerStream) SendClient(message protocol.ClientMessage) error {
80 s.mu.Lock()
81 defer s.mu.Unlock()
82 s.sent = append(s.sent, message)
83 return nil
84}
85
86func (s *seamBrokerStream) ReceiveServer() (protocol.ServerMessage, error) {
87 s.mu.Lock()
88 defer s.mu.Unlock()
89 if len(s.replies) == 0 {
90 return nil, io.EOF
91 }
92 reply := s.replies[0]
93 s.replies = s.replies[1:]
94 return reply, nil
95}
96
97func (s *seamBrokerStream) Capabilities() protocol.ConnectionCapabilities {
98 return protocol.ConnectionCapabilities{}
99}
100
101func (s *seamBrokerStream) LinkState() ports.LinkState { return ports.LinkStateConnected }
102func (s *seamBrokerStream) LinkEvents() <-chan ports.LinkEvent { return nil }
103func (s *seamBrokerStream) Done() <-chan struct{} { return s.done }
104func (s *seamBrokerStream) Err() error { return nil }
105
106func (s *seamBrokerStream) Close() error {
107 s.mu.Lock()
108 defer s.mu.Unlock()
109 if !s.closed {
110 s.closed = true
111 close(s.done)
112 }
113 return nil
114}
115
116// seamBrokerService is a scripted ports.BrokerService. Every open returns the
117// same scripted stream and records the exact open request, so a test can assert
118// the start mode each caller authorized.
119type seamBrokerService struct {
120 snapshot ports.BrokerSnapshot
121 stream *seamBrokerStream
122
123 mu sync.Mutex
124 next ports.BrokerStreamID
125 opens []ports.BrokerOpenStreamRequest
126 closed int
127}
128
129func newSeamBrokerService(stream *seamBrokerStream) *seamBrokerService {
130 return &seamBrokerService{snapshot: seamBrokerTestSnapshot(), stream: stream}
131}
132
133func (s *seamBrokerService) ConnectionID() ports.BrokerConnectionID {
134 return ports.BrokerConnectionID{4}
135}
136
137func (s *seamBrokerService) NextStreamID() (ports.BrokerStreamID, error) {
138 s.mu.Lock()
139 defer s.mu.Unlock()
140 s.next++
141 return s.next, nil
142}
143
144func (s *seamBrokerService) Done() <-chan struct{} { return make(chan struct{}) }
145func (s *seamBrokerService) Err() error { return nil }
146
147func (s *seamBrokerService) Snapshot() ports.BrokerSnapshot { return s.snapshot.Clone() }
148
149func (s *seamBrokerService) Subscribe() (ports.BrokerSubscription, error) {
150 return nil, errors.New("the seam broker service does not support subscriptions")
151}
152
153func (s *seamBrokerService) SubscribePreview(ports.BrokerPreviewRequest) (ports.BrokerPreviewSubscription, error) {
154 return nil, errors.New("the seam broker service does not support preview")
155}
156
157func (s *seamBrokerService) OpenStream(_ context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
158 s.mu.Lock()
159 s.opens = append(s.opens, request)
160 s.mu.Unlock()
161 return s.stream, nil
162}
163
164func (s *seamBrokerService) CloseStream(ports.BrokerConnectionID, ports.BrokerStreamID) error {
165 return nil
166}
167
168func (s *seamBrokerService) AddHost(context.Context, string, ports.BrokerPolicy) (domain.RemoteRegistration, error) {
169 return domain.RemoteRegistration{}, nil
170}
171
172func (s *seamBrokerService) RemoveHost(context.Context, domain.RemoteRegistration) (bool, error) {
173 return false, nil
174}
175
176func (s *seamBrokerService) UpdateHostPolicy(context.Context, domain.RemoteRegistration, ports.BrokerPolicy) (domain.RemoteRegistration, error) {
177 return domain.RemoteRegistration{}, nil
178}
179
180func (s *seamBrokerService) RequestReconcile(string) {}
181
182func (s *seamBrokerService) Close() error {
183 s.mu.Lock()
184 s.closed++
185 s.mu.Unlock()
186 return nil
187}
188
189func (s *seamBrokerService) openedRequests() []ports.BrokerOpenStreamRequest {
190 s.mu.Lock()
191 defer s.mu.Unlock()
192 return append([]ports.BrokerOpenStreamRequest(nil), s.opens...)
193}
194
195func (s *seamBrokerService) closeCount() int {
196 s.mu.Lock()
197 defer s.mu.Unlock()
198 return s.closed
199}
200
201var _ ports.BrokerService = (*seamBrokerService)(nil)
202
203// withSeamBroker installs one scripted broker connection on the connectBroker
204// seam for the duration of a test, so no production broker is ever dialed.
205func withSeamBroker(t *testing.T, service ports.BrokerService) {
206 t.Helper()
207 previous := connectBroker
208 previousList := connectListBroker
209 previousStop := connectDaemonStopBroker
210 connectBroker = func(context.Context) (ports.BrokerService, error) { return service, nil }
211 connectListBroker = connectBroker
212 connectDaemonStopBroker = connectBroker
213 t.Cleanup(func() {
214 connectBroker = previous
215 connectListBroker = previousList
216 connectDaemonStopBroker = previousStop
217 })
218}
219
220func seamKillResult(requestID uint64, outcome protocol.KillOutcome, code uint16, text string) protocol.KillResult {
221 return protocol.KillResult{RequestID: requestID, Outcome: outcome, Code: code, Text: text}
222}
223
224func TestRunBrokerSnapshotListReportsNoDaemon(t *testing.T) {
225 remote := ports.BrokerDaemonObservation{
226 Endpoint: "demo@host.test", DisplayOrigin: "demo@host.test",
227 Registration: domain.RemoteRegistration{Endpoint: "demo@host.test", Incarnation: [16]byte{1}, Generation: 1},
228 Policy: seamBrokerTestPolicy(), Availability: domain.RemoteAvailabilityNoDaemon,
229 Sessions: []catalogue.RemoteCatalogSession{},
230 }
231 snapshot := ports.BrokerSnapshot{Epoch: seamBrokerTestEpoch, Revision: 1, Daemons: []ports.BrokerDaemonObservation{remote}}
232
233 var all bytes.Buffer
234 require.NoError(t, runBrokerSnapshotList(command{listAll: true}, snapshot, &all))
235 require.Contains(t, all.String(), "demo@host.test: no vev daemon")
236 require.Contains(t, all.String(), "Enter in the picker to create a session")
237 require.NotContains(t, all.String(), "no sessions")
238
239 var host bytes.Buffer
240 require.NoError(t, runBrokerSnapshotList(command{listHost: "demo@host.test"}, snapshot, &host))
241 require.Contains(t, host.String(), "no vev daemon")
242}
243
244func TestRunBrokerSnapshotListAllShowsNoDaemonAlongsideSessions(t *testing.T) {
245 remote := ports.BrokerDaemonObservation{
246 Endpoint: "demo@host.test", DisplayOrigin: "demo@host.test",
247 Registration: domain.RemoteRegistration{Endpoint: "demo@host.test", Incarnation: [16]byte{1}, Generation: 1},
248 Policy: seamBrokerTestPolicy(), Availability: domain.RemoteAvailabilityNoDaemon,
249 Sessions: []catalogue.RemoteCatalogSession{},
250 }
251 local := seamBrokerTestSnapshot().Daemons[0]
252 local.InventoryKnown = true
253 local.Sessions = []catalogue.RemoteCatalogSession{{LifecycleID: domain.SessionLifecycleID{1}, Name: "work", State: catalogue.RemoteCatalogSessionUp, Tabs: []catalogue.RemoteCatalogTab{}}}
254 snapshot := ports.BrokerSnapshot{Epoch: seamBrokerTestEpoch, Revision: 1, Daemons: []ports.BrokerDaemonObservation{local, remote}}
255 var output bytes.Buffer
256 require.NoError(t, runBrokerSnapshotList(command{listAll: true}, snapshot, &output))
257 require.Contains(t, output.String(), "demo@host.test: no vev daemon")
258 require.Contains(t, output.String(), "work")
259}
260
261// TestRunListRendersBrokerSessionsWithoutADaemon pins that runList reads the
262// live session list through BrokerOperations over the per-user broker and
263// renders it, so a local list never reads durable state or dials a daemon.
264func TestRunListRendersBrokerSessionsWithoutADaemon(t *testing.T) {
265 stream := newSeamBrokerStream(protocol.Sessions{Sessions: []protocol.SessionInfo{
266 {Name: "work", State: protocol.SessionUp, Tabs: 2, Attached: true},
267 }})
268 service := newSeamBrokerService(stream)
269 withSeamBroker(t, service)
270
271 got := captureStdout(t, func() {
272 require.NoError(t, runList(context.Background(), command{kind: kindList}))
273 })
274 for _, want := range []string{"NAME", "work", "up", "2", "yes"} {
275 require.Contains(t, got, want)
276 }
277 require.Equal(t, 1, service.closeCount(), "the borrowed broker service is closed exactly once")
278 opens := service.openedRequests()
279 require.Len(t, opens, 1)
280 require.Equal(t, ports.BrokerStreamControl, opens[0].Purpose)
281 require.True(t, opens[0].Local)
282 require.Equal(t, ports.BrokerDaemonExistingOnly, opens[0].StartMode,
283 "a listing must not start a stopped daemon")
284}
285
286// TestRunKillExplicitOutcomesTable pins the runKill outcome contract: an
287// explicit success prints, a definite failure keeps its bounded error at exit
288// one, and a lost or uncorrelated reply is an exit-three unknown that is never
289// printed as success.
290func TestRunKillExplicitOutcomesTable(t *testing.T) {
291 tests := []struct {
292 name string
293 kill []string
294 replies []protocol.ServerMessage
295 wantCode int
296 wantUnknown bool
297 wantStdout string
298 wantErr string
299 }{
300 {
301 name: "named success prints only on explicit success",
302 kill: []string{"work", "false", "false"},
303 replies: []protocol.ServerMessage{seamKillResult(1, protocol.KillSucceeded, 0, "")},
304 wantCode: 0,
305 wantStdout: "killed work",
306 },
307 {
308 name: "kill-sessions success",
309 kill: []string{"", "true", "false"},
310 replies: []protocol.ServerMessage{seamKillResult(1, protocol.KillSucceeded, 0, "")},
311 wantCode: 0,
312 wantStdout: "killed all sessions",
313 },
314 {
315 name: "definite failure is a bounded exit-one failure",
316 kill: []string{"work", "false", "false"},
317 replies: []protocol.ServerMessage{seamKillResult(1, protocol.KillFailed, protocol.ErrInternal, "no such session: work")},
318 wantCode: 1,
319 wantErr: "no such session: work",
320 },
321 {
322 name: "explicit unknown outcome is exit three",
323 kill: []string{"work", "false", "false"},
324 replies: []protocol.ServerMessage{seamKillResult(1, protocol.KillOutcomeUnknown, protocol.ErrServerShutdown, "daemon is shutting down")},
325 wantCode: 3,
326 wantUnknown: true,
327 },
328 {
329 name: "lost reply is exit three and never success",
330 kill: []string{"work", "false", "false"},
331 wantCode: 3,
332 wantUnknown: true,
333 },
334 {
335 name: "uncorrelated success reply is exit three",
336 kill: []string{"work", "false", "false"},
337 replies: []protocol.ServerMessage{seamKillResult(999, protocol.KillSucceeded, 0, "")},
338 wantCode: 3,
339 wantUnknown: true,
340 },
341 }
342 for _, tt := range tests {
343 t.Run(tt.name, func(t *testing.T) {
344 stream := newSeamBrokerStream(tt.replies...)
345 service := newSeamBrokerService(stream)
346 withSeamBroker(t, service)
347
348 var err error
349 got := captureStdout(t, func() {
350 err = runKill(context.Background(), tt.kill[0], tt.kill[1] == "true")
351 })
352 require.Equal(t, tt.wantCode, ExitCode(err), "exit code for %v", tt.kill)
353 if tt.wantUnknown {
354 require.ErrorIs(t, err, errKillOutcomeUnknown, "a lost or uncorrelated reply is unknown")
355 require.NotContains(t, got, "killed", "an unknown outcome is never printed as success")
356 } else if tt.wantErr != "" {
357 require.ErrorContains(t, err, tt.wantErr)
358 } else {
359 require.NoError(t, err)
360 }
361 if tt.wantStdout != "" {
362 require.Contains(t, got, tt.wantStdout)
363 }
364 opens := service.openedRequests()
365 require.Len(t, opens, 1, "exactly one control stream per kill")
366 require.Equal(t, ports.BrokerStreamControl, opens[0].Purpose)
367 require.Equal(t, 1, stream.sentCount(), "the kill is never replayed")
368 })
369 }
370}
371
372// TestRunKillUnreachableBrokerIsExitThree pins that a broker that cannot be
373// reached is reported as a distinct exit-three unreachable failure before any
374// stream is opened.
375func TestRunKillUnreachableBrokerIsExitThree(t *testing.T) {
376 previous := connectBroker
377 connectBroker = func(context.Context) (ports.BrokerService, error) {
378 return nil, errors.New("broker not reachable")
379 }
380 t.Cleanup(func() { connectBroker = previous })
381
382 err := runKill(context.Background(), "work", false)
383 require.Equal(t, 3, ExitCode(err))
384 require.ErrorIs(t, err, errDaemonUnreachable)
385}
386
387// TestRequestDaemonStopUsesExistingOnlyAndExplicitOutcomes pins the daemon-stop
388// contract: the stop authorizes ExistingOnly so it can never start the daemon it
389// is stopping, success is exactly an explicit KillSucceeded, a definite failure
390// is a plain error, and a lost reply is an explicit unknown outcome.
391func TestRequestDaemonStopUsesExistingOnlyAndExplicitOutcomes(t *testing.T) {
392 tests := []struct {
393 name string
394 replies []protocol.ServerMessage
395 wantUnknown bool
396 wantErr string
397 }{
398 {
399 name: "explicit success",
400 replies: []protocol.ServerMessage{seamKillResult(1, protocol.KillSucceeded, 0, "")},
401 },
402 {
403 name: "definite failure keeps its text",
404 replies: []protocol.ServerMessage{seamKillResult(1, protocol.KillFailed, protocol.ErrInternal, "daemon refused stop")},
405 wantErr: "daemon refused stop",
406 },
407 {
408 name: "lost reply is unknown",
409 wantUnknown: true,
410 },
411 {
412 name: "explicit unknown stays unknown",
413 replies: []protocol.ServerMessage{seamKillResult(1, protocol.KillOutcomeUnknown, protocol.ErrServerShutdown, "daemon is shutting down")},
414 wantUnknown: true,
415 },
416 }
417 for _, tt := range tests {
418 t.Run(tt.name, func(t *testing.T) {
419 stream := newSeamBrokerStream(tt.replies...)
420 service := newSeamBrokerService(stream)
421 withSeamBroker(t, service)
422
423 err := requestDaemonStop(context.Background())
424 opens := service.openedRequests()
425 require.Len(t, opens, 1)
426 require.Equal(t, ports.BrokerStreamControl, opens[0].Purpose)
427 require.Equal(t, ports.BrokerDaemonExistingOnly, opens[0].StartMode,
428 "a daemon-stop must never authorize starting the daemon it stops")
429
430 if tt.wantUnknown {
431 require.ErrorIs(t, err, errKillOutcomeUnknown)
432 } else if tt.wantErr != "" {
433 require.ErrorContains(t, err, tt.wantErr)
434 require.NotErrorIs(t, err, errKillOutcomeUnknown)
435 } else {
436 require.NoError(t, err)
437 }
438 })
439 }
440}
441
442// sentCount reports how many typed messages one scripted stream received.
443func (s *seamBrokerStream) sentCount() int {
444 s.mu.Lock()
445 defer s.mu.Unlock()
446 return len(s.sent)
447}
448
449// TestCreateDetachedLocalSessionUsesBrokerOperations proves nested `vev new`
450// creates through one broker control stream and never needs a daemon dialer.
451func TestCreateDetachedLocalSessionUsesBrokerOperations(t *testing.T) {
452 t.Setenv("VEV", "session=outer,tab=1,pane=1")
453 stream := newSeamBrokerStream(protocol.CommandResult{RequestID: 1, Outcome: protocol.CommandSucceeded})
454 service := newSeamBrokerService(stream)
455 withSeamBroker(t, service)
456
457 require.NoError(t, createDetachedLocalSession(context.Background(), "smoke"))
458 require.Equal(t, 1, service.closeCount())
459 require.Len(t, service.openedRequests(), 1)
460 stream.mu.Lock()
461 defer stream.mu.Unlock()
462 require.Len(t, stream.sent, 1)
463 request, ok := stream.sent[0].(protocol.CommandRequest)
464 require.True(t, ok)
465 require.Equal(t, "new-session", request.Slug)
466 require.Equal(t, []string{"smoke"}, request.Args)
467 require.Equal(t, "outer", request.TargetSession)
468 require.False(t, request.Attached)
469 require.False(t, request.Self)
470}
471
472// TestDetachedCreationProductionSourceHasNoDaemonBypass keeps the production
473// helper on the broker contract: direct IPC and daemon lifecycle helpers may
474// not reappear in its body as a fallback.
475func TestDetachedCreationProductionSourceHasNoDaemonBypass(t *testing.T) {
476 sources := loadAppSources(t, false)
477 body := sources["sessions.go"]
478 start := strings.Index(body, "func createDetachedLocalSession")
479 require.NotEqual(t, -1, start)
480 end := strings.Index(body[start:], "\nfunc renderDetachedCreationResult")
481 require.NotEqual(t, -1, end)
482 helper := body[start : start+end]
483 require.Contains(t, helper, "client.NewBrokerOperations")
484 require.Contains(t, helper, "operations.Command")
485 require.NotContains(t, helper, "ipc.DialContext")
486 require.NotContains(t, helper, "ensureDaemon")
487 require.NotContains(t, helper, "realDial")
488}