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