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// awaitDetached consumes typed messages until the daemon signals the session's
416// end with the expected Detached reason, ignoring interleaved Output frames.
417func awaitDetached(t *testing.T, p *typedPump, reason uint8) {
418 t.Helper()
419 timeout := time.After(5 * time.Second)
420 for {
421 select {
422 case message, ok := <-p.ch:
423 if !ok {
424 t.Fatal("connection closed before Detached")
425 }
426 if detached, ok := message.(protocol.Detached); ok {
427 require.Equal(t, reason, detached.Reason)
428 return
429 }
430 case <-timeout:
431 t.Fatal("timed out waiting for Detached")
432 }
433 }
434}
435
436// awaitScreenText is like awaitText, but returns the reconstructed screen text
437// at the point the wanted text appears so callers can make additional checks.
438func awaitScreenText(t *testing.T, p *typedPump, sz domain.Size, want string) string {
439 t.Helper()
440 screen := vt.NewScreen(sz.Cols, sz.Rows)
441 timeout := time.After(5 * time.Second)
442 for {
443 select {
444 case message, ok := <-p.ch:
445 if !ok {
446 t.Fatalf("connection closed before %q appeared; screen=%q", want, screenText(screen))
447 }
448 if output, ok := message.(protocol.Output); ok {
449 screen.Write(output.Data)
450 text := screenText(screen)
451 if strings.Contains(text, want) {
452 return text
453 }
454 }
455 case <-timeout:
456 t.Fatalf("timed out waiting for %q; screen=%q", want, screenText(screen))
457 }
458 }
459}
460
461func assertNoTextAfterInput(t *testing.T, p *typedPump, sz domain.Size, absent string) {
462 t.Helper()
463 screen := vt.NewScreen(sz.Cols, sz.Rows)
464 timeout := time.After(300 * time.Millisecond)
465 for {
466 select {
467 case message, ok := <-p.ch:
468 if !ok {
469 return
470 }
471 if output, ok := message.(protocol.Output); ok {
472 screen.Write(output.Data)
473 require.NotContains(t, screenText(screen), absent)
474 }
475 case <-timeout:
476 return
477 }
478 }
479}
480
481func screenText(s *vt.Screen) string {
482 var b strings.Builder
483 for y := range s.Rows() {
484 for x := range s.Columns() {
485 r := s.Cell(x, y).Rune
486 if r == 0 {
487 r = ' '
488 }
489 b.WriteRune(r)
490 }
491 }
492 return b.String()
493}
494
495func shellFixture(t *testing.T, label string) string {
496 t.Helper()
497 path := filepath.Join(t.TempDir(), "shell")
498 require.NoError(t, os.WriteFile(path, []byte("#!/bin/sh\nprintf 'SHELL_COMMAND="+label+"\\n'\nexec /bin/sh\n"), 0o700))
499 return path
500}
501
502func assertChildEnvironment(t *testing.T, tr ports.ClientConnection, p *typedPump, sz domain.Size, wantTestEnv, wantShell, wantRuntimeDir, wantWayland string) {
503 t.Helper()
504 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"
505 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte(command)}))
506 text := awaitScreenText(t, p, sz, "TERM_PROGRAM=vev")
507 // The concrete PTY adapter selects the profile from host terminfo. Exact
508 // direct/fallback behavior is covered by the adapter's hermetic tests.
509 require.Regexp(t, `TERM=xterm-(direct|256color)\s`, text)
510 for _, want := range []string{
511 "VEV_TEST_ENV=" + wantTestEnv,
512 "SHELL=" + filepath.Base(wantShell),
513 "XDG_RUNTIME_DIR=" + wantRuntimeDir,
514 "WAYLAND_DISPLAY=" + wantWayland,
515 "COLORTERM=truecolor",
516 "TERM_PROGRAM=vev",
517 "VEV_PREFIX=session=environment,tab=",
518 } {
519 require.Contains(t, text, want)
520 }
521}
522
523func TestIntegration_AttachEnvironmentRefreshesFuturePTYChildren(t *testing.T) {
524 sz := domain.Size{Cols: 80, Rows: 24}
525 dir, _ := startDaemon(t)
526 firstShell := shellFixture(t, "first")
527 secondShell := shellFixture(t, "second")
528 firstEnv := []string{
529 "VEV_TEST_ENV=first", "SHELL=" + firstShell, "XDG_RUNTIME_DIR=/run/first", "WAYLAND_DISPLAY=wayland-first",
530 "TERM=client", "COLORTERM=client", "TERM_PROGRAM=client", "VEV=client",
531 }
532
533 tr1, p1 := attachWithEnvironment(t, dir, protocol.IntentNew, "environment", sz, firstEnv)
534 defer func() { _ = tr1.Close() }()
535 awaitText(t, p1, sz, "SHELL_COMMAND=first")
536 assertChildEnvironment(t, tr1, p1, sz, "first", firstShell, "/run/first", "wayland-first")
537
538 secondEnv := []string{
539 "VEV_TEST_ENV=second", "SHELL=" + secondShell, "XDG_RUNTIME_DIR=/run/second", "WAYLAND_DISPLAY=wayland-second",
540 "TERM=client", "COLORTERM=client", "TERM_PROGRAM=client", "VEV=client",
541 }
542 tr2, p2 := attachWithEnvironment(t, dir, protocol.IntentAttach, "environment", sz, secondEnv)
543 defer func() { _ = tr2.Close() }()
544
545 // The first shell was already running, so it retains its original environment.
546 assertChildEnvironment(t, tr2, p2, sz, "first", firstShell, "/run/first", "wayland-first")
547
548 require.NoError(t, tr2.SendClient(protocol.Input{Data: []byte("\x1b ")}))
549 awaitText(t, p2, sz, "Commands")
550 require.NoError(t, tr2.SendClient(protocol.Input{Data: []byte("CNT\r")}))
551 awaitText(t, p2, sz, "SHELL_COMMAND=second")
552 assertChildEnvironment(t, tr2, p2, sz, "second", secondShell, "/run/second", "wayland-second")
553}
554
555func TestIntegration_TwoAttachmentsReceiveSharedMutationPTYOutput(t *testing.T) {
556 sz := domain.Size{Cols: 80, Rows: 24}
557 dir, _ := startDaemon(t)
558 firstShell := shellFixture(t, "first")
559 const sharedOutput = "SECOND_SHARED_PTY_OUTPUT"
560 secondShell := filepath.Join(t.TempDir(), "second-shell")
561 require.NoError(t, os.WriteFile(secondShell, []byte("#!/bin/sh\nprintf 'SHELL_COMMAND=second\\n'\nIFS= read -r _\nprintf '"+sharedOutput+"\\n'\nexec /bin/sh\n"), 0o700))
562 firstEnv := []string{
563 "VEV_TEST_ENV=first", "SHELL=" + firstShell, "XDG_RUNTIME_DIR=/run/first", "WAYLAND_DISPLAY=wayland-first",
564 "TERM=client", "COLORTERM=client", "TERM_PROGRAM=client", "VEV=client",
565 }
566 tr1, p1 := attachWithEnvironment(t, dir, protocol.IntentNew, "environment", sz, firstEnv)
567 defer func() { _ = tr1.Close() }()
568 awaitText(t, p1, sz, "SHELL_COMMAND=first")
569 assertChildEnvironment(t, tr1, p1, sz, "first", firstShell, "/run/first", "wayland-first")
570
571 secondEnv := []string{
572 "VEV_TEST_ENV=second", "SHELL=" + secondShell, "XDG_RUNTIME_DIR=/run/second", "WAYLAND_DISPLAY=wayland-second",
573 "TERM=client", "COLORTERM=client", "TERM_PROGRAM=client", "VEV=client",
574 }
575 tr2, p2 := attachWithEnvironment(t, dir, protocol.IntentAttach, "environment", sz, secondEnv)
576 defer func() { _ = tr2.Close() }()
577 assertChildEnvironment(t, tr2, p2, sz, "first", firstShell, "/run/first", "wayland-first")
578
579 // The second attachment mutates shared tab state. Its newly opened PTY must
580 // also be rendered to that attachment, not only to the coordinator primary.
581 require.NoError(t, tr2.SendClient(protocol.Input{Data: []byte("\x1b ")}))
582 awaitText(t, p2, sz, "Commands")
583 require.NoError(t, tr2.SendClient(protocol.Input{Data: []byte("CNT\r")}))
584 awaitText(t, p2, sz, "SHELL_COMMAND=second")
585
586 // Each attachment owns its selected tab. Move the first attachment to the
587 // new PTY before releasing its gate, so the fresh PTY output must reach both
588 // views rather than only being recovered by a later first paint.
589 require.NoError(t, tr1.SendClient(protocol.Input{Data: []byte("\x1b2")}))
590 awaitText(t, p1, sz, "SHELL_COMMAND=second")
591 require.NoError(t, tr1.SendClient(protocol.Input{Data: []byte("go\n")}))
592 awaitText(t, p2, sz, sharedOutput)
593 awaitText(t, p1, sz, sharedOutput)
594}
595
596func TestIntegration_AttachFirstOutput(t *testing.T) {
597 sz := domain.Size{Cols: 80, Rows: 24}
598 dir, _ := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf HELLO; sleep 30"}))
599
600 tr, p := attach(t, dir, protocol.IntentEphemeral, "", sz)
601 defer func() { _ = tr.Close() }()
602
603 awaitText(t, p, sz, "HELLO")
604}
605
606func TestIntegration_InputRoundtrip(t *testing.T) {
607 sz := domain.Size{Cols: 80, Rows: 24}
608 dir, _ := startDaemon(t, daemon.WithShell("/bin/cat", nil))
609
610 tr, p := attach(t, dir, protocol.IntentEphemeral, "", sz)
611 defer func() { _ = tr.Close() }()
612
613 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("PINGPONG\n")}))
614 awaitText(t, p, sz, "PINGPONG")
615}
616
617func TestIntegration_CommandPaletteCreatesTab(t *testing.T) {
618 sz := domain.Size{Cols: 80, Rows: 24}
619 dir, _ := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf READY; sleep 30"}))
620
621 tr, p := attach(t, dir, protocol.IntentEphemeral, "", sz)
622 defer func() { _ = tr.Close() }()
623 awaitText(t, p, sz, "READY")
624
625 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("\x1b ")}))
626 awaitText(t, p, sz, "Commands")
627
628 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("CNT\r")}))
629 // Tab labels are enriched with the focused pane's title; with no process
630 // inspector wired in this test, that title falls back to the shell's
631 // basename ("sh").
632 text := awaitScreenText(t, p, sz, " 1 (sh) 2 (sh) ")
633 require.Contains(t, text, " 1 (sh) 2 (sh) ")
634}
635
636func TestIntegration_CommandPaletteRenamesEphemeralSession(t *testing.T) {
637 sz := domain.Size{Cols: 80, Rows: 24}
638 dir, _ := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf READY; sleep 30"}))
639
640 tr, p := attach(t, dir, protocol.IntentEphemeral, "", sz)
641 defer func() { _ = tr.Close() }()
642 text := awaitScreenText(t, p, sz, "READY")
643 require.Contains(t, text, " 0* ")
644
645 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("\x1b ")}))
646 awaitText(t, p, sz, "Commands")
647
648 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("RNS\r")}))
649 awaitText(t, p, sz, "Rename session")
650
651 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("\x7fwork\r")}))
652 text = awaitScreenText(t, p, sz, " work ")
653 require.NotContains(t, text, "work*")
654
655 listRaw, err := ipc.DialContext(context.Background(), dir)
656 require.NoError(t, err)
657 listTr := sessionwire.NewClientConnection(listRaw)
658 defer func() { _ = listTr.Close() }()
659 require.NoError(t, listTr.SendClient(protocol.List{}))
660 message, err := listTr.ReceiveServer()
661 require.NoError(t, err)
662 sessions, ok := message.(protocol.Sessions)
663 require.True(t, ok, "expected session listing, got %T", message)
664 require.NotEmpty(t, sessions.Sessions)
665 require.Equal(t, "work", sessions.Sessions[0].Name)
666 require.False(t, sessions.Sessions[0].Ephemeral)
667}
668
669func TestIntegration_AltCWithoutPaletteDoesNotCreateTab(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.NotContains(t, text, " 1 2 ")
677
678 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("\x1bc")}))
679 assertNoTextAfterInput(t, p, sz, " 1 2 ")
680}
681
682func TestIntegration_EphemeralSurvivesDetachAndReattaches(t *testing.T) {
683 sz := domain.Size{Cols: 80, Rows: 24}
684 dir, served := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf MARKER; sleep 30"}))
685
686 tr1, p1 := attach(t, dir, protocol.IntentEphemeral, "", sz)
687 awaitText(t, p1, sz, "MARKER")
688 require.NoError(t, tr1.Close())
689
690 select {
691 case err := <-served:
692 t.Fatalf("daemon shut down while an ephemeral session was alive: %v", err)
693 case <-time.After(300 * time.Millisecond):
694 }
695
696 tr2, p2 := attach(t, dir, protocol.IntentAttach, "0", sz)
697 defer func() { _ = tr2.Close() }()
698 awaitText(t, p2, sz, "MARKER")
699
700 // An ephemeral session's daemon is ended only by the explicit daemon stop.
701 require.NoError(t, killDaemon(dir))
702 select {
703 case err := <-served:
704 require.NoError(t, err)
705 case <-time.After(5 * time.Second):
706 t.Fatal("daemon did not stop after a daemon stop request")
707 }
708}
709
710func TestIntegration_EphemeralNotListedAfterDaemonRestart(t *testing.T) {
711 sz := domain.Size{Cols: 80, Rows: 24}
712 stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
713 t.Setenv("XDG_STATE_HOME", stateRoot)
714 t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
715 dir := ipc.SocketDir()
716
717 _, served := startDaemonInDir(t, dir, daemon.WithShell("/bin/sh", []string{"-c", "printf TEMP; sleep 30"}))
718 tr, p := attach(t, dir, protocol.IntentEphemeral, "", sz)
719 awaitText(t, p, sz, "TEMP")
720 require.NoError(t, tr.Close())
721
722 // An ephemeral session lives only while its daemon does; the explicit daemon
723 // stop over the typed control wire ends that daemon.
724 require.NoError(t, killDaemon(dir))
725 select {
726 case err := <-served:
727 require.NoError(t, err)
728 case <-time.After(5 * time.Second):
729 t.Fatal("daemon did not stop after a daemon stop request")
730 }
731
732 _, served2 := startDaemonInDir(t, dir, daemon.WithShell("/bin/sh", []string{"-c", "sleep 30"}))
733 sessions := listRemoteSessions(t, dir)
734 require.Empty(t, sessions.Sessions)
735
736 require.NoError(t, killDaemon(dir))
737 select {
738 case err := <-served2:
739 require.NoError(t, err)
740 case <-time.After(5 * time.Second):
741 t.Fatal("restarted daemon did not stop after a daemon stop request")
742 }
743}
744
745func TestIntegration_KillDaemonPreservesMultipleNamedSessions(t *testing.T) {
746 sz := domain.Size{Cols: 80, Rows: 24}
747 stateDir := filepath.Join(t.TempDir(), "state")
748 dir := filepath.Join(t.TempDir(), "runtime")
749 repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
750
751 start := func() <-chan error {
752 opened, err := persist.OpenOrCreate(stateDir)
753 require.NoError(t, err)
754 coordinator := recovery.NewCoordinator(opened.Catalogue, repository, rand.Reader)
755 _, served := startDaemonInDir(t, dir,
756 daemon.WithShell("/bin/sh", []string{"-c", "sleep 30"}),
757 daemon.WithCatalogue(opened.Catalogue, opened.Records),
758 daemon.WithSnapshotRepository(repository),
759 daemon.WithRecoveryCoordinator(coordinator),
760 )
761 return served
762 }
763
764 served := start()
765 first, _ := attach(t, dir, protocol.IntentNew, "alpha", sz)
766 require.NoError(t, first.Close())
767 second, _ := attach(t, dir, protocol.IntentNew, "beta", sz)
768 require.NoError(t, second.Close())
769
770 require.NoError(t, killDaemon(dir))
771 select {
772 case err := <-served:
773 require.NoError(t, err)
774 case <-time.After(5 * time.Second):
775 t.Fatal("daemon did not stop")
776 }
777
778 served = start()
779 sessions := listRemoteSessions(t, dir)
780 require.Equal(t, []protocol.SessionInfo{
781 {Name: "alpha", State: protocol.SessionDown},
782 {Name: "beta", State: protocol.SessionDown},
783 }, sessions.Sessions)
784
785 // Kill-all purges every durable record but leaves this daemon serving.
786 require.NoError(t, killAll(dir))
787 require.Empty(t, listRemoteSessions(t, dir).Sessions)
788 select {
789 case err := <-served:
790 t.Fatalf("kill-all stopped the restarted daemon: %v", err)
791 default:
792 }
793 require.NoError(t, killDaemon(dir))
794 select {
795 case err := <-served:
796 require.NoError(t, err)
797 case <-time.After(5 * time.Second):
798 t.Fatal("restarted daemon did not stop after a daemon stop request")
799 }
800
801 served = start()
802 require.Empty(t, listRemoteSessions(t, dir).Sessions)
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("empty daemon did not stop")
809 }
810}
811
812func TestIntegration_NamedSurvivesReattach(t *testing.T) {
813 sz := domain.Size{Cols: 80, Rows: 24}
814 dir, served := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "printf MARKER; sleep 30"}))
815
816 tr1, p1 := attach(t, dir, protocol.IntentNew, "work", sz)
817 awaitText(t, p1, sz, "MARKER")
818
819 // Detach: a named session must survive.
820 require.NoError(t, tr1.Close())
821
822 // Daemon must still be running (session alive).
823 select {
824 case err := <-served:
825 t.Fatalf("daemon shut down while a named session was alive: %v", err)
826 case <-time.After(300 * time.Millisecond):
827 }
828
829 // Re-attach: the first paint must reproduce the retained screen state.
830 tr2, p2 := attach(t, dir, protocol.IntentAttach, "work", sz)
831 defer func() { _ = tr2.Close() }()
832 awaitText(t, p2, sz, "MARKER")
833}
834
835// TestIntegration_NamedSessionLastRemovalKeepsDaemonServing is the P4.1
836// app-level contract (P4.1 review GO-003): a named session killed as the last
837// registry entry is removed without stopping Serve, and the same daemon process
838// accepts the same name again. It runs over the real socket, PTY, and
839// catalogue/recovery/snapshot persistence the production composition installs.
840// A named session's final live removal is a durable purge, so the re-attached
841// name carries a fresh persisted incarnation rather than resuming the deleted
842// one.
843func TestIntegration_NamedSessionLastRemovalKeepsDaemonServing(t *testing.T) {
844 sz := domain.Size{Cols: 80, Rows: 24}
845 stateDir := filepath.Join(t.TempDir(), "state")
846 dir := filepath.Join(t.TempDir(), "runtime")
847 repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
848 opened, err := persist.OpenOrCreate(stateDir)
849 require.NoError(t, err)
850 coordinator := recovery.NewCoordinator(opened.Catalogue, repository, rand.Reader)
851
852 _, served := startDaemonInDir(t, dir,
853 daemon.WithShell("/bin/cat", nil),
854 daemon.WithCatalogue(opened.Catalogue, opened.Records),
855 daemon.WithSnapshotRepository(repository),
856 daemon.WithRecoveryCoordinator(coordinator),
857 )
858
859 first, p1 := attach(t, dir, protocol.IntentNew, "work", sz)
860 require.NoError(t, first.SendClient(protocol.Input{Data: []byte("ping\n")}))
861 awaitText(t, p1, sz, "ping")
862 firstRecord, ok, err := opened.Catalogue.Record("work")
863 require.NoError(t, err)
864 require.True(t, ok, "a live named session must own a durable record")
865
866 // End it as the last session: the registry empties, but that is not daemon
867 // shutdown in the P4.1 model.
868 require.NoError(t, killNamedSession(dir, "work"))
869 awaitDetached(t, p1, protocol.ReasonSessionKilled)
870 require.Eventually(t, func() bool { return len(listRemoteSessions(t, dir).Sessions) == 0 },
871 5*time.Second, 10*time.Millisecond, "session was not removed")
872 select {
873 case err := <-served:
874 t.Fatalf("Serve stopped after the last session was removed: %v", err)
875 default:
876 }
877 _, ok, err = opened.Catalogue.Record("work")
878 require.NoError(t, err)
879 require.False(t, ok, "final live removal purges the durable record")
880
881 // The same daemon process accepts the same name again and can still serve a
882 // fresh terminal operation.
883 second, p2 := attach(t, dir, protocol.IntentNew, "work", sz)
884 require.NoError(t, second.SendClient(protocol.Input{Data: []byte("pong\n")}))
885 awaitText(t, p2, sz, "pong")
886 sessions := listRemoteSessions(t, dir).Sessions
887 require.Len(t, sessions, 1)
888 require.Equal(t, "work", sessions[0].Name)
889 require.Equal(t, protocol.SessionUp, sessions[0].State)
890 secondRecord, ok, err := opened.Catalogue.Record("work")
891 require.NoError(t, err)
892 require.True(t, ok, "re-attached named session must persist its identity")
893 require.NotEqual(t, firstRecord.IncarnationID, secondRecord.IncarnationID,
894 "a re-created name must not reuse the purged incarnation")
895
896 require.NoError(t, second.Close())
897 require.NoError(t, killDaemon(dir))
898 select {
899 case err := <-served:
900 require.NoError(t, err)
901 case <-time.After(5 * time.Second):
902 t.Fatal("daemon did not stop after a daemon stop request")
903 }
904}
905
906// TestIntegration_NamedSessionRestoresPersistedIdentity proves a stopped named
907// session is restored inside the daemon process that is already serving it,
908// keeping its exact persisted identity. The seed daemon stops while holding the
909// session as its only registry entry, which preserves it as stopped authority;
910// the daemon under test then resumes it over a real socket with the production
911// catalogue/recovery/snapshot options and serves a fresh operation.
912func TestIntegration_NamedSessionRestoresPersistedIdentity(t *testing.T) {
913 sz := domain.Size{Cols: 80, Rows: 24}
914 stateDir := filepath.Join(t.TempDir(), "state")
915 repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
916 name := publishRestorableCheckpoint(t, stateDir, repository)
917
918 opened, err := persist.OpenOrCreate(stateDir)
919 require.NoError(t, err)
920 stopped, ok, err := opened.Catalogue.Record(name)
921 require.NoError(t, err)
922 require.True(t, ok, "stopped named session must keep its durable record")
923 require.NotNil(t, stopped.Committed, "stopped named session must keep its committed checkpoint")
924
925 dir := filepath.Join(t.TempDir(), "runtime")
926 coordinator := recovery.NewCoordinator(opened.Catalogue, repository, rand.Reader)
927 _, served := startDaemonInDir(t, dir,
928 daemon.WithShell("/bin/cat", nil),
929 daemon.WithCatalogue(opened.Catalogue, opened.Records),
930 daemon.WithSnapshotRepository(repository),
931 daemon.WithRecoveryCoordinator(coordinator),
932 )
933 require.Equal(t, []protocol.SessionInfo{{Name: name, State: protocol.SessionDown}}, listRemoteSessions(t, dir).Sessions)
934 select {
935 case err := <-served:
936 t.Fatalf("Serve stopped before the restore: %v", err)
937 default:
938 }
939
940 conn, p := attach(t, dir, protocol.IntentAttach, name, sz)
941 require.NoError(t, conn.SendClient(protocol.Input{Data: []byte("revived\n")}))
942 awaitText(t, p, sz, "revived")
943 require.Eventually(t, func() bool {
944 sessions := listRemoteSessions(t, dir).Sessions
945 return len(sessions) == 1 && sessions[0].State == protocol.SessionUp
946 }, 5*time.Second, 10*time.Millisecond, "named session was not restored")
947
948 restored, ok, err := opened.Catalogue.Record(name)
949 require.NoError(t, err)
950 require.True(t, ok)
951 require.Equal(t, stopped.IncarnationID, restored.IncarnationID,
952 "restore must retain the persisted incarnation")
953 require.Equal(t, stopped.CreatedAt, restored.CreatedAt,
954 "restore must retain the persisted lifecycle timestamp")
955
956 require.NoError(t, conn.Close())
957 select {
958 case err := <-served:
959 t.Fatalf("Serve stopped while the restored session was live: %v", err)
960 default:
961 }
962 require.NoError(t, killDaemon(dir))
963 select {
964 case err := <-served:
965 require.NoError(t, err)
966 case <-time.After(5 * time.Second):
967 t.Fatal("daemon did not stop after a daemon stop request")
968 }
969}
970
971func TestMultipleClientsOneLifecycleOwner(t *testing.T) {
972 dir := filepath.Join(t.TempDir(), "runtime")
973 owner, err := lifecycle.TryAcquire(dir)
974 require.NoError(t, err)
975
976 ready := make(chan struct{})
977 transport := &integrationTransport{}
978 dial := func(context.Context, string) (wire.Transport, error) {
979 select {
980 case <-ready:
981 return transport, nil
982 default:
983 return nil, os.ErrNotExist
984 }
985 }
986 var spawns atomic.Int32
987 cfg := backoffConfig{initial: time.Millisecond, max: 2 * time.Millisecond, total: time.Second}
988
989 const clients = 4
990 errs := make(chan error, clients)
991 var wg sync.WaitGroup
992 for range clients {
993 wg.Go(func() {
994 _, err := ensureDaemonWithLifecycle(context.Background(), dir, dial, func() error {
995 spawns.Add(1)
996 return nil
997 }, cfg)
998 errs <- err
999 })
1000 }
1001 require.Never(t, func() bool { return spawns.Load() != 0 }, 50*time.Millisecond, time.Millisecond)
1002 close(ready)
1003 require.NoError(t, owner.Release())
1004 wg.Wait()
1005 close(errs)
1006 for err := range errs {
1007 require.NoError(t, err)
1008 }
1009 require.Zero(t, spawns.Load())
1010}
1011
1012func TestLifecycleOwnershipOutlivesMaintenanceWriter(t *testing.T) {
1013 runtimeDir := filepath.Join(t.TempDir(), "runtime")
1014 stateDir := filepath.Join(t.TempDir(), "state")
1015 catalogue := &lifecycleObservedCatalogue{
1016 Catalogue: newTestPersister(t, stateDir),
1017 closed: make(chan struct{}),
1018 }
1019 require.NoError(t, catalogue.Create(domain.CatalogueRecord{Name: "work", IncarnationID: domain.IncarnationID{1}}))
1020 repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
1021 shutdownClock := newLifecycleShutdownClock(t)
1022 maintenanceRepository := &lifecycleBlockingMaintenanceRepository{
1023 SnapshotRepository: repository,
1024 entered: make(chan struct{}),
1025 release: make(chan struct{}),
1026 returned: make(chan struct{}),
1027 }
1028 ctx, cancel := context.WithCancel(context.Background())
1029 t.Cleanup(cancel)
1030 var releaseOnce sync.Once
1031 releaseMaintenance := func() { releaseOnce.Do(func() { close(maintenanceRepository.release) }) }
1032 t.Cleanup(releaseMaintenance)
1033
1034 callbackReturned := make(chan struct{})
1035 wrapperReturned := make(chan error, 1)
1036 go func() {
1037 wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
1038 defer close(callbackReturned)
1039 d := daemon.New(
1040 pty.NewFactory(),
1041 shutdownClock,
1042 discardLog(),
1043 daemon.WithSnapshotRepository(repository),
1044 daemon.WithRecoveryCoordinator(recovery.NewCoordinator(catalogue, maintenanceRepository, nil)),
1045 daemon.WithSnapshotGarbageCollection(),
1046 daemon.WithCatalogue(catalogue, nil),
1047 )
1048 listener, err := ipc.Listen(runtimeDir)
1049 if err != nil {
1050 return err
1051 }
1052 return d.Serve(ctx, sessionwire.NewServerListener(listener))
1053 })
1054 }()
1055
1056 awaitLifecycleStage(t, maintenanceRepository.entered, "background maintenance repository call")
1057 cancel()
1058 select {
1059 case <-callbackReturned:
1060 t.Fatal("Serve callback returned after shutdown deadline while maintenance was blocked")
1061 case <-time.After(100 * time.Millisecond):
1062 }
1063 assertLifecycleStagePending(t, catalogue.closed, "catalogue close while maintenance is blocked")
1064 assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return while maintenance is blocked")
1065 _, err := lifecycle.TryAcquire(runtimeDir)
1066 require.ErrorIs(t, err, lifecycle.ErrBusy)
1067
1068 releaseMaintenance()
1069 awaitLifecycleStage(t, maintenanceRepository.returned, "maintenance repository return")
1070 awaitLifecycleStage(t, catalogue.closed, "catalogue close after maintenance")
1071 awaitLifecycleStage(t, callbackReturned, "Serve callback return after catalogue close")
1072 require.NoError(t, awaitLifecycleResult(t, wrapperReturned, "lifecycle wrapper return"))
1073 owner, err := lifecycle.TryAcquire(runtimeDir)
1074 require.NoError(t, err)
1075 require.NoError(t, owner.Release())
1076}
1077
1078type lifecycleBlockingMaintenanceRepository struct {
1079 ports.SnapshotRepository
1080 entered chan struct{}
1081 release chan struct{}
1082 returned chan struct{}
1083 enterOnce sync.Once
1084 returnOnce sync.Once
1085}
1086
1087func (r *lifecycleBlockingMaintenanceRepository) SnapshotIncarnations(context.Context) ([]domain.IncarnationID, error) {
1088 return []domain.IncarnationID{{1}}, nil
1089}
1090
1091func (r *lifecycleBlockingMaintenanceRepository) CollectIncarnationGarbage(context.Context, domain.IncarnationID, *domain.CheckpointRef) error {
1092 r.enterOnce.Do(func() { close(r.entered) })
1093 <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
1094 r.returnOnce.Do(func() { close(r.returned) })
1095 return nil
1096}
1097
1098type lifecycleObservedCatalogue struct {
1099 ports.Catalogue
1100 closed chan struct{}
1101 closeOnce sync.Once
1102}
1103
1104func (c *lifecycleObservedCatalogue) Close() error {
1105 err := c.Catalogue.Close()
1106 c.closeOnce.Do(func() { close(c.closed) })
1107 return err
1108}
1109
1110func TestLifecycleOwnershipOutlivesSnapshotWriter(t *testing.T) {
1111 result := runBlockedSnapshotWriterShutdown(t)
1112 assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")
1113 assertLifecycleResultPending(t, result.wrapperReturned, "lifecycle wrapper return while snapshot writer is blocked")
1114 _, err := lifecycle.TryAcquire(result.runtimeDir)
1115 require.ErrorIs(t, err, lifecycle.ErrBusy)
1116
1117 result.releaseWriter()
1118 awaitLifecycleStage(t, result.publishReturned, "snapshot writer return")
1119 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
1120 owner, err := lifecycle.TryAcquire(result.runtimeDir)
1121 require.NoError(t, err)
1122 require.NoError(t, owner.Release())
1123}
1124
1125func TestLifecycleCallbackWaitsForEveryWriter(t *testing.T) {
1126 result := runBlockedSnapshotWriterShutdown(t)
1127 assertLifecycleStagePending(t, result.callbackReturned, "Serve callback return while snapshot writer is blocked")
1128
1129 result.releaseWriter()
1130 awaitLifecycleStage(t, result.publishReturned, "snapshot writer return")
1131 awaitLifecycleStage(t, result.callbackReturned, "Serve callback return after snapshot writer")
1132 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
1133}
1134
1135func TestFinalCheckpointTimeoutKeepsOwner(t *testing.T) {
1136 result := runBlockedSnapshotWriterShutdown(t)
1137 assertLifecycleResultPending(t, result.wrapperReturned, "ownership release after checkpoint timeout with writer alive")
1138 _, err := lifecycle.TryAcquire(result.runtimeDir)
1139 require.ErrorIs(t, err, lifecycle.ErrBusy)
1140
1141 result.releaseWriter()
1142 require.NoError(t, awaitLifecycleResult(t, result.wrapperReturned, "lifecycle wrapper return"))
1143}
1144
1145// Startup restoration repairs HEADs, promotes fallbacks, and replaces catalogue
1146// records. Lifecycle ownership must therefore outlive it: releasing the flock
1147// while a restoration repository call is still running would let a second
1148// daemon mutate the same durable state.
1149func TestLifecycleOwnershipOutlivesRestorationWriter(t *testing.T) {
1150 stateDir := filepath.Join(t.TempDir(), "state")
1151 repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
1152 checkpointed := publishRestorableCheckpoint(t, stateDir, repository)
1153
1154 runtimeDir := filepath.Join(t.TempDir(), "runtime")
1155 blocking := &lifecycleBlockingRestore{
1156 SnapshotRepository: repository,
1157 entered: make(chan struct{}),
1158 release: make(chan struct{}),
1159 returned: make(chan struct{}),
1160 }
1161 var releaseOnce sync.Once
1162 releaseRestore := func() { releaseOnce.Do(func() { close(blocking.release) }) }
1163 t.Cleanup(releaseRestore)
1164
1165 shutdownClock := newLifecycleShutdownClock(t)
1166 ctx, cancel := context.WithCancel(context.Background())
1167 t.Cleanup(cancel)
1168 listenerReady := make(chan struct{})
1169 callbackReturned := make(chan struct{})
1170 wrapperReturned := make(chan error, 1)
1171 go func() {
1172 wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
1173 defer close(callbackReturned)
1174 opened, err := persist.OpenOrCreate(stateDir)
1175 if err != nil {
1176 return err
1177 }
1178 listener, err := ipc.Listen(runtimeDir)
1179 if err != nil {
1180 return errors.Join(err, opened.Catalogue.Close())
1181 }
1182 close(listenerReady)
1183 d := daemon.New(
1184 pty.NewFactory(),
1185 shutdownClock,
1186 discardLog(),
1187 daemon.WithShell("/bin/cat", nil),
1188 daemon.WithCatalogue(opened.Catalogue, opened.Records),
1189 daemon.WithSnapshotRepository(blocking),
1190 )
1191 return d.Serve(ctx, sessionwire.NewServerListener(listener))
1192 })
1193 }()
1194
1195 awaitLifecycleStage(t, listenerReady, "daemon listener")
1196 awaitLifecycleStage(t, blocking.entered, "restoration repository call")
1197 require.Equal(t, checkpointed, blocking.repairedName(), "restoration must target the checkpointed session")
1198
1199 cancel()
1200 shutdownClock.nextTimer(t).fire()
1201 select {
1202 case <-callbackReturned:
1203 t.Fatal("Serve callback returned after shutdown deadline while restoration was blocked")
1204 case <-time.After(100 * time.Millisecond):
1205 }
1206 assertLifecycleStagePending(t, callbackReturned, "Serve callback return while restoration is blocked")
1207 assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return while restoration is blocked")
1208 _, err := lifecycle.TryAcquire(runtimeDir)
1209 require.ErrorIs(t, err, lifecycle.ErrBusy)
1210
1211 releaseRestore()
1212 awaitLifecycleStage(t, blocking.returned, "restoration repository return")
1213 awaitLifecycleStage(t, callbackReturned, "Serve callback return after restoration")
1214 require.NoError(t, awaitLifecycleResult(t, wrapperReturned, "lifecycle wrapper return"))
1215 owner, err := lifecycle.TryAcquire(runtimeDir)
1216 require.NoError(t, err)
1217 require.NoError(t, owner.Release())
1218}
1219
1220// publishRestorableCheckpoint runs a complete daemon over the real catalogue and
1221// repository until one named session owns a committed checkpoint, then shuts it
1222// down. The returned name is restorable by any later daemon on the same state.
1223func publishRestorableCheckpoint(t *testing.T, stateDir string, repository *snapshot.Repository) string {
1224 t.Helper()
1225 const name = "restorable"
1226 runtimeDir := filepath.Join(t.TempDir(), "seed-runtime")
1227 ctx, cancel := context.WithCancel(context.Background())
1228 defer cancel()
1229
1230 opened, err := persist.OpenOrCreate(stateDir)
1231 require.NoError(t, err)
1232 coordinator := recovery.NewCoordinator(opened.Catalogue, repository, rand.Reader)
1233 listener, err := ipc.Listen(runtimeDir)
1234 require.NoError(t, err)
1235 d := daemon.New(
1236 pty.NewFactory(),
1237 clock.New(),
1238 discardLog(),
1239 daemon.WithShell("/bin/cat", nil),
1240 daemon.WithCatalogue(opened.Catalogue, opened.Records),
1241 daemon.WithSnapshotRepository(repository),
1242 daemon.WithRecoveryCoordinator(coordinator),
1243 )
1244 served := make(chan error, 1)
1245 go func() { served <- d.Serve(ctx, sessionwire.NewServerListener(listener)) }()
1246
1247 tr, _ := attach(t, runtimeDir, protocol.IntentNew, name, domain.Size{Cols: 80, Rows: 24})
1248 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("checkpoint me\n")}))
1249 require.Eventually(t, func() bool {
1250 record, ok, _ := opened.Catalogue.Record(name)
1251 return ok && record.Committed != nil && record.DegradedReason == ""
1252 }, 5*time.Second, 10*time.Millisecond, "session never committed a checkpoint")
1253 require.NoError(t, tr.Close())
1254
1255 cancel()
1256 require.NoError(t, awaitLifecycleResult(t, served, "seed daemon shutdown"))
1257 return name
1258}
1259
1260type lifecycleBlockingRestore struct {
1261 ports.SnapshotRepository
1262 entered chan struct{}
1263 release chan struct{}
1264 returned chan struct{}
1265 enterOnce sync.Once
1266 returnOnce sync.Once
1267 mu sync.Mutex
1268 repaired string
1269}
1270
1271func (r *lifecycleBlockingRestore) LoadCheckpoint(ctx context.Context, id domain.IncarnationID, name string, ref domain.CheckpointRef) (ports.SnapshotGeneration, error) {
1272 r.mu.Lock()
1273 r.repaired = name
1274 r.mu.Unlock()
1275 return r.SnapshotRepository.LoadCheckpoint(ctx, id, name, ref)
1276}
1277
1278func (r *lifecycleBlockingRestore) ReconcileCheckpoint(ctx context.Context, id domain.IncarnationID, ref domain.CheckpointRef) error {
1279 r.enterOnce.Do(func() { close(r.entered) })
1280 <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
1281 r.returnOnce.Do(func() { close(r.returned) })
1282 return r.SnapshotRepository.ReconcileCheckpoint(ctx, id, ref)
1283}
1284
1285func (r *lifecycleBlockingRestore) repairedName() string {
1286 r.mu.Lock()
1287 defer r.mu.Unlock()
1288 return r.repaired
1289}
1290
1291type blockedSnapshotWriterShutdown struct {
1292 runtimeDir string
1293 callbackReturned <-chan struct{}
1294 wrapperReturned <-chan error
1295 publishReturned <-chan struct{}
1296 releaseWriter func()
1297}
1298
1299func runBlockedSnapshotWriterShutdown(t *testing.T) blockedSnapshotWriterShutdown {
1300 t.Helper()
1301 runtimeDir := filepath.Join(t.TempDir(), "runtime")
1302 stateDir := filepath.Join(t.TempDir(), "state")
1303 shutdownClock := newLifecycleShutdownClock(t)
1304 repository := newLifecycleBlockingRepository()
1305 ctx, cancel := context.WithCancel(context.Background())
1306 t.Cleanup(cancel)
1307
1308 listenerReady := make(chan struct{})
1309 listenerClosed := make(chan struct{})
1310 callbackReturned := make(chan struct{})
1311 wrapperReturned := make(chan error, 1)
1312 go func() {
1313 wrapperReturned <- runWithLifecycleOwner(ctx, runtimeDir, stateDir, func(ctx context.Context) error {
1314 defer close(callbackReturned)
1315 listener, err := ipc.Listen(runtimeDir)
1316 if err != nil {
1317 return err
1318 }
1319 observed := &lifecycleObservedListener{Listener: listener, closed: listenerClosed}
1320 close(listenerReady)
1321 d := daemon.New(
1322 pty.NewFactory(),
1323 shutdownClock,
1324 discardLog(),
1325 daemon.WithShell("/bin/cat", nil),
1326 daemon.WithSnapshotRepository(repository),
1327 )
1328 return d.Serve(ctx, sessionwire.NewServerListener(observed))
1329 })
1330 }()
1331
1332 awaitLifecycleStage(t, listenerReady, "daemon listener")
1333 tr, _ := attach(t, runtimeDir, protocol.IntentNew, "work", domain.Size{Cols: 80, Rows: 24})
1334 require.NoError(t, tr.SendClient(protocol.Input{Data: []byte("dirty state\n")}))
1335 awaitLifecycleStage(t, repository.entered, "snapshot publication")
1336 require.NoError(t, tr.Close())
1337 require.Eventually(t, func() bool {
1338 sessions := listRemoteSessions(t, runtimeDir)
1339 return len(sessions.Sessions) == 1 && !sessions.Sessions[0].Attached
1340 }, time.Second, 10*time.Millisecond)
1341 cancel()
1342 awaitLifecycleStage(t, listenerClosed, "listener close")
1343
1344 shutdownClock.nextTimer(t).fire()
1345 select {
1346 case <-callbackReturned:
1347 t.Fatal("Serve callback returned after checkpoint timeout while snapshot writer was blocked")
1348 case <-time.After(100 * time.Millisecond):
1349 }
1350 assertLifecycleStagePending(t, callbackReturned, "Serve callback return after checkpoint timeout")
1351 assertLifecycleResultPending(t, wrapperReturned, "lifecycle wrapper return after checkpoint timeout")
1352
1353 var releaseOnce sync.Once
1354 releaseWriter := func() { releaseOnce.Do(func() { close(repository.release) }) }
1355 t.Cleanup(releaseWriter)
1356 return blockedSnapshotWriterShutdown{
1357 runtimeDir: runtimeDir,
1358 callbackReturned: callbackReturned,
1359 wrapperReturned: wrapperReturned,
1360 publishReturned: repository.returned,
1361 releaseWriter: releaseWriter,
1362 }
1363}
1364
1365type lifecycleBlockingRepository struct {
1366 ports.SnapshotRepository
1367 entered chan struct{}
1368 release chan struct{}
1369 returned chan struct{}
1370 enterOnce sync.Once
1371 returnOnce sync.Once
1372}
1373
1374func newLifecycleBlockingRepository() *lifecycleBlockingRepository {
1375 return &lifecycleBlockingRepository{
1376 entered: make(chan struct{}),
1377 release: make(chan struct{}),
1378 returned: make(chan struct{}),
1379 }
1380}
1381
1382func (r *lifecycleBlockingRepository) Publish(context.Context, ports.SnapshotPublication) error {
1383 r.enterOnce.Do(func() { close(r.entered) })
1384 <-r.release // Intentionally ignore cancellation: lifecycle ownership must outlive this call.
1385 r.returnOnce.Do(func() { close(r.returned) })
1386 return nil
1387}
1388
1389type lifecycleShutdownClock struct {
1390 t *testing.T
1391 base ports.Clock
1392 timers chan *lifecycleManualTimer
1393}
1394
1395func newLifecycleShutdownClock(t *testing.T) *lifecycleShutdownClock {
1396 t.Helper()
1397 return &lifecycleShutdownClock{t: t, base: clock.New(), timers: make(chan *lifecycleManualTimer, 4)}
1398}
1399
1400func (c *lifecycleShutdownClock) Now() time.Time { return c.base.Now() }
1401func (c *lifecycleShutdownClock) NewTimer(delay time.Duration) ports.Timer {
1402 if delay != daemon.SnapshotShutdownTimeout() {
1403 return c.base.NewTimer(delay)
1404 }
1405 timer := &lifecycleManualTimer{t: c.t, ch: make(chan time.Time, 1)}
1406 select {
1407 case c.timers <- timer:
1408 default:
1409 c.t.Errorf("unexpected additional snapshot shutdown timer with delay %s", delay)
1410 timer.fire()
1411 }
1412 return timer
1413}
1414
1415func (c *lifecycleShutdownClock) nextTimer(t *testing.T) *lifecycleManualTimer {
1416 t.Helper()
1417 select {
1418 case timer := <-c.timers:
1419 return timer
1420 case <-time.After(3 * time.Second):
1421 t.Fatal("timed out waiting for shutdown deadline timer")
1422 return nil
1423 }
1424}
1425
1426type lifecycleManualTimer struct {
1427 t *testing.T
1428 ch chan time.Time
1429 stopped atomic.Bool
1430}
1431
1432func (t *lifecycleManualTimer) C() <-chan time.Time { return t.ch }
1433func (t *lifecycleManualTimer) Reset(delay time.Duration) bool {
1434 if delay != daemon.SnapshotShutdownTimeout() {
1435 t.t.Errorf("snapshot shutdown timer reset with delay %s, want %s", delay, daemon.SnapshotShutdownTimeout())
1436 }
1437 t.stopped.Store(false)
1438 return false
1439}
1440func (t *lifecycleManualTimer) Stop() bool { return !t.stopped.Swap(true) }
1441func (t *lifecycleManualTimer) fire() {
1442 if t.stopped.Load() {
1443 return
1444 }
1445 select {
1446 case t.ch <- time.Now():
1447 default:
1448 }
1449}
1450
1451func TestLifecycleManualTimerRepeatedFireIsNonBlocking(t *testing.T) {
1452 timer := &lifecycleManualTimer{t: t, ch: make(chan time.Time, 1)}
1453 timer.fire()
1454
1455 returned := make(chan struct{})
1456 go func() {
1457 timer.fire()
1458 close(returned)
1459 }()
1460
1461 select {
1462 case <-returned:
1463 case <-time.After(time.Second):
1464 t.Fatal("repeated manual timer fire blocked on a pending tick")
1465 }
1466}
1467
1468type lifecycleObservedListener struct {
1469 wire.Listener
1470 closed chan struct{}
1471 once sync.Once
1472}
1473
1474func (l *lifecycleObservedListener) Close() error {
1475 err := l.Listener.Close()
1476 l.once.Do(func() { close(l.closed) })
1477 return err
1478}
1479
1480func TestLifecycleSocketCloseCatalogueRace(t *testing.T) {
1481 runtimeDir := filepath.Join(t.TempDir(), "runtime")
1482 stateDir := filepath.Join(t.TempDir(), "state")
1483 oldOwner, err := lifecycle.TryAcquire(runtimeDir)
1484 require.NoError(t, err)
1485 oldOpened, err := persist.OpenOrCreate(stateDir)
1486 require.NoError(t, err)
1487 oldListener, err := ipc.Listen(runtimeDir)
1488 require.NoError(t, err)
1489
1490 socketCloseEntered := make(chan struct{})
1491 allowSocketClose := make(chan struct{})
1492 socketClosed := make(chan struct{})
1493 catalogueCloseEntered := make(chan struct{})
1494 allowCatalogueClose := make(chan struct{})
1495 catalogueClosed := make(chan struct{})
1496 ownerReleaseEntered := make(chan struct{})
1497 allowOwnerRelease := make(chan struct{})
1498 ownerReleased := make(chan struct{})
1499 teardownEvents := make(chan string, 3)
1500
1501 controlledListener := &lifecycleControlledListener{
1502 Listener: oldListener,
1503 entered: socketCloseEntered,
1504 proceed: allowSocketClose,
1505 closed: socketClosed,
1506 events: teardownEvents,
1507 }
1508 controlledCatalogue := &lifecycleControlledCatalogue{
1509 Catalogue: oldOpened.Catalogue,
1510 entered: catalogueCloseEntered,
1511 proceed: allowCatalogueClose,
1512 closed: catalogueClosed,
1513 events: teardownEvents,
1514 }
1515 oldDaemon := daemon.New(pty.NewFactory(), clock.New(), discardLog(), daemon.WithCatalogue(controlledCatalogue, oldOpened.Records))
1516 oldCtx, stopOld := context.WithCancel(context.Background())
1517 oldDone := make(chan error, 1)
1518 go func() {
1519 serveErr := oldDaemon.Serve(oldCtx, sessionwire.NewServerListener(controlledListener))
1520 teardownEvents <- "owner-release"
1521 close(ownerReleaseEntered)
1522 <-allowOwnerRelease
1523 releaseErr := oldOwner.Release()
1524 close(ownerReleased)
1525 oldDone <- errors.Join(serveErr, releaseErr)
1526 }()
1527
1528 newAcquireAttempted := make(chan struct{})
1529 newDurableOpen := make(chan struct{})
1530 newListen := make(chan struct{})
1531 startupEvents := make(chan string, 3)
1532 newDone := make(chan error, 1)
1533 go func() {
1534 newDone <- runWithLifecycleOwnerDeps(context.Background(), runtimeDir, stateDir, func(ctx context.Context) error {
1535 close(newDurableOpen)
1536 opened, err := persist.OpenOrCreate(stateDir)
1537 if err != nil {
1538 return err
1539 }
1540 startupEvents <- "durable-open"
1541
1542 close(newListen)
1543 listener, err := ipc.Listen(runtimeDir)
1544 if err != nil {
1545 return errors.Join(err, opened.Catalogue.Close())
1546 }
1547 startupEvents <- "listen"
1548 return errors.Join(listener.Close(), opened.Catalogue.Close())
1549 }, lifecycleStartupDeps{
1550 ensurePrivate: func(string) error { return nil },
1551 acquire: func(ctx context.Context, dir string, retry time.Duration) (lifecycleOwnership, error) {
1552 close(newAcquireAttempted)
1553 owner, err := lifecycle.Acquire(ctx, dir, retry)
1554 if err != nil {
1555 return nil, err
1556 }
1557 <-ownerReleased
1558 startupEvents <- "owner-acquired"
1559 return owner, nil
1560 },
1561 })
1562 }()
1563
1564 awaitLifecycleStage(t, newAcquireAttempted, "new lifecycle acquisition attempt")
1565 assertLifecycleStagePending(t, newDurableOpen, "new durable open before old teardown")
1566 assertLifecycleStagePending(t, newListen, "new listen before old teardown")
1567
1568 stopOld()
1569 awaitLifecycleStage(t, socketCloseEntered, "old socket close")
1570 assertNewDaemonStartupPending(t, newDurableOpen, newListen, "socket close")
1571 close(allowSocketClose)
1572 awaitLifecycleStage(t, socketClosed, "old socket closed")
1573
1574 awaitLifecycleStage(t, catalogueCloseEntered, "old catalogue close")
1575 assertNewDaemonStartupPending(t, newDurableOpen, newListen, "catalogue close")
1576 close(allowCatalogueClose)
1577 awaitLifecycleStage(t, catalogueClosed, "old catalogue closed")
1578
1579 awaitLifecycleStage(t, ownerReleaseEntered, "old lifecycle owner release")
1580 assertNewDaemonStartupPending(t, newDurableOpen, newListen, "owner release")
1581 close(allowOwnerRelease)
1582 awaitLifecycleStage(t, ownerReleased, "old lifecycle owner released")
1583
1584 require.NoError(t, awaitLifecycleResult(t, oldDone, "old daemon teardown"))
1585 require.NoError(t, awaitLifecycleResult(t, newDone, "new daemon startup"))
1586 require.Equal(t, []string{"socket-close", "catalogue-close", "owner-release"}, []string{
1587 <-teardownEvents,
1588 <-teardownEvents,
1589 <-teardownEvents,
1590 })
1591 require.Equal(t, []string{"owner-acquired", "durable-open", "listen"}, []string{
1592 <-startupEvents,
1593 <-startupEvents,
1594 <-startupEvents,
1595 })
1596}
1597
1598func awaitLifecycleStage(t *testing.T, stage <-chan struct{}, name string) {
1599 t.Helper()
1600 select {
1601 case <-stage:
1602 case <-time.After(3 * time.Second):
1603 t.Fatalf("timed out waiting for %s", name)
1604 }
1605}
1606
1607func awaitLifecycleResult(t *testing.T, result <-chan error, name string) error {
1608 t.Helper()
1609 select {
1610 case err := <-result:
1611 return err
1612 case <-time.After(3 * time.Second):
1613 t.Fatalf("timed out waiting for %s", name)
1614 return nil
1615 }
1616}
1617
1618func assertLifecycleStagePending(t *testing.T, stage <-chan struct{}, name string) {
1619 t.Helper()
1620 select {
1621 case <-stage:
1622 t.Fatalf("unexpected %s", name)
1623 default:
1624 }
1625}
1626
1627func assertLifecycleResultPending(t *testing.T, result <-chan error, name string) {
1628 t.Helper()
1629 select {
1630 case err := <-result:
1631 t.Fatalf("unexpected %s: %v", name, err)
1632 default:
1633 }
1634}
1635
1636func assertNewDaemonStartupPending(t *testing.T, durableOpen, listen <-chan struct{}, heldAt string) {
1637 t.Helper()
1638 assertLifecycleStagePending(t, durableOpen, "new durable open while old daemon held at "+heldAt)
1639 assertLifecycleStagePending(t, listen, "new listen while old daemon held at "+heldAt)
1640}
1641
1642type lifecycleControlledListener struct {
1643 wire.Listener
1644 entered chan struct{}
1645 proceed chan struct{}
1646 closed chan struct{}
1647 events chan<- string
1648 once sync.Once
1649 err error
1650}
1651
1652func (l *lifecycleControlledListener) Close() error {
1653 l.once.Do(func() {
1654 l.events <- "socket-close"
1655 close(l.entered)
1656 <-l.proceed
1657 l.err = l.Listener.Close()
1658 close(l.closed)
1659 })
1660 return l.err
1661}
1662
1663type lifecycleControlledCatalogue struct {
1664 ports.Catalogue
1665 entered chan struct{}
1666 proceed chan struct{}
1667 closed chan struct{}
1668 events chan<- string
1669 once sync.Once
1670 err error
1671}
1672
1673func (c *lifecycleControlledCatalogue) Close() error {
1674 c.once.Do(func() {
1675 c.events <- "catalogue-close"
1676 close(c.entered)
1677 <-c.proceed
1678 c.err = c.Catalogue.Close()
1679 close(c.closed)
1680 })
1681 return c.err
1682}
1683
1684type integrationTransport struct{}
1685
1686func (*integrationTransport) Send(wire.Envelope) error { return nil }
1687func (*integrationTransport) Recv() (wire.Envelope, error) { return wire.Envelope{}, io.EOF }
1688func (*integrationTransport) Close() error { return nil }
1689func (*integrationTransport) LocalAddr() net.Addr { return nil }
1690func (*integrationTransport) RemoteAddr() net.Addr { return nil }
1691
1692// TestIntegration_KillSessionsPurgesSessionsAndKeepsDaemon is the P4.2 app-level
1693// contract: `kill --sessions` removes every live and stopped session but leaves the
1694// daemon serving, and the same process accepts fresh named and ephemeral
1695// sessions that start PTYs, forward input, and open a tab through the ordinary
1696// command palette. Only the distinct explicit daemon stop ends it.
1697func TestIntegration_KillSessionsPurgesSessionsAndKeepsDaemon(t *testing.T) {
1698 sz := domain.Size{Cols: 80, Rows: 24}
1699 dir, served := startDaemon(t, daemon.WithShell("/bin/cat", nil))
1700
1701 first, firstPump := attach(t, dir, protocol.IntentNew, "one", sz)
1702 require.NoError(t, first.SendClient(protocol.Input{Data: []byte("alpha\n")}))
1703 awaitText(t, firstPump, sz, "alpha")
1704 // A detached named session becomes stopped authority, so kill-all must purge
1705 // both the live and the stopped record in one operation.
1706 second, _ := attach(t, dir, protocol.IntentNew, "two", sz)
1707 require.NoError(t, second.Close())
1708 require.Len(t, listRemoteSessions(t, dir).Sessions, 2)
1709
1710 require.NoError(t, killAll(dir))
1711 // The daemon survives the purge: Serve has not returned and the registry is
1712 // empty.
1713 select {
1714 case err := <-served:
1715 t.Fatalf("KillAll stopped Serve: %v", err)
1716 default:
1717 }
1718 require.Eventually(t, func() bool { return len(listRemoteSessions(t, dir).Sessions) == 0 },
1719 5*time.Second, 10*time.Millisecond, "kill-all did not purge every session")
1720
1721 // Reuse: a fresh named session starts its PTY and forwards input.
1722 third, thirdPump := attach(t, dir, protocol.IntentNew, "again", sz)
1723 require.NoError(t, third.SendClient(protocol.Input{Data: []byte("beta\n")}))
1724 awaitText(t, thirdPump, sz, "beta")
1725 // A fresh ephemeral session is admitted too.
1726 fourth, fourthPump := attach(t, dir, protocol.IntentEphemeral, "", sz)
1727 require.NoError(t, fourth.SendClient(protocol.Input{Data: []byte("gamma\n")}))
1728 awaitText(t, fourthPump, sz, "gamma")
1729
1730 // A tab opened after the purge is real, live topology.
1731 require.NoError(t, third.SendClient(protocol.Input{Data: []byte("\x1b ")}))
1732 awaitText(t, thirdPump, sz, "Commands")
1733 require.NoError(t, third.SendClient(protocol.Input{Data: []byte("CNT\r")}))
1734 require.Eventually(t, func() bool {
1735 for _, info := range listRemoteSessions(t, dir).Sessions {
1736 if info.Name == "again" {
1737 return info.Tabs == 2
1738 }
1739 }
1740 return false
1741 }, 5*time.Second, 10*time.Millisecond, "a tab created after the purge was not published")
1742
1743 _ = first.Close()
1744 require.NoError(t, third.Close())
1745 require.NoError(t, fourth.Close())
1746
1747 // Only the distinct explicit daemon stop ends the daemon.
1748 require.NoError(t, killDaemon(dir))
1749 select {
1750 case err := <-served:
1751 require.NoError(t, err)
1752 case <-time.After(5 * time.Second):
1753 t.Fatal("daemon did not stop after a daemon stop request")
1754 }
1755}