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