1//go:build linux
2
3package app
4
5import (
6 "context"
7 "crypto/rand"
8 "errors"
9 "fmt"
10 "io"
11 "log/slog"
12 "net"
13 "os"
14 "path/filepath"
15 "strings"
16 "sync"
17 "sync/atomic"
18 "testing"
19 "time"
20
21 "github.com/stretchr/testify/require"
22
23 vt "github.com/bnema/vev-vt"
24 "github.com/bnema/vev/internal/adapters/clock"
25 "github.com/bnema/vev/internal/adapters/ipc"
26 "github.com/bnema/vev/internal/adapters/lifecycle"
27 "github.com/bnema/vev/internal/adapters/pty"
28 "github.com/bnema/vev/internal/adapters/sessionwire"
29 "github.com/bnema/vev/internal/adapters/snapshot"
30 "github.com/bnema/vev/internal/domain"
31 "github.com/bnema/vev/internal/persist"
32 "github.com/bnema/vev/internal/ports"
33 "github.com/bnema/vev/internal/protocol"
34 "github.com/bnema/vev/internal/protocol/wire"
35 "github.com/bnema/vev/internal/usecase/daemon"
36 "github.com/bnema/vev/internal/usecase/recovery"
37)
38
39// These integration tests drive the real daemon over a real unix socket with a
40// real PTY and the real VT/renderer pipeline. They live in internal/app (which
41// is allowed to import adapters) because the layering guard forbids
42// internal/usecase from importing internal/adapters, even in tests.
43
44func discardLog() *slog.Logger {
45 return slog.New(slog.NewTextHandler(io.Discard, nil))
46}
47
48// startDaemon binds a real listener in a temp dir and serves in the background.
49// The returned served channel receives Serve's result exactly once.
50func startDaemon(t *testing.T, opts ...daemon.Option) (dir string, served <-chan error) {
51 t.Helper()
52 return startDaemonInDir(t, filepath.Join(t.TempDir(), "vev"), opts...)
53}
54
55func startDaemonInDir(t *testing.T, dir string, opts ...daemon.Option) (string, <-chan error) {
56 t.Helper()
57 ln, err := ipc.Listen(dir)
58 require.NoError(t, err)
59
60 d := daemon.New(pty.NewFactory(), clock.New(), discardLog(), opts...)
61 ctx, cancel := context.WithCancel(context.Background())
62 ch := make(chan error, 1)
63 go func() { ch <- d.Serve(ctx, sessionwire.NewServerListener(ln)) }()
64 t.Cleanup(cancel)
65 return dir, ch
66}
67
68// pump wires a background Recv pump to a raw transport so tests can await
69// envelopes with a timeout (wire.Transport has no read deadline).
70type pump struct{ ch chan wire.Envelope }
71
72func recvPump(tr wire.Transport) *pump {
73 p := &pump{ch: make(chan wire.Envelope, 128)}
74 go func() {
75 for {
76 f, err := tr.Recv()
77 if err != nil {
78 close(p.ch)
79 return
80 }
81 p.ch <- f
82 }
83 }()
84 return p
85}
86
87// typedPump wires a background ReceiveServer pump to a typed client connection
88// so tests can await semantic messages with a timeout.
89type typedPump struct{ ch chan protocol.ServerMessage }
90
91func recvTypedPump(conn ports.ClientConnection) *typedPump {
92 p := &typedPump{ch: make(chan protocol.ServerMessage, 128)}
93 go func() {
94 for {
95 message, err := conn.ReceiveServer()
96 if err != nil {
97 close(p.ch)
98 return
99 }
100 p.ch <- message
101 }
102 }()
103 return p
104}
105
106// attach dials, handshakes, and returns the typed client connection plus its
107// semantic message pump.
108func attach(t *testing.T, dir string, intent uint8, name string, sz domain.Size) (ports.ClientConnection, *typedPump) {
109 t.Helper()
110 return attachWithEnvironment(t, dir, intent, name, sz, nil)
111}
112
113func attachWithEnvironment(t *testing.T, dir string, intent uint8, name string, sz domain.Size, env []string) (ports.ClientConnection, *typedPump) {
114 t.Helper()
115 raw, err := ipc.DialContext(context.Background(), dir)
116 require.NoError(t, err)
117 conn := sessionwire.NewClientConnection(raw)
118 hello := protocol.Hello{Version: protocol.Version, Intent: intent, Name: name, Size: sz, TermEnv: "xterm-256color", TrueColor: true, Env: env}
119 require.NoError(t, conn.SendClient(hello))
120 p := recvTypedPump(conn)
121 select {
122 case message, ok := <-p.ch:
123 require.True(t, ok, "connection closed before welcome")
124 require.IsType(t, protocol.Welcome{}, message)
125 case <-time.After(3 * time.Second):
126 t.Fatal("timed out waiting for welcome")
127 }
128 return conn, p
129}
130
131func listRemoteSessions(t *testing.T, dir string) protocol.Sessions {
132 t.Helper()
133 raw, err := ipc.DialContext(context.Background(), dir)
134 require.NoError(t, err)
135 conn := sessionwire.NewClientConnection(raw)
136 defer func() { _ = conn.Close() }()
137 require.NoError(t, conn.SendClient(protocol.List{}))
138 message, err := conn.ReceiveServer()
139 require.NoError(t, err)
140 sessions, ok := message.(protocol.Sessions)
141 require.True(t, ok, "expected session listing, got %T", message)
142 return sessions
143}
144
145func killAll(dir string) error {
146 return requestKill(dir, protocol.KillAll)
147}
148
149func killDaemon(dir string) error {
150 return requestKill(dir, protocol.KillDaemon)
151}
152
153// nextKillRequestID allocates a unique nonzero correlation ID for one control
154// request, matching the production caller contract.
155var killRequestSeq atomic.Uint64
156
157func nextKillRequestID() uint64 { return killRequestSeq.Add(1) }
158
159// killNamedSession deletes one named live or stopped session over the control
160// connection, using the same wire scope the client sends for `kill -s NAME`.
161// It sends a unique RequestID and consumes the matching KillResult: a failed
162// outcome is an error, and a close without a result is never read as success.
163func killNamedSession(dir, name string) error {
164 return controlKill(dir, protocol.KillScope(protocol.KillSession), name)
165}
166
167func requestKill(dir string, scope protocol.KillScope) error {
168 return controlKill(dir, scope, "")
169}
170
171func controlKill(dir string, scope protocol.KillScope, name string) error {
172 raw, err := ipc.DialContext(context.Background(), dir)
173 if err != nil {
174 return err
175 }
176 conn := sessionwire.NewClientConnection(raw)
177 defer func() { _ = conn.Close() }()
178 requestID := nextKillRequestID()
179 if err := conn.SendClient(protocol.Kill{RequestID: requestID, Scope: scope, Name: name}); err != nil {
180 return err
181 }
182 result, err := awaitKillResult(conn, requestID, daemonStopTimeout)
183 if err != nil {
184 return err
185 }
186 switch result.Outcome {
187 case protocol.KillSucceeded:
188 return nil
189 case protocol.KillFailed:
190 return fmt.Errorf("kill failed: %s", result.Text)
191 default:
192 return fmt.Errorf("kill outcome unknown: %s", result.Text)
193 }
194}
195
196// awaitKillResult reads one correlated KillResult with a bounded timeout. A
197// close, a wrong type, or a mismatched RequestID is a definite error rather
198// than an inferred success, so a lost result is never reported as a kill.
199func awaitKillResult(conn ports.ClientConnection, requestID uint64, timeout time.Duration) (protocol.KillResult, error) {
200 type result struct {
201 message protocol.ServerMessage
202 err error
203 }
204 received := make(chan result, 1)
205 go func() {
206 message, err := conn.ReceiveServer()
207 received <- result{message: message, err: err}
208 }()
209
210 timer := time.NewTimer(timeout)
211 defer timer.Stop()
212 var reply result
213 select {
214 case reply = <-received:
215 case <-timer.C:
216 _ = conn.Close()
217 reply = <-received
218 return protocol.KillResult{}, errors.Join(context.DeadlineExceeded, reply.err)
219 }
220 if reply.err != nil {
221 return protocol.KillResult{}, reply.err
222 }
223 killResult, ok := reply.message.(protocol.KillResult)
224 if !ok {
225 return protocol.KillResult{}, fmt.Errorf("unexpected reply %T to kill", reply.message)
226 }
227 if killResult.RequestID != requestID {
228 return protocol.KillResult{}, fmt.Errorf("kill result request id %d, want %d", killResult.RequestID, requestID)
229 }
230 return killResult, nil
231}
232
233// blockingClientConnection parks ReceiveServer until Close, so a test can prove
234// the bounded await returns on timeout rather than hanging.
235type blockingClientConnection struct {
236 closed chan struct{}
237 once sync.Once
238}
239
240func (*blockingClientConnection) SendClient(protocol.ClientMessage) error { return nil }
241
242func (c *blockingClientConnection) ReceiveServer() (protocol.ServerMessage, error) {
243 <-c.closed
244 return nil, io.ErrClosedPipe
245}
246
247func (*blockingClientConnection) Capabilities() protocol.ConnectionCapabilities {
248 return protocol.ConnectionCapabilities{}
249}
250
251func (*blockingClientConnection) LinkState() ports.LinkState { return ports.LinkState(0) }
252func (*blockingClientConnection) LinkEvents() <-chan ports.LinkEvent { return nil }
253
254func (c *blockingClientConnection) Close() error {
255 c.once.Do(func() { close(c.closed) })
256 return nil
257}
258
259// TestAwaitKillResultTimesOut proves the bounded result await reports a timeout
260// (never a silent success) and closes the exact connection when no
261// KillResult arrives.
262func TestAwaitKillResultTimesOut(t *testing.T) {
263 conn := &blockingClientConnection{closed: make(chan struct{})}
264 _, err := awaitKillResult(conn, 1, 10*time.Millisecond)
265 require.ErrorIs(t, err, context.DeadlineExceeded)
266 require.ErrorIs(t, err, io.ErrClosedPipe)
267}
268
269// TestAwaitKillResultRejectsUncorrelatedReply proves a reply of the wrong type
270// or with a mismatched RequestID is a definite error, so a lost or stale result
271// is never mistaken for this request's success.
272func TestAwaitKillResultRejectsUncorrelatedReply(t *testing.T) {
273 tests := []struct {
274 name string
275 reply protocol.ServerMessage
276 }{
277 {name: "wrong type", reply: protocol.Pong{}},
278 {name: "wrong request id", reply: protocol.KillResult{RequestID: 9, Outcome: protocol.KillSucceeded}},
279 }
280 for _, tt := range tests {
281 t.Run(tt.name, func(t *testing.T) {
282 conn := &scriptedKillReplyConnection{reply: tt.reply}
283 _, err := awaitKillResult(conn, 1, time.Second)
284 require.Error(t, err)
285 })
286 }
287}
288
289// scriptedKillReplyConnection returns exactly one canned server message.
290type scriptedKillReplyConnection struct {
291 reply protocol.ServerMessage
292}
293
294func (*scriptedKillReplyConnection) SendClient(protocol.ClientMessage) error { return nil }
295
296func (c *scriptedKillReplyConnection) ReceiveServer() (protocol.ServerMessage, error) {
297 if c.reply == nil {
298 return nil, io.EOF
299 }
300 reply := c.reply
301 c.reply = nil
302 return reply, nil
303}
304
305func (*scriptedKillReplyConnection) Capabilities() protocol.ConnectionCapabilities {
306 return protocol.ConnectionCapabilities{}
307}
308
309func (*scriptedKillReplyConnection) LinkState() ports.LinkState { return ports.LinkState(0) }
310func (*scriptedKillReplyConnection) LinkEvents() <-chan ports.LinkEvent { return nil }
311func (*scriptedKillReplyConnection) Close() error { return nil }
312
313// mustHelloBytes encodes a minimal valid Hello for first-frame shape probes.
314func mustHelloBytes() []byte {
315 raw, err := sessionwire.EncodeClientMessage(protocol.Hello{
316 Version: protocol.Version, Size: domain.Size{Cols: 80, Rows: 24},
317 })
318 if err != nil {
319 panic(err)
320 }
321 return raw
322}
323
324func TestIntegration_MalformedCommandPreservesVersionAndRequestID(t *testing.T) {
325 versionMismatch, err := sessionwire.EncodeClientMessage(protocol.CommandRequest{
326 Version: protocol.Version + 1, RequestID: 42, Slug: "list-sessions",
327 })
328 require.NoError(t, err)
329 valid, err := sessionwire.EncodeClientMessage(protocol.CommandRequest{
330 Version: protocol.Version, RequestID: 43, Slug: "list-sessions",
331 })
332 require.NoError(t, err)
333
334 tests := []struct {
335 name string
336 payload []byte
337 want protocol.ServerMessage
338 }{
339 {
340 name: "incompatible version",
341 payload: versionMismatch,
342 want: &protocol.CommandResult{Outcome: protocol.CommandFailed, RequestID: 42, Code: protocol.ErrVersionMismatch, Text: "protocol version mismatch"},
343 },
344 {
345 // A single zero byte carries no recoverable envelope tag:
346 // the daemon answers the generic hello error.
347 name: "truncated version prefix",
348 payload: []byte{0},
349 want: &protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "expected hello"},
350 },
351 {
352 // Scan-level trailing garbage on a one-shot command
353 // connection carries no recoverable request: the daemon
354 // logs the rejection and closes without a typed refusal,
355 // so the client observes EOF.
356 name: "trailing garbage closes without refusal",
357 payload: append(append([]byte(nil), valid...), 0xff),
358 },
359 {
360 // A decodable Hello with trailing garbage fails semantic
361 // validation after scanning: it is neither a command nor a
362 // well-formed attach, so the daemon answers the hello
363 // refusal before closing.
364 name: "hello-shaped trailing garbage replies hello error",
365 payload: append(append([]byte(nil), mustHelloBytes()...), 0xff),
366 want: &protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "malformed hello"},
367 },
368 }
369 for _, tt := range tests {
370 t.Run(tt.name, func(t *testing.T) {
371 dir, _ := startDaemon(t)
372 raw, err := ipc.DialContext(context.Background(), dir)
373 require.NoError(t, err)
374 defer func() { _ = raw.Close() }()
375
376 preambleRequest, err := sessionwire.EncodePreambleRequestForTest()
377 require.NoError(t, err)
378 require.NoError(t, raw.Send(wire.Envelope{Payload: preambleRequest}))
379 response, err := raw.Recv()
380 require.NoError(t, err)
381 require.True(t, sessionwire.DecodePreambleAcceptanceForTest(response.Payload))
382
383 require.NoError(t, raw.Send(wire.Envelope{Payload: tt.payload}))
384 frame, err := raw.Recv()
385 if tt.want == nil {
386 require.ErrorIs(t, err, io.EOF)
387 return
388 }
389 require.NoError(t, err)
390 message, err := sessionwire.DecodeServerEnvelope(frame.Payload)
391 require.NoError(t, err)
392 switch want := tt.want.(type) {
393 case *protocol.CommandResult:
394 result, ok := message.(protocol.CommandResult)
395 require.True(t, ok, "expected command result, got %T", message)
396 require.Equal(t, *want, result)
397 case *protocol.ErrorMsg:
398 reply, ok := message.(protocol.ErrorMsg)
399 require.True(t, ok, "expected error reply, got %T", message)
400 require.Equal(t, *want, reply)
401 default:
402 t.Fatalf("unsupported want type %T", tt.want)
403 }
404 })
405 }
406}
407
408// awaitText decodes MsgOutput frames into a fresh VT screen and returns once
409// the reconstructed grid contains want.
410func awaitText(t *testing.T, p *typedPump, sz domain.Size, want string) {
411 t.Helper()
412 _ = awaitScreenText(t, p, sz, want)
413}
414
415// requireListedAfterRestart asserts that every wanted named session is listed,
416// in name order, as a healthy durable session. A restarted daemon restores
417// named sessions in the background, publishing each one atomically from the
418// stopped registry (SessionDown) to the live one (SessionUp), so a listing taken
419// right after startup legitimately observes either state per session. Only
420// SessionBroken, a missing name, or an unexpected extra name is a failure.
421func requireListedAfterRestart(t *testing.T, sessions []protocol.SessionInfo, names ...string) {
422 t.Helper()
423 got := make([]string, 0, len(sessions))
424 for _, info := range sessions {
425 got = append(got, info.Name)
426 require.Contains(t, []protocol.SessionState{protocol.SessionDown, protocol.SessionUp}, info.State,
427 "session %q must be stopped or restored, not broken", info.Name)
428 require.False(t, info.Ephemeral, "session %q must be a durable named session", info.Name)
429 require.False(t, info.Attached, "session %q has no client after restart", info.Name)
430 }
431 require.Equal(t, names, got)
432}
433
434// awaitDetached consumes typed messages until the daemon signals the session's
435// end with the expected Detached reason, ignoring interleaved Output frames.
436func awaitDetached(t *testing.T, p *typedPump, reason uint8) {
437 t.Helper()
438 timeout := time.After(5 * time.Second)
439 for {
440 select {
441 case message, ok := <-p.ch:
442 if !ok {
443 t.Fatal("connection closed before Detached")
444 }
445 if detached, ok := message.(protocol.Detached); ok {
446 require.Equal(t, reason, detached.Reason)
447 return
448 }
449 case <-timeout:
450 t.Fatal("timed out waiting for Detached")
451 }
452 }
453}
454
455// awaitScreenText is like awaitText, but returns the reconstructed screen text
456// at the point the wanted text appears so callers can make additional checks.
457func awaitScreenText(t *testing.T, p *typedPump, sz domain.Size, want string) string {
458 t.Helper()
459 screen := vt.NewScreen(sz.Cols, sz.Rows)
460 timeout := time.After(5 * time.Second)
461 for {
462 select {
463 case message, ok := <-p.ch:
464 if !ok {
465 t.Fatalf("connection closed before %q appeared; screen=%q", want, screenText(screen))
466 }
467 if output, ok := message.(protocol.Output); ok {
468 screen.Write(output.Data)
469 text := screenText(screen)
470 if strings.Contains(text, want) {
471 return text
472 }
473 }
474 case <-timeout:
475 t.Fatalf("timed out waiting for %q; screen=%q", want, screenText(screen))
476 }
477 }
478}
479
480func assertNoTextAfterInput(t *testing.T, p *typedPump, sz domain.Size, absent string) {
481 t.Helper()
482 screen := vt.NewScreen(sz.Cols, sz.Rows)
483 timeout := time.After(300 * time.Millisecond)
484 for {
485 select {
486 case message, ok := <-p.ch:
487 if !ok {
488 return
489 }
490 if output, ok := message.(protocol.Output); ok {
491 screen.Write(output.Data)
492 require.NotContains(t, screenText(screen), absent)
493 }
494 case <-timeout:
495 return
496 }
497 }
498}
499
500func screenText(s *vt.Screen) string {
501 var b strings.Builder
502 for y := range s.Rows() {
503 for x := range s.Columns() {
504 r := s.Cell(x, y).Rune
505 if r == 0 {
506 r = ' '
507 }
508 b.WriteRune(r)
509 }
510 }
511 return b.String()
512}
513
514func shellFixture(t *testing.T, label string) string {
515 t.Helper()
516 path := filepath.Join(t.TempDir(), "shell")
517 require.NoError(t, os.WriteFile(path, []byte("#!/bin/sh\nprintf 'SHELL_COMMAND="+label+"\\n'\nexec /bin/sh\n"), 0o700))
518 return path
519}
520
521func assertChildEnvironment(t *testing.T, tr ports.ClientConnection, p *typedPump, sz domain.Size, wantTestEnv, wantShell, wantRuntimeDir, wantWayland string) {
522 t.Helper()
523 command := "printf '\\033[2J\\033[H'; printf 'VEV_TEST_ENV=%s SHELL=%s XDG_RUNTIME_DIR=%s WAYLAND_DISPLAY=%s TERM=%s COLORTERM=%s TERM_PROGRAM=%s VEV_PREFIX=%.24s\\n' \"$VEV_TEST_ENV\" \"${SHELL##*/}\" \"$XDG_RUNTIME_DIR\" \"$WAYLAND_DISPLAY\" \"$TERM\" \"$COLORTERM\" \"$TERM_PROGRAM\" \"$VEV\"\n"
524 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte(command)}))
525 text := awaitScreenText(t, p, sz, "TERM_PROGRAM=vev")
526 // The concrete PTY adapter selects the profile from host terminfo. Exact
527 // direct/fallback behavior is covered by the adapter's hermetic tests.
528 require.Regexp(t, `TERM=xterm-(direct|256color)\s`, text)
529 for _, want := range []string{
530 "VEV_TEST_ENV=" + wantTestEnv,
531 "SHELL=" + filepath.Base(wantShell),
532 "XDG_RUNTIME_DIR=" + wantRuntimeDir,
533 "WAYLAND_DISPLAY=" + wantWayland,
534 "COLORTERM=truecolor",
535 "TERM_PROGRAM=vev",
536 "VEV_PREFIX=session=environment,tab=",
537 } {
538 require.Contains(t, text, want)
539 }
540}
541
542// displayShellFixture names each shell after the WAYLAND_DISPLAY it started
543// with, so a test can tell which session-bound environment a PTY received.
544func displayShellFixture(t *testing.T) string {
545 t.Helper()
546 path := filepath.Join(t.TempDir(), "shell")
547 require.NoError(t, os.WriteFile(path, []byte("#!/bin/sh\nprintf 'SHELL_COMMAND=%s\\n' \"${WAYLAND_DISPLAY#wayland-}\"\nexec /bin/sh\n"), 0o700))
548 return path
549}
550
551func TestIntegration_AttachEnvironmentRefreshesFuturePTYChildren(t *testing.T) {
552 sz := domain.Size{Cols: 80, Rows: 24}
553 dir, _ := startDaemon(t)
554 firstShell := displayShellFixture(t)
555 secondShell := shellFixture(t, "second")
556 firstEnv := []string{
557 "VEV_TEST_ENV=first", "SHELL=" + firstShell, "XDG_RUNTIME_DIR=/run/first", "WAYLAND_DISPLAY=wayland-first",
558 "TERM=client", "COLORTERM=client", "TERM_PROGRAM=client", "VEV=client",
559 }
560
561 tr1, p1 := attachWithEnvironment(t, dir, protocol.IntentNew, "environment", sz, firstEnv)
562 defer func() { _ = tr1.Close() }()
563 awaitText(t, p1, sz, "SHELL_COMMAND=first")
564 assertChildEnvironment(t, tr1, p1, sz, "first", firstShell, "/run/first", "wayland-first")
565
566 secondEnv := []string{
567 "VEV_TEST_ENV=second", "SHELL=" + secondShell, "XDG_RUNTIME_DIR=/run/second", "WAYLAND_DISPLAY=wayland-second",
568 "TERM=client", "COLORTERM=client", "TERM_PROGRAM=client", "VEV=client",
569 }
570 tr2, p2 := attachWithEnvironment(t, dir, protocol.IntentAttach, "environment", sz, secondEnv)
571 defer func() { _ = tr2.Close() }()
572
573 // The first shell was already running, so it retains its original environment.
574 assertChildEnvironment(t, tr2, p2, sz, "first", firstShell, "/run/first", "wayland-first")
575
576 require.NoError(t, tr2.SendClient(protocol.Input{Data: []byte("\x1b ")}))
577 awaitText(t, p2, sz, "Commands")
578 require.NoError(t, tr2.SendClient(protocol.Input{Data: []byte("CNT\r")}))
579 // The new tab keeps the creator's shell and ordinary variables; only the
580 // session-bound desktop variables follow the latest attach.
581 awaitText(t, p2, sz, "SHELL_COMMAND=second")
582 assertChildEnvironment(t, tr2, p2, sz, "first", firstShell, "/run/second", "wayland-second")
583}
584
585func TestIntegration_TwoAttachmentsReceiveSharedMutationPTYOutput(t *testing.T) {
586 sz := domain.Size{Cols: 80, Rows: 24}
587 dir, _ := startDaemon(t)
588 const sharedOutput = "SECOND_SHARED_PTY_OUTPUT"
589 // The session keeps its creator's SHELL, so one script serves both tabs and
590 // gates only the PTY opened after the second attach refreshed the display.
591 firstShell := filepath.Join(t.TempDir(), "shell")
592 require.NoError(t, os.WriteFile(firstShell, []byte("#!/bin/sh\nif [ \"$WAYLAND_DISPLAY\" = wayland-second ]; then\nprintf 'SHELL_COMMAND=second\\n'\nIFS= read -r _\nprintf '"+sharedOutput+"\\n'\nelse\nprintf 'SHELL_COMMAND=first\\n'\nfi\nexec /bin/sh\n"), 0o700))
593 secondShell := shellFixture(t, "unused")
594 firstEnv := []string{
595 "VEV_TEST_ENV=first", "SHELL=" + firstShell, "XDG_RUNTIME_DIR=/run/first", "WAYLAND_DISPLAY=wayland-first",
596 "TERM=client", "COLORTERM=client", "TERM_PROGRAM=client", "VEV=client",
597 }
598 tr1, p1 := attachWithEnvironment(t, dir, protocol.IntentNew, "environment", sz, firstEnv)
599 defer func() { _ = tr1.Close() }()
600 awaitText(t, p1, sz, "SHELL_COMMAND=first")
601 assertChildEnvironment(t, tr1, p1, sz, "first", firstShell, "/run/first", "wayland-first")
602
603 secondEnv := []string{
604 "VEV_TEST_ENV=second", "SHELL=" + secondShell, "XDG_RUNTIME_DIR=/run/second", "WAYLAND_DISPLAY=wayland-second",
605 "TERM=client", "COLORTERM=client", "TERM_PROGRAM=client", "VEV=client",
606 }
607 tr2, p2 := attachWithEnvironment(t, dir, protocol.IntentAttach, "environment", sz, secondEnv)
608 defer func() { _ = tr2.Close() }()
609 assertChildEnvironment(t, tr2, p2, sz, "first", firstShell, "/run/first", "wayland-first")
610
611 // The second attachment mutates shared tab state. Its newly opened PTY must
612 // also be rendered to that attachment, not only to the coordinator primary.
613 require.NoError(t, tr2.SendClient(protocol.Input{Data: []byte("\x1b ")}))
614 awaitText(t, p2, sz, "Commands")
615 require.NoError(t, tr2.SendClient(protocol.Input{Data: []byte("CNT\r")}))
616 awaitText(t, p2, sz, "SHELL_COMMAND=second")
617
618 // Each attachment owns its selected tab. Move the first attachment to the
619 // new PTY before releasing its gate, so the fresh PTY output must reach both
620 // views rather than only being recovered by a later first paint.
621 require.NoError(t, tr1.SendClient(protocol.Input{Data: []byte("\x1b2")}))
622 awaitText(t, p1, sz, "SHELL_COMMAND=second")
623 require.NoError(t, tr1.SendClient(protocol.Input{Data: []byte("go\n")}))
624 awaitText(t, p2, sz, sharedOutput)
625 awaitText(t, p1, sz, sharedOutput)
626}
627
628func TestIntegration_AttachFirstOutput(t *testing.T) {
629 sz := domain.Size{Cols: 80, Rows: 24}
630 dir, _ := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf HELLO; sleep 30"}))
631
632 tr, p := attach(t, dir, protocol.IntentEphemeral, "", sz)
633 defer func() { _ = tr.Close() }()
634
635 awaitText(t, p, sz, "HELLO")
636}
637
638func TestIntegration_InputRoundtrip(t *testing.T) {
639 sz := domain.Size{Cols: 80, Rows: 24}
640 dir, _ := startDaemon(t, daemon.WithShell("/bin/cat", nil))
641
642 tr, p := attach(t, dir, protocol.IntentEphemeral, "", sz)
643 defer func() { _ = tr.Close() }()
644
645 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("PINGPONG\n")}))
646 awaitText(t, p, sz, "PINGPONG")
647}
648
649func TestIntegration_CommandPaletteCreatesTab(t *testing.T) {
650 sz := domain.Size{Cols: 80, Rows: 24}
651 dir, _ := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf READY; sleep 30"}))
652
653 tr, p := attach(t, dir, protocol.IntentEphemeral, "", sz)
654 defer func() { _ = tr.Close() }()
655 awaitText(t, p, sz, "READY")
656
657 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("\x1b ")}))
658 awaitText(t, p, sz, "Commands")
659
660 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("CNT\r")}))
661 // Tab labels are enriched with the focused pane's title; with no process
662 // inspector wired in this test, that title falls back to the shell's
663 // basename ("sh").
664 text := awaitScreenText(t, p, sz, " 1 (sh) 2 (sh) ")
665 require.Contains(t, text, " 1 (sh) 2 (sh) ")
666}
667
668func TestIntegration_CommandPaletteRenamesEphemeralSession(t *testing.T) {
669 sz := domain.Size{Cols: 80, Rows: 24}
670 dir, _ := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf READY; sleep 30"}))
671
672 tr, p := attach(t, dir, protocol.IntentEphemeral, "", sz)
673 defer func() { _ = tr.Close() }()
674 text := awaitScreenText(t, p, sz, "READY")
675 require.Contains(t, text, " 0* ")
676
677 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("\x1b ")}))
678 awaitText(t, p, sz, "Commands")
679
680 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("RNS\r")}))
681 awaitText(t, p, sz, "Rename session")
682
683 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("\x7fwork\r")}))
684 text = awaitScreenText(t, p, sz, " work ")
685 require.NotContains(t, text, "work*")
686
687 listRaw, err := ipc.DialContext(context.Background(), dir)
688 require.NoError(t, err)
689 listTr := sessionwire.NewClientConnection(listRaw)
690 defer func() { _ = listTr.Close() }()
691 require.NoError(t, listTr.SendClient(protocol.List{}))
692 message, err := listTr.ReceiveServer()
693 require.NoError(t, err)
694 sessions, ok := message.(protocol.Sessions)
695 require.True(t, ok, "expected session listing, got %T", message)
696 require.NotEmpty(t, sessions.Sessions)
697 require.Equal(t, "work", sessions.Sessions[0].Name)
698 require.False(t, sessions.Sessions[0].Ephemeral)
699}
700
701func TestIntegration_AltCWithoutPaletteDoesNotCreateTab(t *testing.T) {
702 sz := domain.Size{Cols: 80, Rows: 24}
703 dir, _ := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf READY; sleep 30"}))
704
705 tr, p := attach(t, dir, protocol.IntentEphemeral, "", sz)
706 defer func() { _ = tr.Close() }()
707 text := awaitScreenText(t, p, sz, "READY")
708 require.NotContains(t, text, " 1 2 ")
709
710 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("\x1bc")}))
711 assertNoTextAfterInput(t, p, sz, " 1 2 ")
712}
713
714func TestIntegration_EphemeralSurvivesDetachAndReattaches(t *testing.T) {
715 sz := domain.Size{Cols: 80, Rows: 24}
716 dir, served := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf MARKER; sleep 30"}))
717
718 tr1, p1 := attach(t, dir, protocol.IntentEphemeral, "", sz)
719 awaitText(t, p1, sz, "MARKER")
720 require.NoError(t, tr1.Close())
721
722 select {
723 case err := <-served:
724 t.Fatalf("daemon shut down while an ephemeral session was alive: %v", err)
725 case <-time.After(300 * time.Millisecond):
726 }
727
728 tr2, p2 := attach(t, dir, protocol.IntentAttach, "0", sz)
729 defer func() { _ = tr2.Close() }()
730 awaitText(t, p2, sz, "MARKER")
731
732 // An ephemeral session's daemon is ended only by the explicit daemon stop.
733 require.NoError(t, killDaemon(dir))
734 select {
735 case err := <-served:
736 require.NoError(t, err)
737 case <-time.After(5 * time.Second):
738 t.Fatal("daemon did not stop after a daemon stop request")
739 }
740}
741
742func TestIntegration_EphemeralNotListedAfterDaemonRestart(t *testing.T) {
743 sz := domain.Size{Cols: 80, Rows: 24}
744 stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
745 t.Setenv("XDG_STATE_HOME", stateRoot)
746 t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
747 dir := ipc.SocketDir()
748
749 _, served := startDaemonInDir(t, dir, daemon.WithShell("/bin/sh", []string{"-c", "printf TEMP; sleep 30"}))
750 tr, p := attach(t, dir, protocol.IntentEphemeral, "", sz)
751 awaitText(t, p, sz, "TEMP")
752 require.NoError(t, tr.Close())
753
754 // An ephemeral session lives only while its daemon does; the explicit daemon
755 // stop over the typed control wire ends that daemon.
756 require.NoError(t, killDaemon(dir))
757 select {
758 case err := <-served:
759 require.NoError(t, err)
760 case <-time.After(5 * time.Second):
761 t.Fatal("daemon did not stop after a daemon stop request")
762 }
763
764 _, served2 := startDaemonInDir(t, dir, daemon.WithShell("/bin/sh", []string{"-c", "sleep 30"}))
765 sessions := listRemoteSessions(t, dir)
766 require.Empty(t, sessions.Sessions)
767
768 require.NoError(t, killDaemon(dir))
769 select {
770 case err := <-served2:
771 require.NoError(t, err)
772 case <-time.After(5 * time.Second):
773 t.Fatal("restarted daemon did not stop after a daemon stop request")
774 }
775}
776
777func TestIntegration_KillDaemonPreservesMultipleNamedSessions(t *testing.T) {
778 sz := domain.Size{Cols: 80, Rows: 24}
779 stateDir := filepath.Join(t.TempDir(), "state")
780 dir := filepath.Join(t.TempDir(), "runtime")
781 repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
782
783 start := func() <-chan error {
784 opened, err := persist.OpenOrCreate(stateDir)
785 require.NoError(t, err)
786 coordinator := recovery.NewCoordinator(opened.Catalogue, repository, rand.Reader)
787 _, served := startDaemonInDir(t, dir,
788 daemon.WithShell("/bin/sh", []string{"-c", "sleep 30"}),
789 daemon.WithCatalogue(opened.Catalogue, opened.Records),
790 daemon.WithSnapshotRepository(repository),
791 daemon.WithRecoveryCoordinator(coordinator),
792 )
793 return served
794 }
795
796 served := start()
797 first, _ := attach(t, dir, protocol.IntentNew, "alpha", sz)
798 require.NoError(t, first.Close())
799 second, _ := attach(t, dir, protocol.IntentNew, "beta", sz)
800 require.NoError(t, second.Close())
801
802 require.NoError(t, killDaemon(dir))
803 select {
804 case err := <-served:
805 require.NoError(t, err)
806 case <-time.After(5 * time.Second):
807 t.Fatal("daemon did not stop")
808 }
809
810 served = start()
811 requireListedAfterRestart(t, listRemoteSessions(t, dir).Sessions, "alpha", "beta")
812
813 // Kill-all purges every durable record but leaves this daemon serving.
814 require.NoError(t, killAll(dir))
815 require.Empty(t, listRemoteSessions(t, dir).Sessions)
816 select {
817 case err := <-served:
818 t.Fatalf("kill-all stopped the restarted daemon: %v", err)
819 default:
820 }
821 require.NoError(t, killDaemon(dir))
822 select {
823 case err := <-served:
824 require.NoError(t, err)
825 case <-time.After(5 * time.Second):
826 t.Fatal("restarted daemon did not stop after a daemon stop request")
827 }
828
829 served = start()
830 require.Empty(t, listRemoteSessions(t, dir).Sessions)
831 require.NoError(t, killDaemon(dir))
832 select {
833 case err := <-served:
834 require.NoError(t, err)
835 case <-time.After(5 * time.Second):
836 t.Fatal("empty daemon did not stop")
837 }
838}
839
840func TestIntegration_NamedSurvivesReattach(t *testing.T) {
841 sz := domain.Size{Cols: 80, Rows: 24}
842 dir, served := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf MARKER; sleep 30"}))
843
844 tr1, p1 := attach(t, dir, protocol.IntentNew, "work", sz)
845 awaitText(t, p1, sz, "MARKER")
846
847 // Detach: a named session must survive.
848 require.NoError(t, tr1.Close())
849
850 // Daemon must still be running (session alive).
851 select {
852 case err := <-served:
853 t.Fatalf("daemon shut down while a named session was alive: %v", err)
854 case <-time.After(300 * time.Millisecond):
855 }
856
857 // Re-attach: the first paint must reproduce the retained screen state.
858 tr2, p2 := attach(t, dir, protocol.IntentAttach, "work", sz)
859 defer func() { _ = tr2.Close() }()
860 awaitText(t, p2, sz, "MARKER")
861}
862
863// TestIntegration_NamedSessionLastRemovalKeepsDaemonServing is the P4.1
864// app-level contract (P4.1 review GO-003): a named session killed as the last
865// registry entry is removed without stopping Serve, and the same daemon process
866// accepts the same name again. It runs over the real socket, PTY, and
867// catalogue/recovery/snapshot persistence the production composition installs.
868// A named session's final live removal is a durable purge, so the re-attached
869// name carries a fresh persisted incarnation rather than resuming the deleted
870// one.
871func TestIntegration_NamedSessionLastRemovalKeepsDaemonServing(t *testing.T) {
872 sz := domain.Size{Cols: 80, Rows: 24}
873 stateDir := filepath.Join(t.TempDir(), "state")
874 dir := filepath.Join(t.TempDir(), "runtime")
875 repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
876 opened, err := persist.OpenOrCreate(stateDir)
877 require.NoError(t, err)
878 coordinator := recovery.NewCoordinator(opened.Catalogue, repository, rand.Reader)
879
880 _, served := startDaemonInDir(t, dir,
881 daemon.WithShell("/bin/cat", nil),
882 daemon.WithCatalogue(opened.Catalogue, opened.Records),
883 daemon.WithSnapshotRepository(repository),
884 daemon.WithRecoveryCoordinator(coordinator),
885 )
886
887 first, p1 := attach(t, dir, protocol.IntentNew, "work", sz)
888 require.NoError(t, first.SendClient(protocol.Input{Data: []byte("ping\n")}))
889 awaitText(t, p1, sz, "ping")
890 firstRecord, ok, err := opened.Catalogue.Record("work")
891 require.NoError(t, err)
892 require.True(t, ok, "a live named session must own a durable record")
893
894 // End it as the last session: the registry empties, but that is not daemon
895 // shutdown in the P4.1 model.
896 require.NoError(t, killNamedSession(dir, "work"))
897 awaitDetached(t, p1, protocol.ReasonSessionKilled)
898 require.Eventually(t, func() bool { return len(listRemoteSessions(t, dir).Sessions) == 0 },
899 5*time.Second, 10*time.Millisecond, "session was not removed")
900 select {
901 case err := <-served:
902 t.Fatalf("Serve stopped after the last session was removed: %v", err)
903 default:
904 }
905 _, ok, err = opened.Catalogue.Record("work")
906 require.NoError(t, err)
907 require.False(t, ok, "final live removal purges the durable record")
908
909 // The same daemon process accepts the same name again and can still serve a
910 // fresh terminal operation.
911 second, p2 := attach(t, dir, protocol.IntentNew, "work", sz)
912 require.NoError(t, second.SendClient(protocol.Input{Data: []byte("pong\n")}))
913 awaitText(t, p2, sz, "pong")
914 sessions := listRemoteSessions(t, dir).Sessions
915 require.Len(t, sessions, 1)
916 require.Equal(t, "work", sessions[0].Name)
917 require.Equal(t, protocol.SessionUp, sessions[0].State)
918 secondRecord, ok, err := opened.Catalogue.Record("work")
919 require.NoError(t, err)
920 require.True(t, ok, "re-attached named session must persist its identity")
921 require.NotEqual(t, firstRecord.IncarnationID, secondRecord.IncarnationID,
922 "a re-created name must not reuse the purged incarnation")
923
924 require.NoError(t, second.Close())
925 require.NoError(t, killDaemon(dir))
926 select {
927 case err := <-served:
928 require.NoError(t, err)
929 case <-time.After(5 * time.Second):
930 t.Fatal("daemon did not stop after a daemon stop request")
931 }
932}
933
934// TestIntegration_NamedSessionRestoresPersistedIdentity proves a stopped named
935// session is restored inside the daemon process that is already serving it,
936// keeping its exact persisted identity. The seed daemon stops while holding the
937// session as its only registry entry, which preserves it as stopped authority;
938// the daemon under test then resumes it over a real socket with the production
939// catalogue/recovery/snapshot options and serves a fresh operation.
940func TestIntegration_NamedSessionRestoresPersistedIdentity(t *testing.T) {
941 sz := domain.Size{Cols: 80, Rows: 24}
942 stateDir := filepath.Join(t.TempDir(), "state")
943 repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
944 name := publishRestorableCheckpoint(t, stateDir, repository)
945
946 opened, err := persist.OpenOrCreate(stateDir)
947 require.NoError(t, err)
948 stopped, ok, err := opened.Catalogue.Record(name)
949 require.NoError(t, err)
950 require.True(t, ok, "stopped named session must keep its durable record")
951 require.NotNil(t, stopped.Committed, "stopped named session must keep its committed checkpoint")
952
953 dir := filepath.Join(t.TempDir(), "runtime")
954 coordinator := recovery.NewCoordinator(opened.Catalogue, repository, rand.Reader)
955 _, served := startDaemonInDir(t, dir,
956 daemon.WithShell("/bin/cat", nil),
957 daemon.WithCatalogue(opened.Catalogue, opened.Records),
958 daemon.WithSnapshotRepository(repository),
959 daemon.WithRecoveryCoordinator(coordinator),
960 )
961 requireListedAfterRestart(t, listRemoteSessions(t, dir).Sessions, name)
962 select {
963 case err := <-served:
964 t.Fatalf("Serve stopped before the restore: %v", err)
965 default:
966 }
967
968 conn, p := attach(t, dir, protocol.IntentAttach, name, sz)
969 require.NoError(t, conn.SendClient(protocol.Input{Data: []byte("revived\n")}))
970 awaitText(t, p, sz, "revived")
971 require.Eventually(t, func() bool {
972 sessions := listRemoteSessions(t, dir).Sessions
973 return len(sessions) == 1 && sessions[0].State == protocol.SessionUp
974 }, 5*time.Second, 10*time.Millisecond, "named session was not restored")
975
976 restored, ok, err := opened.Catalogue.Record(name)
977 require.NoError(t, err)
978 require.True(t, ok)
979 require.Equal(t, stopped.IncarnationID, restored.IncarnationID,
980 "restore must retain the persisted incarnation")
981 require.Equal(t, stopped.CreatedAt, restored.CreatedAt,
982 "restore must retain the persisted lifecycle timestamp")
983
984 require.NoError(t, conn.Close())
985 select {
986 case err := <-served:
987 t.Fatalf("Serve stopped while the restored session was live: %v", err)
988 default:
989 }
990 require.NoError(t, killDaemon(dir))
991 select {
992 case err := <-served:
993 require.NoError(t, err)
994 case <-time.After(5 * time.Second):
995 t.Fatal("daemon did not stop after a daemon stop request")
996 }
997}
998
999func TestMultipleClientsOneLifecycleOwner(t *testing.T) {
1000 dir := filepath.Join(t.TempDir(), "runtime")
1001 owner, err := lifecycle.TryAcquire(dir)
1002 require.NoError(t, err)
1003
1004 ready := make(chan struct{})
1005 transport := &integrationTransport{}
1006 dial := func(context.Context, string) (wire.Transport, error) {
1007 select {
1008 case <-ready:
1009 return transport, nil
1010 default:
1011 return nil, os.ErrNotExist
1012 }
1013 }
1014 var spawns atomic.Int32
1015 cfg := backoffConfig{initial: time.Millisecond, max: 2 * time.Millisecond, total: time.Second}
1016
1017 const clients = 4
1018 errs := make(chan error, clients)
1019 var wg sync.WaitGroup
1020 for range clients {
1021 wg.Go(func() {
1022 _, err := ensureDaemonWithLifecycle(context.Background(), dir, dial, func() error {
1023 spawns.Add(1)
1024 return nil
1025 }, cfg)
1026 errs <- err
1027 })
1028 }
1029 require.Never(t, func() bool { return spawns.Load() != 0 }, 50*time.Millisecond, time.Millisecond)
1030 close(ready)
1031 require.NoError(t, owner.Release())
1032 wg.Wait()
1033 close(errs)
1034 for err := range errs {
1035 require.NoError(t, err)
1036 }
1037 require.Zero(t, spawns.Load())
1038}
1039
1040func TestLifecycleOwnershipOutlivesMaintenanceWriter(t *testing.T) {
1041 runtimeDir := filepath.Join(t.TempDir(), "runtime")
1042 stateDir := filepath.Join(t.TempDir(), "state")
1043 catalogue := &lifecycleObservedCatalogue{
1044 Catalogue: newTestPersister(t, stateDir),
1045 closed: make(chan struct{}),
1046 }
1047 require.NoError(t, catalogue.Create(domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}}))
1048 repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
1049 shutdownClock := newLifecycleShutdownClock(t)
1050 maintenanceRepository := &lifecycleBlockingMaintenanceRepository{
1051 SnapshotRepository: repository,
1052 entered: make(chan struct{}),
1053 release: make(chan struct{}),
1054 returned: make(chan struct{}),
1055 }
1056 ctx, cancel := context.WithCancel(context.Background())
1057 t.Cleanup(cancel)
1058 var releaseOnce sync.Once
1059 releaseMaintenance := func() { releaseOnce.Do(func() { close(maintenanceRepository.release) }) }
1060 t.Cleanup(releaseMaintenance)
1061
1062 callbackReturned := make(chan struct{})
1063 wrapperReturned := make(chan error, 1)
1064 go func() {
1065 wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
1066 defer close(callbackReturned)
1067 d := daemon.New(
1068 pty.NewFactory(),
1069 shutdownClock,
1070 discardLog(),
1071 daemon.WithSnapshotRepository(repository),
1072 daemon.WithRecoveryCoordinator(recovery.NewCoordinator(catalogue, maintenanceRepository, nil)),
1073 daemon.WithSnapshotGarbageCollection(),
1074 daemon.WithCatalogue(catalogue, nil),
1075 )
1076 listener, err := ipc.Listen(runtimeDir)
1077 if err != nil {
1078 return err
1079 }
1080 return d.Serve(ctx, sessionwire.NewServerListener(listener))
1081 })
1082 }()
1083
1084 awaitLifecycleStage(t, maintenanceRepository.entered, "background maintenance repository call")
1085 cancel()
1086 select {
1087 case <-callbackReturned:
1088 t.Fatal("Serve callback returned after shutdown deadline while maintenance was blocked")
1089 case <-time.After(100 * time.Millisecond):
1090 }
1091 assertLifecycleStagePending(t, catalogue.closed, "catalogue close while maintenance is blocked")
1092 assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return while maintenance is blocked")
1093 _, err := lifecycle.TryAcquire(runtimeDir)
1094 require.ErrorIs(t, err, lifecycle.ErrBusy)
1095
1096 releaseMaintenance()
1097 awaitLifecycleStage(t, maintenanceRepository.returned, "maintenance repository return")
1098 awaitLifecycleStage(t, catalogue.closed, "catalogue close after maintenance")
1099 awaitLifecycleStage(t, callbackReturned, "Serve callback return after catalogue close")
1100 require.NoError(t, awaitLifecycleResult(t, wrapperReturned, "lifecycle wrapper return"))
1101 owner, err := lifecycle.TryAcquire(runtimeDir)
1102 require.NoError(t, err)
1103 require.NoError(t, owner.Release())
1104}
1105
1106type lifecycleBlockingMaintenanceRepository struct {
1107 ports.SnapshotRepository
1108 entered chan struct{}
1109 release chan struct{}
1110 returned chan struct{}
1111 enterOnce sync.Once
1112 returnOnce sync.Once
1113}
1114
1115func (r *lifecycleBlockingMaintenanceRepository) SnapshotIncarnations(context.Context) ([]domain.IncarnationID, error) {
1116 return []domain.IncarnationID{{1}}, nil
1117}
1118
1119func (r *lifecycleBlockingMaintenanceRepository) CollectIncarnationGarbage(context.Context, domain.IncarnationID, *domain.CheckpointRef) error {
1120 r.enterOnce.Do(func() { close(r.entered) })
1121 <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
1122 r.returnOnce.Do(func() { close(r.returned) })
1123 return nil
1124}
1125
1126type lifecycleObservedCatalogue struct {
1127 ports.Catalogue
1128 closed chan struct{}
1129 closeOnce sync.Once
1130}
1131
1132func (c *lifecycleObservedCatalogue) Close() error {
1133 err := c.Catalogue.Close()
1134 c.closeOnce.Do(func() { close(c.closed) })
1135 return err
1136}
1137
1138func TestLifecycleOwnershipOutlivesSnapshotWriter(t *testing.T) {
1139 result := runBlockedSnapshotWriterShutdown(t)
1140 assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")
1141 assertLifecycleResultPending(t, result.wrapperReturned, "lifecycle wrapper return while snapshot writer is blocked")
1142 _, err := lifecycle.TryAcquire(result.runtimeDir)
1143 require.ErrorIs(t, err, lifecycle.ErrBusy)
1144
1145 result.releaseWriter()
1146 awaitLifecycleStage(t, result.publishReturned, "snapshot writer return")
1147 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
1148 owner, err := lifecycle.TryAcquire(result.runtimeDir)
1149 require.NoError(t, err)
1150 require.NoError(t, owner.Release())
1151}
1152
1153func TestLifecycleCallbackWaitsForEveryWriter(t *testing.T) {
1154 result := runBlockedSnapshotWriterShutdown(t)
1155 assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")
1156
1157 result.releaseWriter()
1158 awaitLifecycleStage(t, result.publishReturned, "snapshot writer return")
1159 awaitLifecycleStage(t, result.callbackReturned, "Serve callback return after snapshot writer")
1160 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
1161}
1162
1163func TestFinalCheckpointTimeoutKeepsOwner(t *testing.T) {
1164 result := runBlockedSnapshotWriterShutdown(t)
1165 assertLifecycleResultPending(t, result.wrapperReturned, "ownership release after checkpoint timeout with writer alive")
1166 _, err := lifecycle.TryAcquire(result.runtimeDir)
1167 require.ErrorIs(t, err, lifecycle.ErrBusy)
1168
1169 result.releaseWriter()
1170 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
1171}
1172
1173// Startup restoration repairs HEADs, promotes fallbacks, and replaces catalogue
1174// records. Lifecycle ownership must therefore outlive it: releasing the flock
1175// while a restoration repository call is still running would let a second
1176// daemon mutate the same durable state.
1177func TestLifecycleOwnershipOutlivesRestorationWriter(t *testing.T) {
1178 stateDir := filepath.Join(t.TempDir(), "state")
1179 repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
1180 checkpointed := publishRestorableCheckpoint(t, stateDir, repository)
1181
1182 runtimeDir := filepath.Join(t.TempDir(), "runtime")
1183 blocking := &lifecycleBlockingRestore{
1184 SnapshotRepository: repository,
1185 entered: make(chan struct{}),
1186 release: make(chan struct{}),
1187 returned: make(chan struct{}),
1188 }
1189 var releaseOnce sync.Once
1190 releaseRestore := func() { releaseOnce.Do(func() { close(blocking.release) }) }
1191 t.Cleanup(releaseRestore)
1192
1193 shutdownClock := newLifecycleShutdownClock(t)
1194 ctx, cancel := context.WithCancel(context.Background())
1195 t.Cleanup(cancel)
1196 listenerReady := make(chan struct{})
1197 callbackReturned := make(chan struct{})
1198 wrapperReturned := make(chan error, 1)
1199 go func() {
1200 wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
1201 defer close(callbackReturned)
1202 opened, err := persist.OpenOrCreate(stateDir)
1203 if err != nil {
1204 return err
1205 }
1206 listener, err := ipc.Listen(runtimeDir)
1207 if err != nil {
1208 return errors.Join(err, opened.Catalogue.Close())
1209 }
1210 close(listenerReady)
1211 d := daemon.New(
1212 pty.NewFactory(),
1213 shutdownClock,
1214 discardLog(),
1215 daemon.WithShell("/bin/cat", nil),
1216 daemon.WithCatalogue(opened.Catalogue, opened.Records),
1217 daemon.WithSnapshotRepository(blocking),
1218 )
1219 return d.Serve(ctx, sessionwire.NewServerListener(listener))
1220 })
1221 }()
1222
1223 awaitLifecycleStage(t, listenerReady, "daemon listener")
1224 awaitLifecycleStage(t, blocking.entered, "restoration repository call")
1225 require.Equal(t, checkpointed, blocking.repairedName(), "restoration must target the checkpointed session")
1226
1227 cancel()
1228 shutdownClock.nextTimer(t).fire()
1229 select {
1230 case <-callbackReturned:
1231 t.Fatal("Serve callback returned after shutdown deadline while restoration was blocked")
1232 case <-time.After(100 * time.Millisecond):
1233 }
1234 assertLifecycleStagePending(t, callbackReturned, "Serve callback return while restoration is blocked")
1235 assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return while restoration is blocked")
1236 _, err := lifecycle.TryAcquire(runtimeDir)
1237 require.ErrorIs(t, err, lifecycle.ErrBusy)
1238
1239 releaseRestore()
1240 awaitLifecycleStage(t, blocking.returned, "restoration repository return")
1241 awaitLifecycleStage(t, callbackReturned, "Serve callback return after restoration")
1242 require.NoError(t, awaitLifecycleResult(t, wrapperReturned, "lifecycle wrapper return"))
1243 owner, err := lifecycle.TryAcquire(runtimeDir)
1244 require.NoError(t, err)
1245 require.NoError(t, owner.Release())
1246}
1247
1248// publishRestorableCheckpoint runs a complete daemon over the real catalogue and
1249// repository until one named session owns a committed checkpoint, then shuts it
1250// down. The returned name is restorable by any later daemon on the same state.
1251func publishRestorableCheckpoint(t *testing.T, stateDir string, repository *snapshot.Repository) string {
1252 t.Helper()
1253 const name = "restorable"
1254 runtimeDir := filepath.Join(t.TempDir(), "seed-runtime")
1255 ctx, cancel := context.WithCancel(context.Background())
1256 defer cancel()
1257
1258 opened, err := persist.OpenOrCreate(stateDir)
1259 require.NoError(t, err)
1260 coordinator := recovery.NewCoordinator(opened.Catalogue, repository, rand.Reader)
1261 listener, err := ipc.Listen(runtimeDir)
1262 require.NoError(t, err)
1263 d := daemon.New(
1264 pty.NewFactory(),
1265 clock.New(),
1266 discardLog(),
1267 daemon.WithShell("/bin/cat", nil),
1268 daemon.WithCatalogue(opened.Catalogue, opened.Records),
1269 daemon.WithSnapshotRepository(repository),
1270 daemon.WithRecoveryCoordinator(coordinator),
1271 )
1272 served := make(chan error, 1)
1273 go func() { served <- d.Serve(ctx, sessionwire.NewServerListener(listener)) }()
1274
1275 tr, _ := attach(t, runtimeDir, protocol.IntentNew, name, domain.Size{Cols: 80, Rows: 24})
1276 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("checkpoint me\n")}))
1277 require.Eventually(t, func() bool {
1278 record, ok, _ := opened.Catalogue.Record(name)
1279 return ok && record.Committed != nil && record.DegradedReason == ""
1280 }, 5*time.Second, 10*time.Millisecond, "session never committed a checkpoint")
1281 require.NoError(t, tr.Close())
1282
1283 cancel()
1284 require.NoError(t, awaitLifecycleResult(t, served, "seed daemon shutdown"))
1285 return name
1286}
1287
1288type lifecycleBlockingRestore struct {
1289 ports.SnapshotRepository
1290 entered chan struct{}
1291 release chan struct{}
1292 returned chan struct{}
1293 enterOnce sync.Once
1294 returnOnce sync.Once
1295 mu sync.Mutex
1296 repaired string
1297}
1298
1299func (r *lifecycleBlockingRestore) LoadCheckpoint(ctx context.Context, id domain.IncarnationID, name string, ref domain.CheckpointRef) (ports.SnapshotGeneration, error) {
1300 r.mu.Lock()
1301 r.repaired = name
1302 r.mu.Unlock()
1303 return r.SnapshotRepository.LoadCheckpoint(ctx, id, name, ref)
1304}
1305
1306func (r *lifecycleBlockingRestore) ReconcileCheckpoint(ctx context.Context, id domain.IncarnationID, ref domain.CheckpointRef) error {
1307 r.enterOnce.Do(func() { close(r.entered) })
1308 <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
1309 r.returnOnce.Do(func() { close(r.returned) })
1310 return r.SnapshotRepository.ReconcileCheckpoint(ctx, id, ref)
1311}
1312
1313func (r *lifecycleBlockingRestore) repairedName() string {
1314 r.mu.Lock()
1315 defer r.mu.Unlock()
1316 return r.repaired
1317}
1318
1319type blockedSnapshotWriterShutdown struct {
1320 runtimeDir string
1321 callbackReturned <-chan struct{}
1322 wrapperReturned <-chan error
1323 publishReturned <-chan struct{}
1324 releaseWriter func()
1325}
1326
1327func runBlockedSnapshotWriterShutdown(t *testing.T) blockedSnapshotWriterShutdown {
1328 t.Helper()
1329 runtimeDir := filepath.Join(t.TempDir(), "runtime")
1330 stateDir := filepath.Join(t.TempDir(), "state")
1331 shutdownClock := newLifecycleShutdownClock(t)
1332 repository := newLifecycleBlockingRepository()
1333 ctx, cancel := context.WithCancel(context.Background())
1334 t.Cleanup(cancel)
1335
1336 listenerReady := make(chan struct{})
1337 listenerClosed := make(chan struct{})
1338 callbackReturned := make(chan struct{})
1339 wrapperReturned := make(chan error, 1)
1340 go func() {
1341 wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
1342 defer close(callbackReturned)
1343 listener, err := ipc.Listen(runtimeDir)
1344 if err != nil {
1345 return err
1346 }
1347 observed := &lifecycleObservedListener{Listener: listener, closed: listenerClosed}
1348 close(listenerReady)
1349 d := daemon.New(
1350 pty.NewFactory(),
1351 shutdownClock,
1352 discardLog(),
1353 daemon.WithShell("/bin/cat", nil),
1354 daemon.WithSnapshotRepository(repository),
1355 )
1356 return d.Serve(ctx, sessionwire.NewServerListener(observed))
1357 })
1358 }()
1359
1360 awaitLifecycleStage(t, listenerReady, "daemon listener")
1361 tr, _ := attach(t, runtimeDir, protocol.IntentNew, "work", domain.Size{Cols: 80, Rows: 24})
1362 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("dirty state\n")}))
1363 awaitLifecycleStage(t, repository.entered, "snapshot publication")
1364 require.NoError(t, tr.Close())
1365 require.Eventually(t, func() bool {
1366 sessions := listRemoteSessions(t, runtimeDir)
1367 return len(sessions.Sessions) == 1 && !sessions.Sessions[0].Attached
1368 }, time.Second, 10*time.Millisecond)
1369 cancel()
1370 awaitLifecycleStage(t, listenerClosed, "listener close")
1371
1372 shutdownClock.nextTimer(t).fire()
1373 select {
1374 case <-callbackReturned:
1375 t.Fatal("Serve callback returned after checkpoint timeout while snapshot writer was blocked")
1376 case <-time.After(100 * time.Millisecond):
1377 }
1378 assertLifecycleStagePending(t, callbackReturned, "Serve callback return after checkpoint timeout")
1379 assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return after checkpoint timeout")
1380
1381 var releaseOnce sync.Once
1382 releaseWriter := func() { releaseOnce.Do(func() { close(repository.release) }) }
1383 t.Cleanup(releaseWriter)
1384 return blockedSnapshotWriterShutdown{
1385 runtimeDir: runtimeDir,
1386 callbackReturned: callbackReturned,
1387 wrapperReturned: wrapperReturned,
1388 publishReturned: repository.returned,
1389 releaseWriter: releaseWriter,
1390 }
1391}
1392
1393type lifecycleBlockingRepository struct {
1394 ports.SnapshotRepository
1395 entered chan struct{}
1396 release chan struct{}
1397 returned chan struct{}
1398 enterOnce sync.Once
1399 returnOnce sync.Once
1400}
1401
1402func newLifecycleBlockingRepository() *lifecycleBlockingRepository {
1403 return &lifecycleBlockingRepository{
1404 entered: make(chan struct{}),
1405 release: make(chan struct{}),
1406 returned: make(chan struct{}),
1407 }
1408}
1409
1410func (r *lifecycleBlockingRepository) Publish(context.Context, ports.SnapshotPublication) error {
1411 r.enterOnce.Do(func() { close(r.entered) })
1412 <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
1413 r.returnOnce.Do(func() { close(r.returned) })
1414 return nil
1415}
1416
1417type lifecycleShutdownClock struct {
1418 t *testing.T
1419 base ports.Clock
1420 timers chan *lifecycleManualTimer
1421}
1422
1423func newLifecycleShutdownClock(t *testing.T) *lifecycleShutdownClock {
1424 t.Helper()
1425 return &lifecycleShutdownClock{t: t, base: clock.New(), timers: make(chan *lifecycleManualTimer, 4)}
1426}
1427
1428func (c *lifecycleShutdownClock) Now() time.Time { return c.base.Now() }
1429func (c *lifecycleShutdownClock) NewTimer(delay time.Duration) ports.Timer {
1430 if delay != daemon.SnapshotShutdownTimeout() {
1431 return c.base.NewTimer(delay)
1432 }
1433 timer := &lifecycleManualTimer{t: c.t, ch: make(chan time.Time, 1)}
1434 select {
1435 case c.timers <- timer:
1436 default:
1437 c.t.Errorf("unexpected additional snapshot shutdown timer with delay %s", delay)
1438 timer.fire()
1439 }
1440 return timer
1441}
1442
1443func (c *lifecycleShutdownClock) nextTimer(t *testing.T) *lifecycleManualTimer {
1444 t.Helper()
1445 select {
1446 case timer := <-c.timers:
1447 return timer
1448 case <-time.After(3 * time.Second):
1449 t.Fatal("timed out waiting for shutdown deadline timer")
1450 return nil
1451 }
1452}
1453
1454type lifecycleManualTimer struct {
1455 t *testing.T
1456 ch chan time.Time
1457 stopped atomic.Bool
1458}
1459
1460func (t *lifecycleManualTimer) C() <-chan time.Time { return t.ch }
1461func (t *lifecycleManualTimer) Reset(delay time.Duration) bool {
1462 if delay != daemon.SnapshotShutdownTimeout() {
1463 t.t.Errorf("snapshot shutdown timer reset with delay %s, want %s", delay, daemon.SnapshotShutdownTimeout())
1464 }
1465 t.stopped.Store(false)
1466 return false
1467}
1468func (t *lifecycleManualTimer) Stop() bool { return !t.stopped.Swap(true) }
1469func (t *lifecycleManualTimer) fire() {
1470 if t.stopped.Load() {
1471 return
1472 }
1473 select {
1474 case t.ch <- time.Now():
1475 default:
1476 }
1477}
1478
1479func TestLifecycleManualTimerRepeatedFireIsNonBlocking(t *testing.T) {
1480 timer := &lifecycleManualTimer{t: t, ch: make(chan time.Time, 1)}
1481 timer.fire()
1482
1483 returned := make(chan struct{})
1484 go func() {
1485 timer.fire()
1486 close(returned)
1487 }()
1488
1489 select {
1490 case <-returned:
1491 case <-time.After(time.Second):
1492 t.Fatal("repeated manual timer fire blocked on a pending tick")
1493 }
1494}
1495
1496type lifecycleObservedListener struct {
1497 wire.Listener
1498 closed chan struct{}
1499 once sync.Once
1500}
1501
1502func (l *lifecycleObservedListener) Close() error {
1503 err := l.Listener.Close()
1504 l.once.Do(func() { close(l.closed) })
1505 return err
1506}
1507
1508func TestLifecycleSocketCloseCatalogueRace(t *testing.T) {
1509 runtimeDir := filepath.Join(t.TempDir(), "runtime")
1510 stateDir := filepath.Join(t.TempDir(), "state")
1511 oldOwner, err := lifecycle.TryAcquire(runtimeDir)
1512 require.NoError(t, err)
1513 oldOpened, err := persist.OpenOrCreate(stateDir)
1514 require.NoError(t, err)
1515 oldListener, err := ipc.Listen(runtimeDir)
1516 require.NoError(t, err)
1517
1518 socketCloseEntered := make(chan struct{})
1519 allowSocketClose := make(chan struct{})
1520 socketClosed := make(chan struct{})
1521 catalogueCloseEntered := make(chan struct{})
1522 allowCatalogueClose := make(chan struct{})
1523 catalogueClosed := make(chan struct{})
1524 ownerReleaseEntered := make(chan struct{})
1525 allowOwnerRelease := make(chan struct{})
1526 ownerReleased := make(chan struct{})
1527 teardownEvents := make(chan string, 3)
1528
1529 controlledListener := &lifecycleControlledListener{
1530 Listener: oldListener,
1531 entered: socketCloseEntered,
1532 proceed: allowSocketClose,
1533 closed: socketClosed,
1534 events: teardownEvents,
1535 }
1536 controlledCatalogue := &lifecycleControlledCatalogue{
1537 Catalogue: oldOpened.Catalogue,
1538 entered: catalogueCloseEntered,
1539 proceed: allowCatalogueClose,
1540 closed: catalogueClosed,
1541 events: teardownEvents,
1542 }
1543 oldDaemon := daemon.New(pty.NewFactory(), clock.New(), discardLog(), daemon.WithCatalogue(controlledCatalogue, oldOpened.Records))
1544 oldCtx, stopOld := context.WithCancel(context.Background())
1545 oldDone := make(chan error, 1)
1546 go func() {
1547 serveErr := oldDaemon.Serve(oldCtx, sessionwire.NewServerListener(controlledListener))
1548 teardownEvents <- "owner-release"
1549 close(ownerReleaseEntered)
1550 <-allowOwnerRelease
1551 releaseErr := oldOwner.Release()
1552 close(ownerReleased)
1553 oldDone <- errors.Join(serveErr, releaseErr)
1554 }()
1555
1556 newAcquireAttempted := make(chan struct{})
1557 newDurableOpen := make(chan struct{})
1558 newListen := make(chan struct{})
1559 startupEvents := make(chan string, 3)
1560 newDone := make(chan error, 1)
1561 go func() {
1562 newDone <- runWithLifecycleOwnerDeps(context.Background(), runtimeDir, stateDir, func(ctx context.Context) error {
1563 close(newDurableOpen)
1564 opened, err := persist.OpenOrCreate(stateDir)
1565 if err != nil {
1566 return err
1567 }
1568 startupEvents <- "durable-open"
1569
1570 close(newListen)
1571 listener, err := ipc.Listen(runtimeDir)
1572 if err != nil {
1573 return errors.Join(err, opened.Catalogue.Close())
1574 }
1575 startupEvents <- "listen"
1576 return errors.Join(listener.Close(), opened.Catalogue.Close())
1577 }, lifecycleStartupDeps{
1578 ensurePrivate: func(string) error { return nil },
1579 acquire: func(ctx context.Context, dir string, retry time.Duration) (lifecycleOwnership, error) {
1580 close(newAcquireAttempted)
1581 owner, err := lifecycle.Acquire(ctx, dir, retry)
1582 if err != nil {
1583 return nil, err
1584 }
1585 <-ownerReleased
1586 startupEvents <- "owner-acquired"
1587 return owner, nil
1588 },
1589 })
1590 }()
1591
1592 awaitLifecycleStage(t, newAcquireAttempted, "new lifecycle acquisition attempt")
1593 assertLifecycleStagePending(t, newDurableOpen, "new durable open before old teardown")
1594 assertLifecycleStagePending(t, newListen, "new listen before old teardown")
1595
1596 stopOld()
1597 awaitLifecycleStage(t, socketCloseEntered, "old socket close")
1598 assertNewDaemonStartupPending(t, newDurableOpen, newListen, "socket close")
1599 close(allowSocketClose)
1600 awaitLifecycleStage(t, socketClosed, "old socket closed")
1601
1602 awaitLifecycleStage(t, catalogueCloseEntered, "old catalogue close")
1603 assertNewDaemonStartupPending(t, newDurableOpen, newListen, "catalogue close")
1604 close(allowCatalogueClose)
1605 awaitLifecycleStage(t, catalogueClosed, "old catalogue closed")
1606
1607 awaitLifecycleStage(t, ownerReleaseEntered, "old lifecycle owner release")
1608 assertNewDaemonStartupPending(t, newDurableOpen, newListen, "owner release")
1609 close(allowOwnerRelease)
1610 awaitLifecycleStage(t, ownerReleased, "old lifecycle owner released")
1611
1612 require.NoError(t, awaitLifecycleResult(t, oldDone, "old daemon teardown"))
1613 require.NoError(t, awaitLifecycleResult(t, newDone, "new daemon startup"))
1614 require.Equal(t, []string{"socket-close", "catalogue-close", "owner-release"}, []string{
1615 <-teardownEvents,
1616 <-teardownEvents,
1617 <-teardownEvents,
1618 })
1619 require.Equal(t, []string{"owner-acquired", "durable-open", "listen"}, []string{
1620 <-startupEvents,
1621 <-startupEvents,
1622 <-startupEvents,
1623 })
1624}
1625
1626func awaitLifecycleStage(t *testing.T, stage <-chan struct{}, name string) {
1627 t.Helper()
1628 select {
1629 case <-stage:
1630 case <-time.After(3 * time.Second):
1631 t.Fatalf("timed out waiting for %s", name)
1632 }
1633}
1634
1635func awaitLifecycleResult(t *testing.T, result <-chan error, name string) error {
1636 t.Helper()
1637 select {
1638 case err := <-result:
1639 return err
1640 case <-time.After(3 * time.Second):
1641 t.Fatalf("timed out waiting for %s", name)
1642 return nil
1643 }
1644}
1645
1646func assertLifecycleStagePending(t *testing.T, stage <-chan struct{}, name string) {
1647 t.Helper()
1648 select {
1649 case <-stage:
1650 t.Fatalf("unexpected %s", name)
1651 default:
1652 }
1653}
1654
1655func assertLifecycleResultPending(t *testing.T, result <-chan error, name string) {
1656 t.Helper()
1657 select {
1658 case err := <-result:
1659 t.Fatalf("unexpected %s: %v", name, err)
1660 default:
1661 }
1662}
1663
1664func assertNewDaemonStartupPending(t *testing.T, durableOpen, listen <-chan struct{}, heldAt string) {
1665 t.Helper()
1666 assertLifecycleStagePending(t, durableOpen, "new durable open while old daemon held at "+heldAt)
1667 assertLifecycleStagePending(t, listen, "new listen while old daemon held at "+heldAt)
1668}
1669
1670type lifecycleControlledListener struct {
1671 wire.Listener
1672 entered chan struct{}
1673 proceed chan struct{}
1674 closed chan struct{}
1675 events chan<- string
1676 once sync.Once
1677 err error
1678}
1679
1680func (l *lifecycleControlledListener) Close() error {
1681 l.once.Do(func() {
1682 l.events <- "socket-close"
1683 close(l.entered)
1684 <-l.proceed
1685 l.err = l.Listener.Close()
1686 close(l.closed)
1687 })
1688 return l.err
1689}
1690
1691type lifecycleControlledCatalogue struct {
1692 ports.Catalogue
1693 entered chan struct{}
1694 proceed chan struct{}
1695 closed chan struct{}
1696 events chan<- string
1697 once sync.Once
1698 err error
1699}
1700
1701func (c *lifecycleControlledCatalogue) Close() error {
1702 c.once.Do(func() {
1703 c.events <- "catalogue-close"
1704 close(c.entered)
1705 <-c.proceed
1706 c.err = c.Catalogue.Close()
1707 close(c.closed)
1708 })
1709 return c.err
1710}
1711
1712type integrationTransport struct{}
1713
1714func (*integrationTransport) Send(wire.Envelope) error { return nil }
1715func (*integrationTransport) Recv() (wire.Envelope, error) { return wire.Envelope{}, io.EOF }
1716func (*integrationTransport) Close() error { return nil }
1717func (*integrationTransport) LocalAddr() net.Addr { return nil }
1718func (*integrationTransport) RemoteAddr() net.Addr { return nil }
1719
1720// TestIntegration_KillSessionsPurgesSessionsAndKeepsDaemon is the P4.2 app-level
1721// contract: `kill --sessions` removes every live and stopped session but leaves the
1722// daemon serving, and the same process accepts fresh named and ephemeral
1723// sessions that start PTYs, forward input, and open a tab through the ordinary
1724// command palette. Only the distinct explicit daemon stop ends it.
1725func TestIntegration_KillSessionsPurgesSessionsAndKeepsDaemon(t *testing.T) {
1726 sz := domain.Size{Cols: 80, Rows: 24}
1727 dir, served := startDaemon(t, daemon.WithShell("/bin/cat", nil))
1728
1729 first, firstPump := attach(t, dir, protocol.IntentNew, "one", sz)
1730 require.NoError(t, first.SendClient(protocol.Input{Data: []byte("alpha\n")}))
1731 awaitText(t, firstPump, sz, "alpha")
1732 // A detached named session becomes stopped authority, so kill-all must purge
1733 // both the live and the stopped record in one operation.
1734 second, _ := attach(t, dir, protocol.IntentNew, "two", sz)
1735 require.NoError(t, second.Close())
1736 require.Len(t, listRemoteSessions(t, dir).Sessions, 2)
1737
1738 require.NoError(t, killAll(dir))
1739 // The daemon survives the purge: Serve has not returned and the registry is
1740 // empty.
1741 select {
1742 case err := <-served:
1743 t.Fatalf("KillAll stopped Serve: %v", err)
1744 default:
1745 }
1746 require.Eventually(t, func() bool { return len(listRemoteSessions(t, dir).Sessions) == 0 },
1747 5*time.Second, 10*time.Millisecond, "kill-all did not purge every session")
1748
1749 // Reuse: a fresh named session starts its PTY and forwards input.
1750 third, thirdPump := attach(t, dir, protocol.IntentNew, "again", sz)
1751 require.NoError(t, third.SendClient(protocol.Input{Data: []byte("beta\n")}))
1752 awaitText(t, thirdPump, sz, "beta")
1753 // A fresh ephemeral session is admitted too.
1754 fourth, fourthPump := attach(t, dir, protocol.IntentEphemeral, "", sz)
1755 require.NoError(t, fourth.SendClient(protocol.Input{Data: []byte("gamma\n")}))
1756 awaitText(t, fourthPump, sz, "gamma")
1757
1758 // A tab opened after the purge is real, live topology.
1759 require.NoError(t, third.SendClient(protocol.Input{Data: []byte("\x1b ")}))
1760 awaitText(t, thirdPump, sz, "Commands")
1761 require.NoError(t, third.SendClient(protocol.Input{Data: []byte("CNT\r")}))
1762 require.Eventually(t, func() bool {
1763 for _, info := range listRemoteSessions(t, dir).Sessions {
1764 if info.Name == "again" {
1765 return info.Tabs == 2
1766 }
1767 }
1768 return false
1769 }, 5*time.Second, 10*time.Millisecond, "a tab created after the purge was not published")
1770
1771 _ = first.Close()
1772 require.NoError(t, third.Close())
1773 require.NoError(t, fourth.Close())
1774
1775 // Only the distinct explicit daemon stop ends the daemon.
1776 require.NoError(t, killDaemon(dir))
1777 select {
1778 case err := <-served:
1779 require.NoError(t, err)
1780 case <-time.After(5 * time.Second):
1781 t.Fatal("daemon did not stop after a daemon stop request")
1782 }
1783}