purge_test.go RawHistoryFolder

1package daemon
2
3import (
4	"bytes"
5	"context"
6	"errors"
7	"fmt"
8	"log/slog"
9	"strconv"
10	"strings"
11	"sync"
12	"sync/atomic"
13	"testing"
14	"time"
15
16	"github.com/stretchr/testify/mock"
17	"github.com/stretchr/testify/require"
18
19	"github.com/bnema/vev/internal/domain"
20	"github.com/bnema/vev/internal/ports"
21	portsmocks "github.com/bnema/vev/internal/ports/mocks"
22	"github.com/bnema/vev/internal/protocol"
23	snapcodec "github.com/bnema/vev/internal/snapshotcodec"
24	recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
25)
26
27// selectivePurgeRepository fails incarnation deletion only for the configured
28// lifecycles, so a purge can succeed for some records and fail for others. It
29// can also run a one-shot onDelete hook to interleave a state change (such as
30// an explicit shutdown) after the first destructive purge unit.
31type selectivePurgeRepository struct {
32	noOpSnapshotRepository
33
34	mu       sync.Mutex
35	fail     map[domain.IncarnationID]error
36	calls    []domain.IncarnationID
37	onDelete func()
38	once     sync.Once
39}
40
41func (r *selectivePurgeRepository) DeleteIncarnation(_ context.Context, id domain.IncarnationID) error {
42	r.mu.Lock()
43	r.calls = append(r.calls, id)
44	err := r.fail[id]
45	hook := r.onDelete
46	r.mu.Unlock()
47	if hook != nil {
48		r.once.Do(hook)
49	}
50	return err
51}
52
53func newSelectivePurgeRepository(fail map[domain.IncarnationID]error) *selectivePurgeRepository {
54	return &selectivePurgeRepository{fail: fail}
55}
56
57// TestPurgeAllRemovesLiveAndInactiveAndKeepsDaemonActive is the KillAll
58// survival contract: every live, stopped, and broken record is removed, the
59// daemon-wide shutdown facts are untouched, and the same daemon creates and
60// drives new named and ephemeral sessions.
61func TestPurgeAllRemovesLiveAndInactiveAndKeepsDaemonActive(t *testing.T) {
62	sz := domain.Size{Cols: 80, Rows: 24}
63	seedPTY, releaseSeed := newBlockingPTY(t)
64	writes := make(chan []byte, 8)
65	afterPTY, releaseAfter := newBlockingPTYWithWrites(t, writes)
66	tabPTY, releaseTab := newBlockingPTY(t)
67	d := newTestDaemon(t, newFactorySeq(t, seedPTY, afterPTY, tabPTY), stubClock{})
68
69	seed, err := createSessionForTest(d, "0", true, "/tmp", sz, terminalEnv{}, nil)
70	require.NoError(t, err)
71	require.NotNil(t, seed)
72	d.mu.Lock()
73	d.inactive["stopped"] = inactiveSession{name: "stopped", state: protocol.SessionDown, incarnation: domain.IncarnationID{7}}
74	d.inactive["broken"] = inactiveSession{name: "broken", state: protocol.SessionBroken, incarnation: domain.IncarnationID{8}}
75	d.mu.Unlock()
76
77	result := d.purgeAllSessions(protocol.ReasonSessionKilled)
78	require.False(t, result.Failed(), "purge failures: %v", result.summary())
79	require.Equal(t, 0, sessionCount(d))
80	requirePurgeAdmissionBalanced(t, d)
81	d.mu.Lock()
82	require.Empty(t, d.inactive, "purge must remove stopped and broken records")
83	closing := d.closing
84	d.mu.Unlock()
85	require.False(t, closing, "KillAll must not mark the daemon closing")
86	d.moveLifecycleMu.Lock()
87	moveClosing := d.moveLifecycleClosing
88	d.moveLifecycleMu.Unlock()
89	require.False(t, moveClosing, "KillAll must not close move admission")
90	select {
91	case <-d.done:
92		t.Fatal("KillAll must not close done")
93	default:
94	}
95	select {
96	case <-d.paneProcessCtx.Done():
97		t.Fatal("KillAll must not cancel the daemon-wide pane process context")
98	default:
99	}
100
101	// The same daemon admits a new named session, forwards PTY input, and opens
102	// a tab through the ordinary key-driven command API.
103	tr, sends, releaseConn := newConn(t,
104		mustHello(protocol.IntentNew, "after", sz),
105		frameInput([]byte("ping\n")),
106		frameInput([]byte("\x1b ")),
107		frameInput([]byte("CNT\r")),
108	)
109	var hg sync.WaitGroup
110	hg.Go(func() { d.handleConn(tr) })
111	awaitFrame(t, sends, "Welcome")
112	awaitFrame(t, sends, "Output")
113
114	require.Eventually(t, func() bool {
115		select {
116		case written := <-writes:
117			return bytes.Contains(written, []byte("ping"))
118		default:
119			return false
120		}
121	}, 2*time.Second, time.Millisecond, "PTY input was not forwarded after the purge")
122	require.Eventually(t, func() bool {
123		sessions := listSessions(t, d)
124		return len(sessions.Sessions) == 1 && sessions.Sessions[0].Tabs == 2
125	}, 2*time.Second, time.Millisecond, "tab was not created after the purge")
126
127	releaseConn()
128	releaseSeed()
129	releaseAfter()
130	releaseTab()
131	hg.Wait()
132	d.sessWg.Wait()
133	d.waitNotifies()
134}
135
136// TestPurgeAllReportsStructuredPartialFailures proves failed stopped and
137// broken records are reported with their class and name without cancelling the
138// purge of the healthy live sibling, and without stopping the daemon.
139func TestPurgeAllReportsStructuredPartialFailures(t *testing.T) {
140	repository := newSelectivePurgeRepository(map[domain.IncarnationID]error{
141		{2}: errors.New("stopped delete failed"),
142		{3}: errors.New("broken delete failed"),
143	})
144	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
145	WithSnapshotRepository(repository)(d)
146	store, _ := newMockStore(t)
147	WithStore(t, store)(d)
148
149	live := newSnapshotTestSession(t, "live", false, "/live")
150	liveRecord := live.persistRecordLocked(1)
151	require.NoError(t, d.catalogue.Create(liveRecord))
152	livePTY, ok := live.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
153	require.True(t, ok)
154	livePTY.EXPECT().Close().Return(nil).Once()
155	d.sessions = map[domain.SessionID]*session{live.id: live}
156
157	stoppedRecord := domain.CatalogueRecord{Name: "stopped", IncarnationID: domain.IncarnationID{2}, CreatedAt: 7}
158	require.NoError(t, d.catalogue.Create(stoppedRecord))
159	d.inactive["stopped"] = inactiveSessionFromRecord(stoppedRecord, protocol.SessionDown, nil)
160	brokenRecord := domain.CatalogueRecord{Name: "broken", IncarnationID: domain.IncarnationID{3}, CreatedAt: 8}
161	require.NoError(t, d.catalogue.Create(brokenRecord))
162	// A broken registry state with a still-present durable record: the purge
163	// must classify the failure as broken, not stopped.
164	d.inactive["broken"] = inactiveSessionFromRecord(brokenRecord, protocol.SessionBroken, nil)
165
166	result := d.purgeAllSessions(protocol.ReasonSessionKilled)
167	require.True(t, result.Failed())
168	require.Len(t, result.Failures, 2)
169	require.Equal(t, purgeRecordStopped, result.Failures[0].Class)
170	require.Equal(t, "stopped", result.Failures[0].Name)
171	require.Equal(t, purgeRecordBroken, result.Failures[1].Class)
172	require.Equal(t, "broken", result.Failures[1].Name)
173	requirePurgeAdmissionBalanced(t, d)
174
175	require.Equal(t, 0, sessionCount(d), "the healthy live session must still be purged")
176	_, ok, err := d.catalogue.Record("live")
177	require.NoError(t, err)
178	require.False(t, ok)
179	d.mu.Lock()
180	_, stoppedRemains := d.inactive["stopped"]
181	_, brokenRemains := d.inactive["broken"]
182	closing := d.closing
183	d.mu.Unlock()
184	require.True(t, stoppedRemains, "a failed durable purge must leave its hidden record for retry")
185	require.True(t, brokenRemains, "a failed broken purge must leave its hidden record for retry")
186	require.False(t, closing, "a partial purge must not stop the daemon")
187}
188
189// TestPurgeAllReportsDurableOrphanFailure proves a durable catalogue record
190// with no live/stopped/broken registry entry is purged and, when its deletion
191// fails, reported as a durable-class failure.
192func TestPurgeAllReportsDurableOrphanFailure(t *testing.T) {
193	repository := newSelectivePurgeRepository(map[domain.IncarnationID]error{{9}: errors.New("orphan delete failed")})
194	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
195	WithSnapshotRepository(repository)(d)
196	store, _ := newMockStore(t)
197	WithStore(t, store)(d)
198
199	orphan := domain.CatalogueRecord{Name: "orphan", IncarnationID: domain.IncarnationID{9}, CreatedAt: 3}
200	require.NoError(t, d.catalogue.Create(orphan))
201
202	result := d.purgeAllSessions(protocol.ReasonSessionKilled)
203	require.True(t, result.Failed())
204	require.Len(t, result.Failures, 1)
205	require.Equal(t, purgeRecordDurable, result.Failures[0].Class)
206	require.Equal(t, "orphan", result.Failures[0].Name)
207}
208
209// TestPurgeAdmissionGatesAndReopens proves the transient gate: while a purge
210// holds admission a route waits and a move is rejected, and after the gate is
211// reopened both proceed.
212func TestPurgeAdmissionGatesAndReopens(t *testing.T) {
213	routePTY, releaseRoute := newBlockingPTY(t)
214	defer releaseRoute()
215	d := newTestDaemon(t, newFactorySeq(t, routePTY), stubClock{})
216	addControlSession(d, "work", "t_work", "p_work")
217	addNamedMoveDestination(d, "dest", "t_dest", "p_dest")
218	require.Equal(t, 2, sessionCount(d))
219
220	beginTestPurgeAdmission(t, d)
221
222	created := make(chan error, 1)
223	go func() {
224		_, _, err := d.route(helloResumeCapable(protocol.IntentEphemeral, "", 0), &closeTrackingTransport{})
225		created <- err
226	}()
227	select {
228	case err := <-created:
229		t.Fatalf("route bypassed the closed purge admission gate: %v", err)
230	case <-time.After(50 * time.Millisecond):
231	}
232	require.Equal(t, 2, sessionCount(d), "gated creation must not insert a session")
233
234	rejected := sendCommand(t, d, protocol.CommandRequest{
235		Slug: "move-pane", Args: []string{"dest", "t_dest"}, Self: true,
236		TargetSession: "work", TargetTab: "t_work", TargetPane: "p_work",
237	})
238	require.False(t, rejected.Outcome == protocol.CommandSucceeded, "a move during the purge must be rejected")
239
240	d.endPurgeAdmission()
241	require.NoError(t, awaitTestValue(t, created, "route did not resume after purge admission reopened"))
242	require.Equal(t, 3, sessionCount(d), "admission must reopen for creation")
243
244	accepted := sendCommand(t, d, protocol.CommandRequest{
245		Slug: "move-pane", Args: []string{"dest", "t_dest"}, Self: true,
246		TargetSession: "work", TargetTab: "t_work", TargetPane: "p_work",
247	})
248	require.True(t, accepted.Outcome == protocol.CommandSucceeded, accepted.Text)
249}
250
251// TestPurgeAllLinearizesConcurrentCreate races KillAll against a route create.
252// The gate drains the create before its exact snapshot or admits it after, so
253// the create never interleaves with the purge and the daemon stays active.
254func TestPurgeAllLinearizesConcurrentCreate(t *testing.T) {
255	sz := domain.Size{Cols: 80, Rows: 24}
256	for attempt := 0; attempt < 24; attempt++ {
257		seedPTY, releaseSeed := newBlockingPTY(t)
258		routePTY, releaseRoute := newBlockingPTY(t)
259		d := newTestDaemon(t, newFactorySeq(t, seedPTY, routePTY), stubClock{})
260
261		seed, err := createSessionForTest(d, "0", true, "/tmp", sz, terminalEnv{}, nil)
262		require.NoError(t, err)
263		require.NotNil(t, seed)
264
265		start := make(chan struct{})
266		var wg sync.WaitGroup
267		wg.Add(2)
268		var createErr error
269		go func() {
270			defer wg.Done()
271			<-start
272			_, _, createErr = d.route(helloResumeCapable(protocol.IntentEphemeral, "", 0), &closeTrackingTransport{})
273		}()
274		go func() {
275			defer wg.Done()
276			<-start
277			d.purgeAllSessions(protocol.ReasonSessionKilled)
278		}()
279		close(start)
280		wg.Wait()
281
282		require.NoError(t, createErr, "attempt %d: creation racing KillAll must be admitted, not rejected", attempt)
283		count := sessionCount(d)
284		require.LessOrEqual(t, count, 1, "attempt %d: purge must not leak sessions", attempt)
285		d.mu.Lock()
286		closing := d.closing
287		d.mu.Unlock()
288		require.False(t, closing, "attempt %d: KillAll must not stop the daemon", attempt)
289		select {
290		case <-d.done:
291			t.Fatalf("attempt %d: KillAll closed done", attempt)
292		default:
293		}
294
295		killAllSessions(d)
296		releaseSeed()
297		releaseRoute()
298		d.sessWg.Wait()
299		d.waitNotifies()
300	}
301}
302
303// TestPurgeAllLinearizesConcurrentRestore races KillAll against a resume of the
304// only stopped record. The restore is either purged with the set or ordered
305// after it, and the daemon stays active either way.
306func TestPurgeAllLinearizesConcurrentRestore(t *testing.T) {
307	sz := domain.Size{Cols: 80, Rows: 24}
308	for attempt := 0; attempt < 16; attempt++ {
309		seedPTY, releaseSeed := newBlockingPTY(t)
310		restorePTY, releaseRestore := newBlockingPTY(t)
311		d := newTestDaemon(t, newFactorySeq(t, seedPTY, restorePTY), stubClock{})
312
313		seed, err := createSessionForTest(d, "work", false, "/tmp", sz, terminalEnv{}, nil)
314		require.NoError(t, err)
315		require.NoError(t, d.killSession(seed, protocol.ReasonServerShutdown, false))
316		require.Equal(t, 0, sessionCount(d))
317		d.mu.Lock()
318		_, stopped := d.inactive["work"]
319		d.mu.Unlock()
320		require.True(t, stopped)
321
322		start := make(chan struct{})
323		var wg sync.WaitGroup
324		wg.Add(2)
325		var restoreErr error
326		go func() {
327			defer wg.Done()
328			<-start
329			_, _, restoreErr = d.route(helloResumeCapable(protocol.IntentAttach, "work", 0), &closeTrackingTransport{})
330		}()
331		go func() {
332			defer wg.Done()
333			<-start
334			d.purgeAllSessions(protocol.ReasonSessionKilled)
335		}()
336		close(start)
337		wg.Wait()
338
339		if restoreErr != nil {
340			var protoFailure *protoErr
341			require.ErrorAs(t, restoreErr, &protoFailure)
342			require.Equal(t, protocol.ErrNoSuchSession, protoFailure.code)
343		}
344		require.LessOrEqual(t, sessionCount(d), 1, "attempt %d: purge must not leak sessions", attempt)
345		d.mu.Lock()
346		closing := d.closing
347		d.mu.Unlock()
348		require.False(t, closing, "attempt %d: KillAll must not stop the daemon", attempt)
349
350		killAllSessions(d)
351		releaseSeed()
352		releaseRestore()
353		d.sessWg.Wait()
354		d.waitNotifies()
355	}
356}
357
358// TestShutdownAllStillStopsDaemonWithSessionsClientsAndWorkers proves the
359// explicit daemon-stop operation is unchanged: it removes the session, signals
360// Serve, marks closing, closes move admission, and cancels the daemon-wide pane
361// process context.
362func TestShutdownAllStillStopsDaemonWithSessionsClientsAndWorkers(t *testing.T) {
363	sz := domain.Size{Cols: 80, Rows: 24}
364	pty, releasePTY := newBlockingPTY(t)
365	d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
366
367	tr, sends, releaseConn := newConn(t, mustHello(protocol.IntentNew, "work", sz))
368	var hg sync.WaitGroup
369	hg.Go(func() { d.handleConn(tr) })
370	awaitFrame(t, sends, "Welcome")
371	awaitFrame(t, sends, "Output")
372	require.Equal(t, 1, sessionCount(d))
373
374	require.False(t, d.shutdownAll(protocol.ReasonServerShutdown))
375	require.Equal(t, 0, sessionCount(d))
376	select {
377	case <-d.done:
378	default:
379		t.Fatal("explicit daemon stop must signal Serve")
380	}
381	select {
382	case <-d.paneProcessCtx.Done():
383	default:
384		t.Fatal("explicit daemon stop must cancel the daemon-wide pane process context")
385	}
386	d.mu.Lock()
387	closing := d.closing
388	d.mu.Unlock()
389	require.True(t, closing)
390	d.moveLifecycleMu.Lock()
391	moveClosing := d.moveLifecycleClosing
392	d.moveLifecycleMu.Unlock()
393	require.True(t, moveClosing)
394
395	releaseConn()
396	releasePTY()
397	hg.Wait()
398	d.sessWg.Wait()
399	d.waitNotifies()
400}
401
402// beginTestPurgeAdmission closes the KillAll admission gate exactly as a purge
403// does once its drain succeeds. It fails the test if the gate could not be
404// granted (an admitted transition still holding it, shutdown, or cancellation).
405func beginTestPurgeAdmission(t *testing.T, d *Daemon) {
406	t.Helper()
407	if status := d.beginPurgeAdmission(context.Background(), nil); status != purgeAdmissionGranted {
408		t.Fatalf("test purge admission not granted: status %d", status)
409	}
410}
411
412// killAllSessions removes every remaining live session without asserting on the
413// result; it exists to release test PTYs deterministically.
414func killAllSessions(d *Daemon) {
415	d.mu.Lock()
416	sessions := d.sessionsSnapshotLocked()
417	d.mu.Unlock()
418	for _, sess := range sessions {
419		_ = d.killSession(sess, protocol.ReasonSessionKilled, true)
420	}
421}
422
423// gatedCheckpointRepository blocks LoadCheckpoint until released, so a test can
424// hold a restoration worker inside its admitted lifespan deterministically and
425// observe KillAll admission against it.
426type gatedCheckpointRepository struct {
427	noOpSnapshotRepository
428	generation ports.SnapshotGeneration
429	entered    chan struct{}
430	release    chan struct{}
431	once       sync.Once
432}
433
434func newGatedCheckpointRepository(generation ports.SnapshotGeneration) *gatedCheckpointRepository {
435	return &gatedCheckpointRepository{generation: generation, entered: make(chan struct{}), release: make(chan struct{})}
436}
437
438func (r *gatedCheckpointRepository) LoadCheckpoint(ctx context.Context, _ domain.IncarnationID, _ string, _ ports.CheckpointRef) (ports.SnapshotGeneration, error) {
439	r.once.Do(func() { close(r.entered) })
440	select {
441	case <-r.release:
442	case <-ctx.Done():
443		return ports.SnapshotGeneration{}, ctx.Err()
444	}
445	return r.generation, nil
446}
447
448// syncedLogBuffer collects log output for assertions without racing the
449// daemon's asynchronous sends.
450type syncedLogBuffer struct {
451	mu  sync.Mutex
452	buf bytes.Buffer
453}
454
455func (b *syncedLogBuffer) Write(p []byte) (int, error) {
456	b.mu.Lock()
457	defer b.mu.Unlock()
458	return b.buf.Write(p)
459}
460
461func (b *syncedLogBuffer) String() string {
462	b.mu.Lock()
463	defer b.mu.Unlock()
464	return b.buf.String()
465}
466
467// TestPurgeAllWaitsForAdmittedRestoreThenPurges is the deterministic gated
468// checkpoint contract: KillAll blocks while a startup restoration worker holds
469// purge admission inside its load, then purges the entire exact set once the
470// restore settles. No phantom live session, stopped authority, or taken name
471// may survive.
472func TestPurgeAllWaitsForAdmittedRestoreThenPurges(t *testing.T) {
473	snapshot := restoreAcceptanceSession(t, "work")
474	generation := acceptanceGeneration(t, snapshot, 9)
475	repository := newGatedCheckpointRepository(generation)
476	pty, releasePTY := newBlockingPTY(t)
477	defer releasePTY()
478
479	d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
480	d.baseEnv = []string{"TERM=xterm-256color"}
481	checkpoint := domain.CheckpointRef{Generation: generation.Generation, ManifestDigest: snapcodec.ManifestDigest(generation.Manifest)}
482	record := domain.CatalogueRecord{Name: snapshot.Name, IncarnationID: generation.IncarnationID, Cwd: "/snapshot/cwd", CreatedAt: int64(snapshot.CreatedAt), Committed: &checkpoint}
483	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
484	WithCatalogue(catalogue, []domain.CatalogueRecord{record})(d)
485	WithSnapshotRepository(repository)(d)
486	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, repository, nil))(d)
487
488	restoreFinished := make(chan struct{})
489	go func() {
490		defer close(restoreFinished)
491		d.restoreIncrementalSnapshots(context.Background())
492	}()
493	<-repository.entered
494
495	purged := make(chan purgeResult, 1)
496	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
497	select {
498	case result := <-purged:
499		t.Fatalf("KillAll returned while an admitted restore was still loading: %v", result.summary())
500	case <-time.After(100 * time.Millisecond):
501	}
502
503	close(repository.release)
504	<-restoreFinished
505	result := awaitTestValue(t, purged, "KillAll did not complete after the admitted restore settled")
506	require.False(t, result.Superseded())
507	require.False(t, result.Failed(), result.summary())
508
509	require.Equal(t, 0, sessionCount(d), "no live authority may survive the purge")
510	d.mu.Lock()
511	_, stopped := d.inactive["work"]
512	closing := d.closing
513	d.mu.Unlock()
514	require.False(t, stopped, "no phantom stopped authority may survive the purge")
515	require.False(t, closing, "KillAll must leave the daemon serving")
516	select {
517	case <-d.done:
518		t.Fatal("KillAll must not signal Serve")
519	default:
520	}
521	_, ok, err := d.catalogue.Record("work")
522	require.NoError(t, err)
523	require.False(t, ok, "durable authority must be purged")
524
525	// The purged name is reusable by the same daemon.
526	d.mu.Lock()
527	taken := d.nameLiveOrStoppedLocked("work")
528	d.mu.Unlock()
529	require.False(t, taken, "the purged name must not remain taken")
530	d.sessWg.Wait()
531	d.waitNotifies()
532}
533
534// TestRestoreCatalogueAbandonsRecordsSupersededByPurge proves the epoch
535// revalidation: a restoration that captured records before a completed KillAll
536// must not resurrect them from the pre-purge record set.
537func TestRestoreCatalogueAbandonsRecordsSupersededByPurge(t *testing.T) {
538	record := durableRecoveryRecord(1)
539	record.Name = "work"
540	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
541	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
542	WithCatalogue(catalogue, []domain.CatalogueRecord{record})(d)
543	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil))(d)
544
545	epoch := d.purgeEpochSnapshot()
546	result := d.purgeAllSessions(protocol.ReasonSessionKilled)
547	require.False(t, result.Failed(), result.summary())
548	_, ok, err := catalogue.Record("work")
549	require.NoError(t, err)
550	require.False(t, ok, "the purge must remove the durable record")
551
552	d.restoreCatalogueEpoch(context.Background(), []domain.CatalogueRecord{record}, epoch)
553	require.Equal(t, 0, sessionCount(d))
554	d.mu.Lock()
555	_, stopped := d.inactive["work"]
556	d.mu.Unlock()
557	require.False(t, stopped, "a superseded record must not be resurrected as stopped authority")
558	d.sessWg.Wait()
559	d.waitNotifies()
560}
561
562// TestEnsureCatalogueRegistryEntryRejectsSupersededRestore proves both the
563// in-progress gate and the completed-purge epoch are revalidated before a
564// restore publishes any registry entry.
565func TestEnsureCatalogueRegistryEntryRejectsSupersededRestore(t *testing.T) {
566	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
567	record := durableRecoveryRecord(1)
568	record.Name = "work"
569
570	prePurgeEpoch := d.purgeEpochSnapshot()
571	beginTestPurgeAdmission(t, d)
572	done, ok := d.ensureCatalogueRegistryEntry(record, prePurgeEpoch)
573	require.False(t, ok, "a record must not be admitted while a purge owns the gate")
574	require.Nil(t, done)
575	d.endPurgeAdmission()
576
577	// A worker that waited out the purge still holds a stale boundary: the
578	// record set it captured was superseded, so it must stay abandoned.
579	done, ok = d.ensureCatalogueRegistryEntry(record, prePurgeEpoch)
580	require.False(t, ok, "a record captured before a completed purge must stay superseded")
581	require.Nil(t, done)
582
583	// A fresh boundary (post-purge) admits the record.
584	done, ok = d.ensureCatalogueRegistryEntry(record, d.purgeEpochSnapshot())
585	require.True(t, ok)
586	require.NotNil(t, done)
587	d.mu.Lock()
588	_, exists := d.inactive["work"]
589	d.mu.Unlock()
590	require.True(t, exists)
591}
592
593// TestRestoreCatalogueCancellationDrainsAdmission proves the shutdown interlock:
594// a restoration worker blocked on purge admission observes context
595// cancellation, drains its job queue without touching the registry, and lets
596// the producer and the worker join.
597func TestRestoreCatalogueCancellationDrainsAdmission(t *testing.T) {
598	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
599	record := durableRecoveryRecord(1)
600	record.Name = "work"
601
602	ctx, cancel := context.WithCancel(context.Background())
603	done := make(chan struct{})
604	beginTestPurgeAdmission(t, d)
605	go func() {
606		defer close(done)
607		d.restoreCatalogueEpoch(ctx, []domain.CatalogueRecord{record}, d.purgeEpochSnapshot())
608	}()
609	// The worker is parked on admission; cancellation must release it even
610	// though the purge gate is still closed.
611	cancel()
612	awaitTestValue(t, done, "restoration did not drain after cancellation")
613	d.endPurgeAdmission()
614
615	d.mu.Lock()
616	_, exists := d.inactive["work"]
617	d.mu.Unlock()
618	require.False(t, exists, "a cancelled restoration must not register a stopped entry")
619}
620
621// TestPurgeAllSerializesConcurrentKillAll proves purgeAllMu admits one KillAll
622// owner at a time: a second concurrent purge cannot enter the critical section
623// until the first releases it.
624func TestPurgeAllSerializesConcurrentKillAll(t *testing.T) {
625	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
626	entered := make(chan struct{})
627	releaseFirst := make(chan struct{})
628	var calls atomic.Int32
629	d.afterPurgeAdmission = func() {
630		if calls.Add(1) == 1 {
631			close(entered)
632			<-releaseFirst
633		}
634	}
635
636	first := make(chan purgeResult, 1)
637	go func() { first <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
638	awaitTestValue(t, entered, "first purge never entered its critical section")
639
640	second := make(chan purgeResult, 1)
641	go func() { second <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
642	select {
643	case <-second:
644		t.Fatal("a second KillAll entered the critical section while the first still owned it")
645	case <-time.After(100 * time.Millisecond):
646	}
647	close(releaseFirst)
648	require.False(t, awaitTestValue(t, first, "first purge did not finish").Failed())
649	require.False(t, awaitTestValue(t, second, "second purge did not run after the first released").Failed())
650	require.Equal(t, int32(2), calls.Load())
651}
652
653// TestCommandPathPurgeAdmissionMapsToRetryableError proves a command-path
654// create rejected by the purge gate surfaces a deterministic, retryable result
655// rather than silently succeeding or reusing a name verdict.
656func TestCommandPathPurgeAdmissionMapsToRetryableError(t *testing.T) {
657	routePTY, releaseRoute := newBlockingPTY(t)
658	defer releaseRoute()
659	d := newTestDaemon(t, newFactorySeq(t, routePTY), stubClock{})
660	addControlSession(d, "work", "t_work", "p_work")
661
662	beginTestPurgeAdmission(t, d)
663	result := sendCommand(t, d, protocol.CommandRequest{
664		Slug: "new-session", Args: []string{"fresh"}, Self: true,
665		TargetSession: "work", TargetTab: "t_work", TargetPane: "p_work",
666	})
667	require.False(t, result.Outcome == protocol.CommandSucceeded)
668	require.Equal(t, protocol.ErrInternal, result.Code)
669	require.Equal(t, errPurgeAdmissionClosed.Error(), result.Text)
670	d.endPurgeAdmission()
671}
672
673// TestRemoteTargetRouteWaitsForPurgeAdmission proves an exact remote-target
674// handoff is admitted like a local route: it blocks on the closed purge gate,
675// then proceeds after the gate reopens.
676func TestRemoteTargetRouteWaitsForPurgeAdmission(t *testing.T) {
677	sz := domain.Size{Cols: 80, Rows: 24}
678	pty, releasePTY := newBlockingPTY(t)
679	defer releasePTY()
680	d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
681	sess, err := createSessionForTest(d, "work", false, "/tmp", sz, terminalEnv{}, nil)
682	require.NoError(t, err)
683	sess.mu.Lock()
684	incarnation := sess.incarnation
685	tabID := sess.tabs[0].stableID
686	sess.mu.Unlock()
687	target := domain.RemoteSessionTarget{Endpoint: "arch", DisplayOrigin: "arch", LifecycleID: incarnation, SessionName: "work", LiveTabID: domain.TabStableID(tabID)}
688	tr, _ := newCapturingTransport(t)
689
690	beginTestPurgeAdmission(t, d)
691	result := make(chan error, 1)
692	go func() {
693		_, _, routeErr := d.routeWithContext(context.Background(), protocol.Hello{
694			Version: protocol.Version, Intent: protocol.IntentAttach, Name: "work", Size: sz,
695			SessionTarget: ptrSessionAttachTarget(protocol.SessionAttachTargetFromRemote(target)), EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned,
696		}, tr)
697		result <- routeErr
698	}()
699	select {
700	case routeErr := <-result:
701		t.Fatalf("remote target route bypassed the closed purge gate: %v", routeErr)
702	case <-time.After(50 * time.Millisecond):
703	}
704	d.endPurgeAdmission()
705	require.NoError(t, awaitTestValue(t, result, "remote target route did not resume after the gate reopened"))
706
707	_ = d.killSession(sess, protocol.ReasonSessionKilled, true)
708	releasePTY()
709	d.sessWg.Wait()
710	d.waitNotifies()
711}
712
713// TestConcurrentCreateAndExplicitShutdownInterlock restores a real concurrent
714// create-versus-explicit-shutdown test: whichever side wins the registry lock,
715// no session leaks, the daemon is irreversibly closing and signaled, and later
716// Hellos are rejected.
717func TestConcurrentCreateAndExplicitShutdownInterlock(t *testing.T) {
718	for attempt := 0; attempt < 24; attempt++ {
719		pty, releasePTY := newBlockingPTY(t)
720		d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
721
722		start := make(chan struct{})
723		var wg sync.WaitGroup
724		var createErr error
725		wg.Add(2)
726		go func() {
727			defer wg.Done()
728			<-start
729			_, _, createErr = d.route(helloResumeCapable(protocol.IntentEphemeral, "", 0), &closeTrackingTransport{})
730		}()
731		go func() {
732			defer wg.Done()
733			<-start
734			d.shutdownAll(protocol.ReasonServerShutdown)
735		}()
736		close(start)
737		wg.Wait()
738
739		if createErr != nil {
740			var protoFailure *protoErr
741			require.ErrorAs(t, createErr, &protoFailure, "attempt %d", attempt)
742			require.Equal(t, protocol.ErrServerShutdown, protoFailure.code, "attempt %d", attempt)
743		}
744		require.Equal(t, 0, sessionCount(d), "attempt %d: explicit shutdown must not leak a session", attempt)
745		d.mu.Lock()
746		closing := d.closing
747		d.mu.Unlock()
748		require.True(t, closing, "attempt %d: explicit shutdown must mark closing", attempt)
749		select {
750		case <-d.done:
751		default:
752			t.Fatalf("attempt %d: explicit shutdown must signal Serve", attempt)
753		}
754		_, _, afterErr := d.route(helloResumeCapable(protocol.IntentEphemeral, "", 0), &closeTrackingTransport{})
755		require.Error(t, afterErr, "attempt %d: a Hello after shutdown must be rejected", attempt)
756
757		releasePTY()
758		d.sessWg.Wait()
759		d.waitNotifies()
760	}
761}
762
763// TestPurgeAllSupersededByExplicitShutdownPreservesStopAuthority proves an
764// explicit daemon stop wins over a KillAll that has not started removing
765// anything: the stopped authority the stop preserved is left intact.
766func TestPurgeAllSupersededByExplicitShutdownPreservesStopAuthority(t *testing.T) {
767	sz := domain.Size{Cols: 80, Rows: 24}
768	pty, releasePTY := newBlockingPTY(t)
769	defer releasePTY()
770	d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
771	catalogue := newDurableRecoveryCatalogue(nil)
772	WithCatalogue(catalogue, nil)(d)
773	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, bytes.NewReader(bytes.Repeat([]byte{0x5a}, 32))))(d)
774
775	_, err := createSessionForTest(d, "work", false, "/tmp", sz, terminalEnv{}, nil)
776	require.NoError(t, err)
777	require.False(t, d.shutdownAll(protocol.ReasonServerShutdown))
778	require.Equal(t, 0, sessionCount(d))
779
780	result := d.purgeAllSessions(protocol.ReasonSessionKilled)
781	require.True(t, result.Superseded())
782	d.mu.Lock()
783	_, stopped := d.inactive["work"]
784	d.mu.Unlock()
785	require.True(t, stopped, "explicit stop must preserve stopped authority")
786	_, ok, err := catalogue.Record("work")
787	require.NoError(t, err)
788	require.True(t, ok, "explicit stop must preserve durable authority")
789
790	releasePTY()
791	d.sessWg.Wait()
792	d.waitNotifies()
793}
794
795// TestPurgeAllLogsFailedControlSend proves a purge failure result that cannot be
796// delivered is logged, so a client that never observes the explicit correlated
797// KillResult cannot silently infer success from a close.
798func TestPurgeAllLogsFailedControlSend(t *testing.T) {
799	repository := newSelectivePurgeRepository(map[domain.IncarnationID]error{{2}: errors.New("delete failed")})
800	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
801	var logged syncedLogBuffer
802	d.log = slog.New(slog.NewTextHandler(&logged, nil))
803	WithSnapshotRepository(repository)(d)
804	store, _ := newMockStore(t)
805	WithStore(t, store)(d)
806
807	record := domain.CatalogueRecord{Name: "stopped", IncarnationID: domain.IncarnationID{2}, CreatedAt: 7}
808	require.NoError(t, d.catalogue.Create(record))
809	d.inactive["stopped"] = inactiveSessionFromRecord(record, protocol.SessionDown, nil)
810
811	tr := newMockServerConnection(t)
812	tr.EXPECT().Send(mock.Anything).Return(errors.New("client gone")).Maybe()
813	tr.EXPECT().Close().Return(nil).Maybe()
814	d.handleKill(tr, protocol.Kill{RequestID: 1, Scope: protocol.KillAll})
815
816	require.Contains(t, logged.String(), "kill result")
817	require.Contains(t, logged.String(), "client gone")
818	require.Contains(t, logged.String(), "outcome="+strconv.Itoa(int(protocol.KillFailed)), "the send-failure log must name the unobserved outcome")
819	require.Contains(t, logged.String(), "code="+strconv.Itoa(int(protocol.ErrInternal)), "the send-failure log must name the unobserved code")
820}
821
822// TestPurgeResultSummaryBoundsWireDetail proves the partial-failure summary is
823// wire-safe: it reports the total count, only the first N failures, and how
824// many were omitted.
825func TestPurgeResultSummaryBoundsWireDetail(t *testing.T) {
826	result := purgeResult{}
827	for i := 0; i < purgeSummaryFailureLimit+3; i++ {
828		result.Failures = append(result.Failures, purgeFailure{Class: purgeRecordLive, Name: fmt.Sprintf("s%02d", i), Err: errors.New("boom")})
829	}
830	summary := result.summary()
831	require.Contains(t, summary, fmt.Sprintf("%d record(s) could not be purged", purgeSummaryFailureLimit+3))
832	require.Contains(t, summary, "; 3 more omitted")
833	require.Contains(t, summary, "s00")
834	require.NotContains(t, summary, "s07", "failures beyond the bound must be omitted from the wire detail")
835}
836
837// TestPurgeAllPurgesSettledRestore proves the other half of the policy: when the
838// admitted restore settles as a live session before KillAll begins, the purge's
839// exact snapshot includes it and removes it like any other live session.
840func TestPurgeAllPurgesSettledRestore(t *testing.T) {
841	snapshot := restoreAcceptanceSession(t, "work")
842	generation := acceptanceGeneration(t, snapshot, 4)
843	repository := newGatedCheckpointRepository(generation)
844	pty, releasePTY := newBlockingPTY(t)
845	defer releasePTY()
846
847	d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
848	d.baseEnv = []string{"TERM=xterm-256color"}
849	checkpoint := domain.CheckpointRef{Generation: generation.Generation, ManifestDigest: snapcodec.ManifestDigest(generation.Manifest)}
850	record := domain.CatalogueRecord{Name: snapshot.Name, IncarnationID: generation.IncarnationID, Cwd: "/snapshot/cwd", CreatedAt: int64(snapshot.CreatedAt), Committed: &checkpoint}
851	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
852	WithCatalogue(catalogue, []domain.CatalogueRecord{record})(d)
853	WithSnapshotRepository(repository)(d)
854	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, repository, nil))(d)
855
856	restoreFinished := make(chan struct{})
857	go func() {
858		defer close(restoreFinished)
859		d.restoreIncrementalSnapshots(context.Background())
860	}()
861	<-repository.entered
862	close(repository.release)
863	awaitTestValue(t, restoreFinished, "restore did not settle")
864	require.Equal(t, 1, sessionCount(d), "the settled restore must be live before the purge")
865
866	result := d.purgeAllSessions(protocol.ReasonSessionKilled)
867	require.False(t, result.Superseded())
868	require.False(t, result.Failed(), result.summary())
869	require.Equal(t, 0, sessionCount(d), "the settled restore must be purged")
870	_, ok, err := d.catalogue.Record("work")
871	require.NoError(t, err)
872	require.False(t, ok)
873	d.sessWg.Wait()
874	d.waitNotifies()
875}
876
877// TestPurgeAllConcurrentWithExplicitShutdown proves the KillAll-versus-KillDaemon
878// interlock terminates safely: whichever side wins, the daemon irreversibly
879// stops, no live session leaks, and Serve is signaled.
880func TestPurgeAllConcurrentWithExplicitShutdown(t *testing.T) {
881	sz := domain.Size{Cols: 80, Rows: 24}
882	for attempt := 0; attempt < 16; attempt++ {
883		pty, releasePTY := newBlockingPTY(t)
884		d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
885		_, err := createSessionForTest(d, "work", false, "/tmp", sz, terminalEnv{}, nil)
886		require.NoError(t, err)
887
888		start := make(chan struct{})
889		var wg sync.WaitGroup
890		var result purgeResult
891		wg.Add(2)
892		go func() {
893			defer wg.Done()
894			<-start
895			result = d.purgeAllSessions(protocol.ReasonSessionKilled)
896		}()
897		go func() {
898			defer wg.Done()
899			<-start
900			d.shutdownAll(protocol.ReasonServerShutdown)
901		}()
902		close(start)
903		wg.Wait()
904
905		require.Equal(t, 0, sessionCount(d), "attempt %d: no live session may survive the race", attempt)
906		d.mu.Lock()
907		closing := d.closing
908		d.mu.Unlock()
909		require.True(t, closing, "attempt %d: the explicit stop must win", attempt)
910		select {
911		case <-d.done:
912		default:
913			t.Fatalf("attempt %d: the explicit stop must signal Serve", attempt)
914		}
915		if !result.Superseded() {
916			require.False(t, result.Failed(), "attempt %d: %s", attempt, result.summary())
917		}
918		releasePTY()
919		d.sessWg.Wait()
920		d.waitNotifies()
921	}
922}
923
924// cancellationIgnoringRepository blocks LoadCheckpoint until released and
925// deliberately ignores context cancellation, modelling a repository call that
926// never returns. It proves the purge admission drain and daemon shutdown stay
927// bounded instead of waiting on an uncooperative transition.
928type cancellationIgnoringRepository struct {
929	noOpSnapshotRepository
930
931	generation ports.SnapshotGeneration
932	entered    chan struct{}
933	release    chan struct{}
934	once       sync.Once
935}
936
937func newCancellationIgnoringRepository(generation ports.SnapshotGeneration) *cancellationIgnoringRepository {
938	return &cancellationIgnoringRepository{
939		generation: generation,
940		entered:    make(chan struct{}),
941		release:    make(chan struct{}),
942	}
943}
944
945func (r *cancellationIgnoringRepository) LoadCheckpoint(context.Context, domain.IncarnationID, string, ports.CheckpointRef) (ports.SnapshotGeneration, error) {
946	r.once.Do(func() { close(r.entered) })
947	<-r.release
948	return r.generation, nil
949}
950
951// awaitPurgeGateClosed waits until a purge has closed the transient admission
952// gate, so a test can interleave another transition deterministically without
953// polling on a real timer.
954func awaitPurgeGateClosed(t *testing.T, d *Daemon) {
955	t.Helper()
956	deadline := time.NewTimer(testWaitTimeout)
957	defer deadline.Stop()
958	for {
959		d.mu.Lock()
960		closed := d.purgeAdmissionClosing
961		changed := d.purgeAdmissionChangeLocked()
962		d.mu.Unlock()
963		if closed {
964			return
965		}
966		select {
967		case <-changed:
968		case <-deadline.C:
969			t.Fatal("purge never closed admission")
970		}
971	}
972}
973
974// requirePurgeAdmissionBalanced asserts a completed KillAll left no outstanding
975// create/restore reservation and re-opened the transient admission gate, so the
976// daemon keeps admitting transitions after the purge.
977func requirePurgeAdmissionBalanced(t *testing.T, d *Daemon) {
978	t.Helper()
979	d.mu.Lock()
980	active := d.purgeAdmissionActive
981	closing := d.purgeAdmissionClosing
982	d.mu.Unlock()
983	require.Zero(t, active, "a completed purge must leave no outstanding admission reservation")
984	require.False(t, closing, "a completed purge must re-open purge admission")
985}
986
987// TestRestoreCatalogueCaptureOrdersAfterPurgeNotResurrect reproduces the
988// capture-during-purge race deterministically: while a purge owns the admission
989// gate (closed but not yet snapshotted), a catalogue capture must block on the
990// gate. It can then only observe the post-purge catalogue, so it cannot
991// resurrect the record the purge removed with a boundary that looks current.
992func TestRestoreCatalogueCaptureOrdersAfterPurgeNotResurrect(t *testing.T) {
993	record := durableRecoveryRecord(1)
994	record.Name = "work"
995	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
996	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
997	WithCatalogue(catalogue, []domain.CatalogueRecord{record})(d)
998	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil))(d)
999
1000	// Hold one admitted transition so the purge closes the gate and parks in its
1001	// drain: the exact interleaving point where a capture previously slipped in.
1002	require.NoError(t, d.acquirePurgeAdmission(context.Background()))
1003	purged := make(chan purgeResult, 1)
1004	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
1005	awaitPurgeGateClosed(t, d)
1006
1007	captured := make(chan struct{})
1008	go func() {
1009		defer close(captured)
1010		d.restoreIncrementalSnapshots(context.Background())
1011	}()
1012	select {
1013	case <-captured:
1014		t.Fatal("catalogue capture bypassed the closed purge gate")
1015	case <-time.After(50 * time.Millisecond):
1016	}
1017	d.mu.Lock()
1018	_, published := d.inactive["work"]
1019	d.mu.Unlock()
1020	require.False(t, published, "no registry entry may be published while the purge owns the gate")
1021
1022	d.releasePurgeAdmission()
1023	require.False(t, awaitTestValue(t, purged, "purge did not complete after admission drained").Failed())
1024	awaitTestValue(t, captured, "capture did not finish after the purge reopened admission")
1025
1026	require.Equal(t, 0, sessionCount(d))
1027	d.mu.Lock()
1028	_, stopped := d.inactive["work"]
1029	d.mu.Unlock()
1030	require.False(t, stopped, "a capture ordered after the purge must not resurrect the record")
1031	_, ok, err := catalogue.Record("work")
1032	require.NoError(t, err)
1033	require.False(t, ok, "the purge must remove the durable record")
1034	d.sessWg.Wait()
1035	d.waitNotifies()
1036}
1037
1038// TestPurgeAllAdmissionDrainBoundedWithUncooperativeRepository proves a KillAll
1039// whose drain is blocked by a repository call that ignores cancellation returns
1040// a structured failure at its bounded budget, leaves the daemon serving, and
1041// does not prevent a later explicit shutdown from completing.
1042func TestPurgeAllAdmissionDrainBoundedWithUncooperativeRepository(t *testing.T) {
1043	snapshot := restoreAcceptanceSession(t, "work")
1044	generation := acceptanceGeneration(t, snapshot, 5)
1045	repository := newCancellationIgnoringRepository(generation)
1046	clock := &signalClock{timers: make(chan *signalTimer, 8)}
1047	pty, releasePTY := newBlockingPTY(t)
1048	defer releasePTY()
1049
1050	d := newTestDaemon(t, newFactorySeq(t, pty), clock)
1051	d.baseEnv = []string{"TERM=xterm-256color"}
1052	checkpoint := domain.CheckpointRef{Generation: generation.Generation, ManifestDigest: snapcodec.ManifestDigest(generation.Manifest)}
1053	record := domain.CatalogueRecord{Name: snapshot.Name, IncarnationID: generation.IncarnationID, Cwd: "/snapshot/cwd", CreatedAt: int64(snapshot.CreatedAt), Committed: &checkpoint}
1054	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
1055	WithCatalogue(catalogue, []domain.CatalogueRecord{record})(d)
1056	WithSnapshotRepository(repository)(d)
1057	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, repository, nil))(d)
1058
1059	restoreFinished := make(chan struct{})
1060	go func() {
1061		defer close(restoreFinished)
1062		d.restoreIncrementalSnapshots(context.Background())
1063	}()
1064	<-repository.entered
1065
1066	purged := make(chan purgeResult, 1)
1067	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
1068
1069	// The purge is parked in its admission drain; fire the bounded deadline.
1070	timer := awaitTestValue(t, clock.timers, "purge did not arm its bounded admission deadline")
1071	timer.ch <- time.Now()
1072
1073	result := awaitTestValue(t, purged, "KillAll did not return after its bounded drain")
1074	require.True(t, result.Failed(), "a bounded drain timeout must be a structured failure")
1075	require.False(t, result.Superseded())
1076	require.Contains(t, result.summary(), "did not drain")
1077
1078	d.mu.Lock()
1079	_, stopped := d.inactive["work"]
1080	closing := d.closing
1081	d.mu.Unlock()
1082	require.True(t, stopped, "the in-flight record must stay as stopped authority")
1083	require.False(t, closing, "a bounded purge abort must leave the daemon serving")
1084	_, ok, err := catalogue.Record("work")
1085	require.NoError(t, err)
1086	require.True(t, ok, "an aborted purge must not remove the durable record")
1087
1088	// An explicit stop owns teardown and completes even though the repository
1089	// still refuses to observe cancellation.
1090	shutdownDone := make(chan struct{})
1091	go func() {
1092		defer close(shutdownDone)
1093		d.shutdownAll(protocol.ReasonServerShutdown)
1094	}()
1095	awaitTestValue(t, shutdownDone, "explicit shutdown did not complete with a stuck repository")
1096	select {
1097	case <-d.done:
1098	default:
1099		t.Fatal("explicit shutdown must signal Serve")
1100	}
1101
1102	close(repository.release)
1103	awaitTestValue(t, restoreFinished, "restore did not settle after the repository released")
1104	d.sessWg.Wait()
1105	d.waitNotifies()
1106}
1107
1108// TestPurgeAllAdmissionDrainDefersToExplicitShutdown proves a KillAll parked in
1109// its admission drain is woken by an explicit stop and defers immediately,
1110// rather than waiting out its bounded timeout. This is what keeps Serve from
1111// wedging behind a purge when an admitted transition refuses to drain.
1112func TestPurgeAllAdmissionDrainDefersToExplicitShutdown(t *testing.T) {
1113	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
1114	require.NoError(t, d.acquirePurgeAdmission(context.Background()))
1115
1116	purged := make(chan purgeResult, 1)
1117	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
1118	awaitPurgeGateClosed(t, d)
1119
1120	d.shutdownAll(protocol.ReasonServerShutdown)
1121
1122	result := awaitTestValue(t, purged, "a parked purge was not woken by the explicit stop")
1123	require.True(t, result.Superseded(), "a parked purge must defer to the explicit stop")
1124	require.False(t, result.Failed())
1125	select {
1126	case <-d.done:
1127	default:
1128		t.Fatal("the explicit stop must signal Serve")
1129	}
1130	d.releasePurgeAdmission()
1131}
1132
1133// TestPurgeAllDefersToExplicitShutdownPerUnit proves explicit stop wins per
1134// unit: once a shutdown begins mid-purge, no further unit is removed, so the
1135// stop can preserve the not-yet-owned lifecycle. A unit whose teardown the
1136// purge already owns keeps purge authority and completes as a purge (its
1137// stopped authority is removed), while the not-yet-owned unit stays live for
1138// the stop to preserve.
1139func TestPurgeAllDefersToExplicitShutdownPerUnit(t *testing.T) {
1140	repository := newSelectivePurgeRepository(nil)
1141	var d *Daemon
1142	repository.onDelete = func() {
1143		d.mu.Lock()
1144		d.closing = true
1145		d.signalPurgeAdmissionChangedLocked()
1146		d.mu.Unlock()
1147	}
1148	d = newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
1149	catalogue := newDurableRecoveryCatalogue(nil)
1150	WithCatalogue(catalogue, nil)(d)
1151	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, repository, nil))(d)
1152
1153	for _, name := range []string{"a", "b"} {
1154		live := newSnapshotTestSession(t, name, false, "/"+name)
1155		require.NoError(t, catalogue.Create(live.persistRecordLocked(live.createdAt)))
1156		livePTY, ok := live.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
1157		require.True(t, ok)
1158		livePTY.EXPECT().Close().Return(nil).Maybe()
1159		d.sessions[live.id] = live
1160	}
1161
1162	result := d.purgeAllSessions(protocol.ReasonSessionKilled)
1163	require.True(t, result.Superseded(), "an explicit stop that begins mid-purge must win")
1164	require.Len(t, repository.calls, 1, "only the purge-owned unit may be removed before the stop wins")
1165
1166	d.mu.Lock()
1167	remaining := len(d.sessions)
1168	var survivor string
1169	for _, sess := range d.sessions {
1170		survivor = sess.name
1171	}
1172	_, aStopped := d.inactive["a"]
1173	_, bStopped := d.inactive["b"]
1174	d.mu.Unlock()
1175	require.Equal(t, 1, remaining, "exactly the not-yet-owned unit must survive the purge")
1176	require.NotEmpty(t, survivor)
1177	require.False(t, aStopped || bStopped, "the purge-owned unit must not retain stopped authority")
1178
1179	// The explicit stop now owns the surviving session and preserves it as
1180	// stopped authority instead of purging it.
1181	d.shutdownAll(protocol.ReasonServerShutdown)
1182	require.Equal(t, 0, sessionCount(d))
1183	d.mu.Lock()
1184	_, survivorStopped := d.inactive[survivor]
1185	d.mu.Unlock()
1186	require.True(t, survivorStopped, "the explicit stop must preserve the surviving session's authority")
1187	d.sessWg.Wait()
1188	d.waitNotifies()
1189}
1190
1191// occupySessionTeardown marks sess as owned by another lifecycle path without
1192// releasing it, so a bounded kill cannot acquire teardown ownership. The owning
1193// test releases it with finishTeardown. It models a slow unit parked exactly at
1194// the destructive boundary: the gate is frozen and teardown ownership is the
1195// only thing the deadline must outlast.
1196func occupySessionTeardown(sess *session) {
1197	sess.teardownMu.Lock()
1198	sess.teardownActive = true
1199	sess.teardownDone = make(chan struct{})
1200	sess.teardownMu.Unlock()
1201}
1202
1203// seedPurgeableLiveSession inserts a named live session with durable authority
1204// and a closed-safe PTY, then parks its teardown ownership so a purge must reach
1205// the destructive boundary before it can remove the unit.
1206func seedPurgeableLiveSession(t *testing.T, d *Daemon, catalogue *durableRecoveryCatalogue, name string) *session {
1207	t.Helper()
1208	live := newSnapshotTestSession(t, name, false, "/"+name)
1209	require.NoError(t, catalogue.Create(live.persistRecordLocked(live.createdAt)))
1210	livePTY, ok := live.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
1211	require.True(t, ok)
1212	livePTY.EXPECT().Close().Return(nil).Maybe()
1213	d.sessions[live.id] = live
1214	occupySessionTeardown(live)
1215	return live
1216}
1217
1218// TestPurgeAllLiveDeadlineSkipIsTypedAndRetryable proves the destructive
1219// boundary is observable: a live unit whose teardown ownership the shared live
1220// budget outlasts is reported as a typed live failure with its exact name, is
1221// left completely live with its durable authority intact, and is removed by a
1222// later purge once teardown ownership is available. The admission budget and the
1223// live budget are separate deadlines, so a fresh live budget is armed when the
1224// live phase begins.
1225func TestPurgeAllLiveDeadlineSkipIsTypedAndRetryable(t *testing.T) {
1226	clock := &signalClock{timers: make(chan *signalTimer, 64)}
1227	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
1228	catalogue := newDurableRecoveryCatalogue(nil)
1229	WithCatalogue(catalogue, nil)(d)
1230	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil))(d)
1231
1232	live := seedPurgeableLiveSession(t, d, catalogue, "work")
1233
1234	purged := make(chan purgeResult, 1)
1235	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
1236
1237	admission := awaitTestValue(t, clock.timers, "purge did not arm its admission deadline")
1238	liveBudget := awaitTestValue(t, clock.timers, "purge did not arm a separate live-teardown budget")
1239	require.NotSame(t, admission, liveBudget, "live teardown must spend its own budget, not the admission drain's")
1240	liveBudget.ch <- time.Now()
1241
1242	result := awaitTestValue(t, purged, "purge did not return after the live budget expired")
1243	require.True(t, result.Failed(), result.summary())
1244	require.False(t, result.Superseded())
1245	require.Len(t, result.Failures, 1)
1246	require.Equal(t, purgeRecordLive, result.Failures[0].Class)
1247	require.Equal(t, "work", result.Failures[0].Name, "a skipped unit must keep its exact name")
1248	require.ErrorIs(t, result.Failures[0].Err, errSessionKillDeadline)
1249
1250	require.Equal(t, 1, sessionCount(d), "a skipped live unit must stay live")
1251	d.mu.Lock()
1252	taken := d.nameLiveOrStoppedLocked("work")
1253	_, stopped := d.inactive["work"]
1254	d.mu.Unlock()
1255	require.True(t, taken, "the skipped name must remain taken and accurately reported")
1256	require.False(t, stopped, "a skipped live unit must not leave phantom stopped authority")
1257	requirePurgeAdmissionBalanced(t, d)
1258	_, ok, err := catalogue.Record("work")
1259	require.NoError(t, err)
1260	require.True(t, ok, "a skipped live unit must keep its durable authority for the retry")
1261
1262	// The retry has full destructive ownership and must purge the unit exactly.
1263	live.finishTeardown()
1264	retry := d.purgeAllSessions(protocol.ReasonSessionKilled)
1265	require.False(t, retry.Failed(), retry.summary())
1266	require.False(t, retry.Superseded())
1267	require.Equal(t, 0, sessionCount(d), "the retried purge must remove the unit")
1268	d.mu.Lock()
1269	taken = d.nameLiveOrStoppedLocked("work")
1270	d.mu.Unlock()
1271	require.False(t, taken, "the retried purge must free the purged name")
1272	_, ok, err = catalogue.Record("work")
1273	require.NoError(t, err)
1274	require.False(t, ok, "the retried purge must remove durable authority")
1275	d.sessWg.Wait()
1276	d.waitNotifies()
1277}
1278
1279// TestPurgeAllLiveBudgetSkipsEveryRemainingSlowUnit defines the N-unit behavior:
1280// one live budget is shared by every live unit. A single expiry outlasts all N
1281// parked units, so each remaining unit is skipped, reported as its own typed
1282// live failure with its exact name, and left live; no unit is silently reported
1283// as purged.
1284func TestPurgeAllLiveBudgetSkipsEveryRemainingSlowUnit(t *testing.T) {
1285	const units = 3
1286	clock := &signalClock{timers: make(chan *signalTimer, 64)}
1287	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
1288	catalogue := newDurableRecoveryCatalogue(nil)
1289	WithCatalogue(catalogue, nil)(d)
1290	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil))(d)
1291
1292	names := make([]string, 0, units)
1293	for i := range units {
1294		name := fmt.Sprintf("slow-%d", i)
1295		names = append(names, name)
1296		seedPurgeableLiveSession(t, d, catalogue, name)
1297	}
1298	require.Equal(t, units, sessionCount(d))
1299
1300	purged := make(chan purgeResult, 1)
1301	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
1302
1303	_ = awaitTestValue(t, clock.timers, "purge did not arm its admission deadline")
1304	liveBudget := awaitTestValue(t, clock.timers, "purge did not arm a separate live-teardown budget")
1305	liveBudget.ch <- time.Now()
1306
1307	result := awaitTestValue(t, purged, "purge did not return after the shared live budget expired")
1308	require.True(t, result.Failed(), result.summary())
1309	require.False(t, result.Superseded())
1310	require.Len(t, result.Failures, units, "every remaining unit must be reported")
1311	for i, failure := range result.Failures {
1312		require.Equal(t, purgeRecordLive, failure.Class)
1313		require.Equal(t, names[i], failure.Name)
1314		require.ErrorIs(t, failure.Err, errSessionKillDeadline)
1315	}
1316	require.Equal(t, units, sessionCount(d), "no unit may be removed after the shared live budget expired")
1317	for _, name := range names {
1318		_, ok, err := catalogue.Record(name)
1319		require.NoError(t, err)
1320		require.True(t, ok, "a skipped unit must keep its durable authority")
1321	}
1322
1323	d.mu.Lock()
1324	sessions := d.sessionsSnapshotLocked()
1325	d.mu.Unlock()
1326	for _, sess := range sessions {
1327		sess.finishTeardown()
1328	}
1329	for _, sess := range sessions {
1330		require.NoError(t, d.killSession(sess, protocol.ReasonSessionKilled, true))
1331	}
1332	require.Equal(t, 0, sessionCount(d))
1333	d.sessWg.Wait()
1334	d.waitNotifies()
1335}
1336
1337// TestTerminateAllForShutdownReportsDeadlineSkippedSession proves the shutdown
1338// verdict: a session that a bounded shutdown deadline skips before teardown
1339// ownership is reported as an incomplete checkpoint instead of silently
1340// counting as a completed stop. A later deadline-free pass completes the stop
1341// once the competing owner releases teardown.
1342func TestTerminateAllForShutdownReportsDeadlineSkippedSession(t *testing.T) {
1343	clock := &signalClock{timers: make(chan *signalTimer, 64)}
1344	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
1345
1346	live := newSnapshotTestSession(t, "0", true, "/tmp")
1347	pty, ok := live.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
1348	require.True(t, ok)
1349	pty.EXPECT().Close().Return(nil).Maybe()
1350	d.sessions[live.id] = live
1351	occupySessionTeardown(live)
1352
1353	deadline := newSnapshotShutdownDeadline(clock)
1354	defer deadline.stop()
1355	timer := awaitTestValue(t, clock.timers, "shutdown deadline did not arm")
1356	timer.ch <- time.Now()
1357
1358	require.True(t, d.terminateAllForShutdown(protocol.ReasonServerShutdown, deadline),
1359		"a session skipped at the teardown deadline must be reported as an incomplete checkpoint")
1360	require.Equal(t, 1, sessionCount(d), "the skipped session must remain live")
1361
1362	live.finishTeardown()
1363	require.False(t, d.terminateAllForShutdown(protocol.ReasonServerShutdown, nil))
1364	require.Equal(t, 0, sessionCount(d), "a deadline-free pass must complete the stop")
1365	select {
1366	case <-d.done:
1367	default:
1368		t.Fatal("the explicit stop must signal Serve")
1369	}
1370	d.sessWg.Wait()
1371	d.waitNotifies()
1372}
1373
1374// TestPurgeAdmissionDeadlineDefersToConcurrentExplicitStop proves the drain
1375// deadline select rechecks closing: when an explicit stop begins as the drain
1376// budget expires, the purge reports supersession instead of a drain timeout and
1377// removes nothing, so the stop keeps ownership of every lifecycle.
1378func TestPurgeAdmissionDeadlineDefersToConcurrentExplicitStop(t *testing.T) {
1379	clock := &signalClock{timers: make(chan *signalTimer, 64)}
1380	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
1381	catalogue := newDurableRecoveryCatalogue(nil)
1382	WithCatalogue(catalogue, nil)(d)
1383	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil))(d)
1384
1385	live := seedPurgeableLiveSession(t, d, catalogue, "work")
1386	// Hold one admitted transition so the purge parks in its bounded drain.
1387	require.NoError(t, d.acquirePurgeAdmission(context.Background()))
1388
1389	purged := make(chan purgeResult, 1)
1390	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
1391	awaitPurgeGateClosed(t, d)
1392
1393	// The stop begins exactly as the budget expires and is deliberately not
1394	// signalled through the admission change channel, so the deadline is the only
1395	// select case that can fire.
1396	d.mu.Lock()
1397	d.closing = true
1398	d.mu.Unlock()
1399	admission := awaitTestValue(t, clock.timers, "purge did not arm its admission deadline")
1400	admission.ch <- time.Now()
1401
1402	result := awaitTestValue(t, purged, "purge did not return after the drain budget expired")
1403	require.True(t, result.Superseded(), "an explicit stop concurrent with the deadline must win")
1404	require.False(t, result.Failed())
1405	require.Equal(t, 1, sessionCount(d), "a superseded purge must remove nothing")
1406
1407	d.mu.Lock()
1408	d.closing = false
1409	d.mu.Unlock()
1410	d.releasePurgeAdmission()
1411	live.finishTeardown()
1412	require.False(t, d.purgeAllSessions(protocol.ReasonSessionKilled).Failed())
1413	require.Equal(t, 0, sessionCount(d))
1414	d.sessWg.Wait()
1415	d.waitNotifies()
1416}
1417
1418// TestPurgeLiveDeadlineDefersToConcurrentExplicitStop proves the same recheck at
1419// the destructive boundary: a live unit whose budget expires as a stop begins is
1420// classified as superseded rather than a teardown failure, and it stays live for
1421// the stop to preserve.
1422func TestPurgeLiveDeadlineDefersToConcurrentExplicitStop(t *testing.T) {
1423	clock := &signalClock{timers: make(chan *signalTimer, 64)}
1424	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
1425	catalogue := newDurableRecoveryCatalogue(nil)
1426	WithCatalogue(catalogue, nil)(d)
1427	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil))(d)
1428
1429	live := seedPurgeableLiveSession(t, d, catalogue, "work")
1430
1431	purged := make(chan purgeResult, 1)
1432	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
1433	_ = awaitTestValue(t, clock.timers, "purge did not arm its admission deadline")
1434	liveBudget := awaitTestValue(t, clock.timers, "purge did not arm a separate live-teardown budget")
1435
1436	d.mu.Lock()
1437	d.closing = true
1438	d.mu.Unlock()
1439	liveBudget.ch <- time.Now()
1440
1441	result := awaitTestValue(t, purged, "purge did not return after the live budget expired")
1442	require.True(t, result.Superseded(), "the concurrent explicit stop must win over the live deadline")
1443	require.False(t, result.Failed())
1444	require.Equal(t, 1, sessionCount(d), "a superseded live unit must remain live")
1445
1446	d.mu.Lock()
1447	d.closing = false
1448	d.mu.Unlock()
1449	live.finishTeardown()
1450	require.False(t, d.purgeAllSessions(protocol.ReasonSessionKilled).Failed())
1451	require.Equal(t, 0, sessionCount(d))
1452	d.sessWg.Wait()
1453	d.waitNotifies()
1454}
1455
1456// blockingPurgeDeleteRepository blocks its first incarnation deletion until it
1457// is released and ignores context cancellation, so a bounded sweep can only
1458// finish by stopping its wait on the uncooperative delete.
1459type blockingPurgeDeleteRepository struct {
1460	noOpSnapshotRepository
1461
1462	entered chan struct{}
1463	release chan struct{}
1464	once    sync.Once
1465}
1466
1467func newBlockingPurgeDeleteRepository() *blockingPurgeDeleteRepository {
1468	return &blockingPurgeDeleteRepository{entered: make(chan struct{}), release: make(chan struct{})}
1469}
1470
1471func (r *blockingPurgeDeleteRepository) DeleteIncarnation(context.Context, domain.IncarnationID) error {
1472	r.once.Do(func() { close(r.entered) })
1473	<-r.release
1474	return nil
1475}
1476
1477// seedStoppedPurgeRecords publishes one stopped and one broken registry record
1478// with durable authority, so a stopped/broken sweep has real work to bound.
1479func seedStoppedPurgeRecords(t *testing.T, d *Daemon, catalogue *durableRecoveryCatalogue) {
1480	t.Helper()
1481	stopped := domain.CatalogueRecord{Name: "stopped", IncarnationID: domain.IncarnationID{2}, CreatedAt: 7}
1482	require.NoError(t, catalogue.Create(stopped))
1483	d.inactive["stopped"] = inactiveSessionFromRecord(stopped, protocol.SessionDown, nil)
1484	broken := domain.CatalogueRecord{Name: "broken", IncarnationID: domain.IncarnationID{3}, CreatedAt: 8}
1485	require.NoError(t, catalogue.Create(broken))
1486	d.inactive["broken"] = inactiveSessionFromRecord(broken, protocol.SessionBroken, nil)
1487}
1488
1489// TestPurgeAllParticipantChangeRetriesOnceAndPurgesExactly proves a participant
1490// change that races a live purge is never counted as success on its own: the
1491// first kill aborts with the participants-changed sentinel, the purge retries
1492// once against a fresh snapshot, and only that retry removes the exact unit.
1493// Both a detach and a resumed attachment incarnation (a token resume) are
1494// covered.
1495func TestPurgeAllParticipantChangeRetriesOnceAndPurgesExactly(t *testing.T) {
1496	tests := []struct {
1497		name   string
1498		change func(*Daemon, *session, *attachedClient)
1499	}{
1500		{
1501			name: "detach",
1502			change: func(d *Daemon, sess *session, ac *attachedClient) {
1503				require.True(t, d.detachIfCurrentTransport(sess, ac, ac.transportSnapshot()))
1504			},
1505		},
1506		{
1507			name: "token resume",
1508			change: func(d *Daemon, sess *session, ac *attachedClient) {
1509				fresh := &closeTrackingTransport{}
1510				ac.replaceTransport(fresh)
1511				ac.installTestAttachmentCapability(sess.captureAttachmentCapability(ac, fresh))
1512			},
1513		},
1514	}
1515	for _, test := range tests {
1516		t.Run(test.name, func(t *testing.T) {
1517			d, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
1518			catalogue := newDurableRecoveryCatalogue(nil)
1519			WithCatalogue(catalogue, nil)(d)
1520			WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil))(d)
1521			require.NoError(t, catalogue.Create(sess.persistRecordLocked(1)))
1522
1523			attempts := 0
1524			d.afterAttachmentEffectParticipantsSnapshotted = func(action string, _ []*attachedClient) {
1525				if action != "" {
1526					return
1527				}
1528				attempts++
1529				if attempts == 1 {
1530					test.change(d, sess, ac)
1531				}
1532			}
1533			t.Cleanup(func() { d.afterAttachmentEffectParticipantsSnapshotted = nil })
1534
1535			result := d.purgeAllSessions(protocol.ReasonSessionKilled)
1536			require.False(t, result.Failed(), result.summary())
1537			require.False(t, result.Superseded())
1538			require.Equal(t, 2, attempts, "the aborted first attempt must be retried once")
1539			require.Equal(t, 0, sessionCount(d), "only the fresh-snapshot retry may remove the unit")
1540			d.mu.Lock()
1541			taken := d.nameLiveOrStoppedLocked("work")
1542			_, stopped := d.inactive["work"]
1543			d.mu.Unlock()
1544			require.False(t, taken, "the retried purge must free the purged name")
1545			require.False(t, stopped, "a purged unit must not leave phantom stopped authority")
1546			_, ok, err := catalogue.Record("work")
1547			require.NoError(t, err)
1548			require.False(t, ok, "the retried purge must remove durable authority")
1549			d.sessWg.Wait()
1550			d.waitNotifies()
1551		})
1552	}
1553}
1554
1555// TestPurgeAllParticipantChurnIsTypedNotFalseSuccess proves the no-false-success
1556// contract: a live unit whose participants keep changing is reported as a typed
1557// live failure with its exact name, keeps its live identity and durable
1558// authority, and is removed only when the churn stops. The churn alternates a
1559// detach with a resumed membership, so every attempt's snapshot is stale.
1560func TestPurgeAllParticipantChurnIsTypedNotFalseSuccess(t *testing.T) {
1561	d, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
1562	catalogue := newDurableRecoveryCatalogue(nil)
1563	WithCatalogue(catalogue, nil)(d)
1564	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil))(d)
1565	require.NoError(t, catalogue.Create(sess.persistRecordLocked(1)))
1566
1567	var churn atomic.Bool
1568	churn.Store(true)
1569	hookCalls := 0
1570	d.afterAttachmentEffectParticipantsSnapshotted = func(action string, _ []*attachedClient) {
1571		if action != "" || !churn.Load() {
1572			return
1573		}
1574		hookCalls++
1575		if hookCalls%2 == 1 {
1576			require.True(t, d.detachIfCurrentTransport(sess, ac, ac.transportSnapshot()))
1577			return
1578		}
1579		// Resumed incarnation: re-register the attachment after the first kill
1580		// snapshotted the detached set.
1581		d.mu.Lock()
1582		sess.mu.Lock()
1583		if sess.attachments == nil {
1584			sess.attachments = make(map[*attachedClient]struct{})
1585		}
1586		sess.attachments[ac] = struct{}{}
1587		sess.mu.Unlock()
1588		d.mu.Unlock()
1589		ac.setSession(sess)
1590	}
1591	t.Cleanup(func() { d.afterAttachmentEffectParticipantsSnapshotted = nil })
1592
1593	result := d.purgeAllSessions(protocol.ReasonSessionKilled)
1594	require.True(t, result.Failed(), result.summary())
1595	require.False(t, result.Superseded())
1596	require.Len(t, result.Failures, 1)
1597	require.Equal(t, purgeRecordLive, result.Failures[0].Class)
1598	require.Equal(t, "work", result.Failures[0].Name, "a churned unit must keep its exact name")
1599	require.ErrorIs(t, result.Failures[0].Err, errSessionKillParticipantsChanged)
1600	require.Equal(t, 2, hookCalls, "the churn must outlast the single retry")
1601	require.Equal(t, 1, sessionCount(d), "a churned unit must stay live")
1602	d.mu.Lock()
1603	taken := d.nameLiveOrStoppedLocked("work")
1604	_, stopped := d.inactive["work"]
1605	d.mu.Unlock()
1606	require.True(t, taken, "the churned name must remain taken and accurately reported")
1607	require.False(t, stopped, "a churned unit must not leave phantom stopped authority")
1608	requirePurgeAdmissionBalanced(t, d)
1609	_, ok, err := catalogue.Record("work")
1610	require.NoError(t, err)
1611	require.True(t, ok, "a churned unit must keep its durable authority for the retry")
1612
1613	churn.Store(false)
1614	retry := d.purgeAllSessions(protocol.ReasonSessionKilled)
1615	require.False(t, retry.Failed(), retry.summary())
1616	require.Equal(t, 0, sessionCount(d), "the churn-free retry must remove the unit")
1617	_, ok, err = catalogue.Record("work")
1618	require.NoError(t, err)
1619	require.False(t, ok, "the churn-free retry must remove durable authority")
1620	d.sessWg.Wait()
1621	d.waitNotifies()
1622}
1623
1624// TestPurgeAllAlreadyRemovedParticipantChangeIsSuccess proves the
1625// already-removed half of the contract: a session a racing lifecycle path
1626// removed while the purge's participant snapshot went stale owns nothing, so
1627// the purge reports success rather than a phantom live failure.
1628func TestPurgeAllAlreadyRemovedParticipantChangeIsSuccess(t *testing.T) {
1629	d, sess, _, _ := newManualSessionWithPTYs(t, newQuietPTY())
1630	catalogue := newDurableRecoveryCatalogue(nil)
1631	WithCatalogue(catalogue, nil)(d)
1632	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil))(d)
1633	require.NoError(t, catalogue.Create(sess.persistRecordLocked(1)))
1634
1635	d.afterAttachmentEffectParticipantsSnapshotted = func(action string, _ []*attachedClient) {
1636		if action != "" {
1637			return
1638		}
1639		d.mu.Lock()
1640		d.unregisterSessionLocked(sess)
1641		d.mu.Unlock()
1642	}
1643	t.Cleanup(func() { d.afterAttachmentEffectParticipantsSnapshotted = nil })
1644
1645	result := d.purgeAllSessions(protocol.ReasonSessionKilled)
1646	require.False(t, result.Failed(), result.summary())
1647	require.False(t, result.Superseded())
1648	require.Equal(t, 0, sessionCount(d), "the racing path already removed the session")
1649	d.mu.Lock()
1650	taken := d.nameLiveOrStoppedLocked("work")
1651	_, stopped := d.inactive["work"]
1652	d.mu.Unlock()
1653	require.False(t, taken, "an already-removed unit must not stay taken")
1654	require.False(t, stopped, "an already-removed unit must not leave phantom stopped authority")
1655	d.sessWg.Wait()
1656	d.waitNotifies()
1657}
1658
1659// TestPurgeAllStoppedSweepDeadlineIsTypedAndReleasesGate proves the stopped and
1660// broken sweep spends its own bounded budget: a repository deletion that
1661// ignores cancellation cannot extend the purge, every record the sweep did not
1662// finish is reported as its typed stopped/broken failure with its exact name,
1663// admission is released, the daemon keeps serving, and its next purge removes
1664// the leftovers.
1665func TestPurgeAllStoppedSweepDeadlineIsTypedAndReleasesGate(t *testing.T) {
1666	clock := &signalClock{timers: make(chan *signalTimer, 64)}
1667	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
1668	catalogue := newDurableRecoveryCatalogue(nil)
1669	repository := newBlockingPurgeDeleteRepository()
1670	WithCatalogue(catalogue, nil)(d)
1671	WithSnapshotRepository(repository)(d)
1672	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, repository, nil))(d)
1673	seedStoppedPurgeRecords(t, d, catalogue)
1674
1675	purged := make(chan purgeResult, 1)
1676	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
1677
1678	admission := awaitTestValue(t, clock.timers, "purge did not arm its admission deadline")
1679	stoppedBudget := awaitTestValue(t, clock.timers, "purge did not arm a separate stopped sweep budget")
1680	require.NotSame(t, admission, stoppedBudget, "the sweep must spend its own budget, not the admission drain's")
1681	awaitTestValue(t, repository.entered, "the sweep did not start its uncooperative delete")
1682	stoppedBudget.ch <- time.Now()
1683
1684	result := awaitTestValue(t, purged, "purge did not return after the stopped sweep budget expired")
1685	require.True(t, result.Failed(), result.summary())
1686	require.False(t, result.Superseded())
1687	require.Len(t, result.Failures, 2, "every unfinished stopped/broken record must be reported")
1688	require.Equal(t, purgeRecordStopped, result.Failures[0].Class)
1689	require.Equal(t, "stopped", result.Failures[0].Name)
1690	require.ErrorIs(t, result.Failures[0].Err, errSessionPurgeSweepDeadline)
1691	require.Equal(t, purgeRecordBroken, result.Failures[1].Class)
1692	require.Equal(t, "broken", result.Failures[1].Name)
1693	require.ErrorIs(t, result.Failures[1].Err, errSessionPurgeSweepDeadline)
1694
1695	d.mu.Lock()
1696	admissionStillClosed := d.purgeAdmissionClosing
1697	daemonClosing := d.closing
1698	_, stoppedRemains := d.inactive["stopped"]
1699	_, brokenRemains := d.inactive["broken"]
1700	d.mu.Unlock()
1701	require.False(t, admissionStillClosed, "a bounded sweep must reopen purge admission promptly")
1702	require.False(t, daemonClosing, "a bounded sweep must leave the daemon serving")
1703	requirePurgeAdmissionBalanced(t, d)
1704	require.True(t, stoppedRemains, "an unfinished stopped record must keep its authority for the retry")
1705	require.True(t, brokenRemains, "an unfinished broken record must keep its authority for the retry")
1706
1707	close(repository.release)
1708	// The detached sweep finishes the delete it had already started and then
1709	// stops; the leftover record is purged by the next bounded pass.
1710	require.Eventually(t, func() bool {
1711		d.mu.Lock()
1712		defer d.mu.Unlock()
1713		_, broken := d.inactive["broken"]
1714		return !broken
1715	}, testWaitTimeout, time.Millisecond, "the detached sweep did not finish the delete it had started")
1716	require.False(t, d.purgeAllSessions(protocol.ReasonSessionKilled).Failed())
1717	require.Equal(t, 0, sessionCount(d))
1718	d.mu.Lock()
1719	require.Empty(t, d.inactive, "the next purge must remove the leftover records")
1720	d.mu.Unlock()
1721	d.sessWg.Wait()
1722	d.waitNotifies()
1723}
1724
1725// TestPurgeAllStoppedAndLiveBudgetsAreIndependent proves the stopped/broken
1726// sweep, the live teardown, and the admission drain each spend a distinct
1727// budget: the stopped budget expiring leaves a parked live unit untouched and
1728// arms a fresh live budget, whose expiry then reports that live unit.
1729func TestPurgeAllStoppedAndLiveBudgetsAreIndependent(t *testing.T) {
1730	clock := &signalClock{timers: make(chan *signalTimer, 64)}
1731	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
1732	catalogue := newDurableRecoveryCatalogue(nil)
1733	repository := newBlockingPurgeDeleteRepository()
1734	WithCatalogue(catalogue, nil)(d)
1735	WithSnapshotRepository(repository)(d)
1736	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, repository, nil))(d)
1737	seedStoppedPurgeRecords(t, d, catalogue)
1738	live := seedPurgeableLiveSession(t, d, catalogue, "work")
1739
1740	purged := make(chan purgeResult, 1)
1741	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
1742
1743	admission := awaitTestValue(t, clock.timers, "purge did not arm its admission deadline")
1744	stoppedBudget := awaitTestValue(t, clock.timers, "purge did not arm a stopped sweep budget")
1745	require.NotSame(t, admission, stoppedBudget)
1746	awaitTestValue(t, repository.entered, "the sweep did not start its uncooperative delete")
1747	stoppedBudget.ch <- time.Now()
1748
1749	liveBudget := awaitTestValue(t, clock.timers, "the expired sweep did not arm a fresh live budget")
1750	require.NotSame(t, admission, liveBudget)
1751	require.NotSame(t, stoppedBudget, liveBudget)
1752	liveBudget.ch <- time.Now()
1753
1754	result := awaitTestValue(t, purged, "purge did not return after both budgets expired")
1755	require.True(t, result.Failed(), result.summary())
1756	require.False(t, result.Superseded())
1757	require.Len(t, result.Failures, 3)
1758	require.Equal(t, purgeRecordLive, result.Failures[0].Class)
1759	require.Equal(t, "work", result.Failures[0].Name)
1760	require.ErrorIs(t, result.Failures[0].Err, errSessionKillDeadline)
1761	require.Equal(t, purgeRecordStopped, result.Failures[1].Class)
1762	require.Equal(t, "stopped", result.Failures[1].Name)
1763	require.ErrorIs(t, result.Failures[1].Err, errSessionPurgeSweepDeadline)
1764	require.Equal(t, purgeRecordBroken, result.Failures[2].Class)
1765	require.Equal(t, "broken", result.Failures[2].Name)
1766	require.ErrorIs(t, result.Failures[2].Err, errSessionPurgeSweepDeadline)
1767	require.Equal(t, 1, sessionCount(d), "the swept live budget must leave the live unit untouched")
1768
1769	close(repository.release)
1770	live.finishTeardown()
1771	// The detached sweep finishes the delete it had already started and then
1772	// stops; the leftover record is purged by the next bounded pass.
1773	require.Eventually(t, func() bool {
1774		d.mu.Lock()
1775		defer d.mu.Unlock()
1776		_, broken := d.inactive["broken"]
1777		return !broken
1778	}, testWaitTimeout, time.Millisecond, "the detached sweep did not finish after release")
1779	require.False(t, d.purgeAllSessions(protocol.ReasonSessionKilled).Failed())
1780	require.Equal(t, 0, sessionCount(d))
1781	d.sessWg.Wait()
1782	d.waitNotifies()
1783}
1784
1785// TestPurgeAllStoppedSweepDeadlineDefersToExplicitShutdown proves explicit stop
1786// supersedes the sweep classification: when a shutdown begins as the stopped
1787// budget expires, the purge reports supersession rather than typed
1788// stopped/broken failures, so the stop keeps authority over every record.
1789func TestPurgeAllStoppedSweepDeadlineDefersToExplicitShutdown(t *testing.T) {
1790	clock := &signalClock{timers: make(chan *signalTimer, 64)}
1791	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
1792	catalogue := newDurableRecoveryCatalogue(nil)
1793	repository := newBlockingPurgeDeleteRepository()
1794	WithCatalogue(catalogue, nil)(d)
1795	WithSnapshotRepository(repository)(d)
1796	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, repository, nil))(d)
1797	seedStoppedPurgeRecords(t, d, catalogue)
1798
1799	purged := make(chan purgeResult, 1)
1800	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
1801
1802	_ = awaitTestValue(t, clock.timers, "purge did not arm its admission deadline")
1803	stoppedBudget := awaitTestValue(t, clock.timers, "purge did not arm a stopped sweep budget")
1804	awaitTestValue(t, repository.entered, "the sweep did not start its uncooperative delete")
1805
1806	// The stop begins exactly as the sweep budget expires and is deliberately not
1807	// signalled through the admission change channel, so the deadline is the only
1808	// select case that can fire.
1809	d.mu.Lock()
1810	d.closing = true
1811	d.mu.Unlock()
1812	stoppedBudget.ch <- time.Now()
1813
1814	result := awaitTestValue(t, purged, "purge did not return after the sweep budget expired")
1815	require.True(t, result.Superseded(), "an explicit stop concurrent with the sweep deadline must win")
1816	require.False(t, result.Failed())
1817
1818	d.mu.Lock()
1819	d.closing = false
1820	stoppedRemains := len(d.inactive)
1821	d.mu.Unlock()
1822	require.Equal(t, 2, stoppedRemains, "a superseded sweep must leave every record for the stop")
1823	close(repository.release)
1824	d.sessWg.Wait()
1825	d.waitNotifies()
1826}
1827
1828// TestPurgeAllParticipantChangeDefersToConcurrentExplicitShutdown proves
1829// explicit stop supersedes the participant-change classification: when a
1830// shutdown begins as the kill aborts, the purge neither retries against the
1831// stop nor reports a live failure, so the stop preserves the session.
1832func TestPurgeAllParticipantChangeDefersToConcurrentExplicitShutdown(t *testing.T) {
1833	d, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
1834	catalogue := newDurableRecoveryCatalogue(nil)
1835	WithCatalogue(catalogue, nil)(d)
1836	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil))(d)
1837	require.NoError(t, catalogue.Create(sess.persistRecordLocked(1)))
1838
1839	// The racing path changes the participants and begins an explicit stop at
1840	// once, exactly as the kill's post-freeze revalidation runs.
1841	hookCalls := 0
1842	d.afterAttachmentEffectParticipantsSnapshotted = func(action string, _ []*attachedClient) {
1843		if action != "" {
1844			return
1845		}
1846		hookCalls++
1847		require.True(t, d.detachIfCurrentTransport(sess, ac, ac.transportSnapshot()))
1848		d.mu.Lock()
1849		d.closing = true
1850		d.mu.Unlock()
1851	}
1852	t.Cleanup(func() { d.afterAttachmentEffectParticipantsSnapshotted = nil })
1853
1854	result := d.purgeAllSessions(protocol.ReasonSessionKilled)
1855	require.True(t, result.Superseded(), "an explicit stop concurrent with the participant change must win")
1856	require.False(t, result.Failed())
1857	require.Equal(t, 1, hookCalls, "a superseded purge must not retry against the stop")
1858	require.Equal(t, 1, sessionCount(d), "a superseded purge must leave the session for the stop")
1859
1860	d.mu.Lock()
1861	d.closing = false
1862	_, stopped := d.inactive["work"]
1863	d.mu.Unlock()
1864	require.False(t, stopped, "a superseded purge must not leave phantom stopped authority")
1865	d.sessWg.Wait()
1866	d.waitNotifies()
1867}
1868
1869// TestPurgeAllStopBetweenPrecheckAndOwnershipIsSupersededNotSuccess proves the
1870// no-false-success contract at the last window: when an explicit stop wins a
1871// live unit after the purge's per-unit precheck but before that unit attempts
1872// teardown ownership, purgeLiveSession sees an already-removed target and is a
1873// no-op. The purge must defer to the stop and report supersession instead of
1874// counting the stop-owned unit as a successful purge.
1875func TestPurgeAllStopBetweenPrecheckAndOwnershipIsSupersededNotSuccess(t *testing.T) {
1876	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
1877	catalogue := newDurableRecoveryCatalogue(nil)
1878	WithCatalogue(catalogue, nil)(d)
1879	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, noOpSnapshotRepository{}, nil))(d)
1880
1881	live := newSnapshotTestSession(t, "work", false, "/work")
1882	record := live.persistRecordLocked(live.createdAt)
1883	require.NoError(t, catalogue.Create(record))
1884	livePTY, ok := live.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
1885	require.True(t, ok)
1886	livePTY.EXPECT().Close().Return(nil).Maybe()
1887	d.sessions[live.id] = live
1888
1889	d.beforePurgeLiveSessionKill = func(sess *session) {
1890		d.beforePurgeLiveSessionKill = nil
1891		// The explicit stop wins the exact unit in this window: it marks the
1892		// daemon closing, removes the session from the live registry, and
1893		// preserves its stopped authority, exactly as its own teardown does.
1894		d.mu.Lock()
1895		d.closing = true
1896		d.signalPurgeAdmissionChangedLocked()
1897		d.unregisterSessionLocked(sess)
1898		d.inactive["work"] = inactiveSessionFromRecord(record, protocol.SessionDown, nil)
1899		d.mu.Unlock()
1900	}
1901	t.Cleanup(func() { d.beforePurgeLiveSessionKill = nil })
1902
1903	result := d.purgeAllSessions(protocol.ReasonSessionKilled)
1904	require.True(t, result.Superseded(), "a stop that wins the unit before ownership must supersede the purge")
1905	require.False(t, result.Failed(), result.summary())
1906	require.Equal(t, 0, sessionCount(d), "the stop owns the session from this point")
1907	requirePurgeAdmissionBalanced(t, d)
1908
1909	d.mu.Lock()
1910	_, stopped := d.inactive["work"]
1911	closing := d.closing
1912	d.mu.Unlock()
1913	require.True(t, stopped, "the stop's preserved authority must survive the purge")
1914	require.True(t, closing, "the stop must stay in effect")
1915	_, ok, err := catalogue.Record("work")
1916	require.NoError(t, err)
1917	require.True(t, ok, "a superseded purge must not delete durable authority")
1918	d.waitNotifies()
1919}
1920
1921// blockingPublicationRepository blocks its first publication until released and
1922// deliberately ignores context cancellation, modelling an uncooperative
1923// publisher whose repository call outlives the shared teardown budget.
1924type blockingPublicationRepository struct {
1925	noOpSnapshotRepository
1926
1927	entered chan struct{}
1928	release chan struct{}
1929	once    sync.Once
1930}
1931
1932func newBlockingPublicationRepository() *blockingPublicationRepository {
1933	return &blockingPublicationRepository{entered: make(chan struct{}), release: make(chan struct{})}
1934}
1935
1936func (r *blockingPublicationRepository) Publish(context.Context, ports.SnapshotPublication) error {
1937	r.once.Do(func() { close(r.entered) })
1938	<-r.release
1939	return nil
1940}
1941
1942// TestPurgeAllLiveQuarantineJoinBoundedByDeadlineReleasesAdmission proves the
1943// purge's snapshot-coordinator quarantine join spends the shared live budget:
1944// an uncooperative publisher that ignores cancellation can never hold purge
1945// admission open, the live unit is reported as a typed skip instead of being
1946// counted as purged, and a later purge removes it once the publisher returns.
1947func TestPurgeAllLiveQuarantineJoinBoundedByDeadlineReleasesAdmission(t *testing.T) {
1948	clock := &signalClock{timers: make(chan *signalTimer, 64)}
1949	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
1950	catalogue := newDurableRecoveryCatalogue(nil)
1951	repository := newBlockingPublicationRepository()
1952	WithCatalogue(catalogue, nil)(d)
1953	WithSnapshotRepository(repository)(d)
1954	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, repository, nil))(d)
1955	startSnapshotEncodeWorker(t, d)
1956
1957	var releaseOnce sync.Once
1958	release := func() { releaseOnce.Do(func() { close(repository.release) }) }
1959	// Register after the worker cleanup so the uncooperative publication is
1960	// always released before cleanup asks the worker to join.
1961	t.Cleanup(release)
1962
1963	live := newSnapshotTestSession(t, "work", false, "/work")
1964	require.NoError(t, catalogue.Create(live.persistRecordLocked(live.createdAt)))
1965	livePTY, ok := live.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
1966	require.True(t, ok)
1967	livePTY.EXPECT().Close().Return(nil).Maybe()
1968	d.sessions[live.id] = live
1969
1970	markSnapshotDirty(live)
1971	require.True(t, d.scheduleSnapshot(live))
1972	awaitTestValue(t, repository.entered, "the live publication did not start")
1973
1974	purged := make(chan purgeResult, 1)
1975	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
1976
1977	_ = awaitTestValue(t, clock.timers, "purge did not arm its admission deadline")
1978	liveBudget := awaitTestValue(t, clock.timers, "purge did not arm a separate live-teardown budget")
1979	liveBudget.ch <- time.Now()
1980
1981	result := awaitTestValue(t, purged, "purge did not return while the publisher stayed blocked")
1982	require.True(t, result.Failed(), result.summary())
1983	require.False(t, result.Superseded())
1984	require.Len(t, result.Failures, 1)
1985	require.Equal(t, purgeRecordLive, result.Failures[0].Class)
1986	require.Equal(t, "work", result.Failures[0].Name)
1987	require.ErrorIs(t, result.Failures[0].Err, errSessionKillDeadline)
1988	require.Equal(t, 1, sessionCount(d), "a skipped live unit must stay live")
1989	requirePurgeAdmissionBalanced(t, d)
1990
1991	// The retry succeeds once the uncooperative publisher returns: the bound
1992	// released admission without discarding ownership of the unit.
1993	release()
1994	awaitSnapshotIdle(t, live)
1995	require.False(t, d.purgeAllSessions(protocol.ReasonSessionKilled).Failed())
1996	require.Equal(t, 0, sessionCount(d), "the retried purge must remove the unit")
1997	d.mu.Lock()
1998	_, stopped := d.inactive["work"]
1999	d.mu.Unlock()
2000	require.False(t, stopped, "the retried purge must not leave phantom stopped authority")
2001	d.sessWg.Wait()
2002	d.waitNotifies()
2003}
2004
2005// TestPurgeAllLiveQuarantineAbortRestoresSurvivingSession proves the reversible
2006// half of the quarantine contract: when the shared live budget expires while a
2007// purge is joining an uncooperative snapshot publication, the abort rolls the
2008// coordinator quarantine back instead of leaving a registered live session
2009// permanently ineligible. The surviving session checkpoints again, an EOF
2010// teardown completes, and its attached client is never left frozen.
2011func TestPurgeAllLiveQuarantineAbortRestoresSurvivingSession(t *testing.T) {
2012	clock := &signalClock{timers: make(chan *signalTimer, 64)}
2013	d, live, ac, _ := newManualSessionWithPTYsClock(t, clock, newQuietPTY())
2014	live.createdAt = 42
2015	live.snapEligible.Store(true)
2016
2017	catalogue := newDurableRecoveryCatalogue(nil)
2018	repository := newBlockingPublicationRepository()
2019	WithCatalogue(catalogue, nil)(d)
2020	WithSnapshotRepository(repository)(d)
2021	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, repository, nil))(d)
2022	startSnapshotEncodeWorker(t, d)
2023
2024	var releaseOnce sync.Once
2025	release := func() { releaseOnce.Do(func() { close(repository.release) }) }
2026	// Register after the worker cleanup so the uncooperative publication is
2027	// always released before cleanup asks the worker to join.
2028	t.Cleanup(release)
2029
2030	require.NoError(t, catalogue.Create(live.persistRecordLocked(live.createdAt)))
2031
2032	// Park an uncooperative publication so the purge's coordinator join cannot
2033	// finish inside the shared live budget.
2034	markSnapshotDirty(live)
2035	require.True(t, d.scheduleSnapshot(live))
2036	awaitTestValue(t, repository.entered, "the live publication did not start")
2037
2038	purged := make(chan purgeResult, 1)
2039	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
2040
2041	_ = awaitTestValue(t, clock.timers, "purge did not arm its admission deadline")
2042	liveBudget := awaitTestValue(t, clock.timers, "purge did not arm a separate live-teardown budget")
2043	liveBudget.ch <- time.Now()
2044
2045	result := awaitTestValue(t, purged, "purge did not return while the publisher stayed blocked")
2046	require.True(t, result.Failed(), result.summary())
2047	require.False(t, result.Superseded())
2048	require.Len(t, result.Failures, 1)
2049	require.Equal(t, purgeRecordLive, result.Failures[0].Class)
2050	require.Equal(t, "work", result.Failures[0].Name)
2051	require.ErrorIs(t, result.Failures[0].Err, errSessionKillDeadline)
2052	require.Equal(t, 1, sessionCount(d), "a skipped live unit must stay live")
2053	requirePurgeAdmissionBalanced(t, d)
2054
2055	// The abort rolled the quarantine back: the surviving session is eligible
2056	// and dirty again, and its attached client's gate is stable, not frozen.
2057	require.True(t, live.snapEligible.Load(), "an aborted purge must restore snapshot eligibility")
2058	live.snapshotMu.Lock()
2059	quarantined := live.snapshotQuarantined
2060	live.snapshotMu.Unlock()
2061	require.False(t, quarantined, "an aborted purge must roll back the coordinator quarantine")
2062	require.True(t, live.snapDirty.Load(), "an aborted purge must keep the surviving session dirty")
2063	ac.lifecycle.mu.Lock()
2064	phase := ac.lifecycle.phase
2065	ac.lifecycle.mu.Unlock()
2066	require.Equal(t, attachmentEffectsStable, phase, "an aborted purge must not leave the attached client frozen")
2067
2068	// Once the uncooperative publication returns, the surviving session must
2069	// checkpoint again.
2070	release()
2071	awaitSnapshotIdle(t, live)
2072	// The fake clock's routine pacing interval is zero-now based, so clear it and
2073	// prove the rolled-back coordinator schedules a real replacement capture.
2074	live.snapshotMu.Lock()
2075	live.snapshotNextEligibleAt = time.Time{}
2076	live.snapshotMu.Unlock()
2077	require.True(t, d.scheduleSnapshot(live), "the surviving session must schedule a replacement checkpoint")
2078	awaitSnapshotClean(t, live)
2079	_, ok, err := catalogue.Record("work")
2080	require.NoError(t, err)
2081	require.True(t, ok, "the surviving session must keep its durable authority")
2082
2083	// An EOF teardown of the surviving session must complete.
2084	require.NoError(t, d.killSession(live, protocol.ReasonSessionKilled, false))
2085	require.Equal(t, 0, sessionCount(d), "the EOF teardown must remove the surviving session")
2086	d.sessWg.Wait()
2087	d.waitNotifies()
2088}
2089
2090// TestSessionKillPostQuarantineParticipantMismatchRestoresSurvivingSession is
2091// the deterministic post-quarantine half of the reversible teardown contract:
2092// once a direct session kill or last-tab close has quarantined the snapshot
2093// coordinator, a participant change caught by the pre-publication revalidation
2094// aborts without removing anything and rolls the quarantine back. The surviving
2095// session resumes checkpoint scheduling and a later teardown completes.
2096func TestSessionKillPostQuarantineParticipantMismatchRestoresSurvivingSession(t *testing.T) {
2097	tests := []struct {
2098		name string
2099		kill func(t *testing.T, d *Daemon, sess *session) error
2100	}{
2101		{
2102			name: "direct kill",
2103			kill: func(_ *testing.T, d *Daemon, sess *session) error {
2104				return d.killSession(sess, protocol.ReasonSessionKilled, true)
2105			},
2106		},
2107		{
2108			name: "close tab",
2109			kill: func(_ *testing.T, d *Daemon, sess *session) error {
2110				return d.closeTab(sess, sess.tabs[0], false)
2111			},
2112		},
2113	}
2114	for _, tt := range tests {
2115		t.Run(tt.name, func(t *testing.T) {
2116			d, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
2117			sess.snapEligible.Store(true)
2118			catalogue := newDurableRecoveryCatalogue(nil)
2119			repository := noOpSnapshotRepository{}
2120			WithCatalogue(catalogue, nil)(d)
2121			WithSnapshotRepository(repository)(d)
2122			WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, repository, nil))(d)
2123			startSnapshotEncodeWorker(t, d)
2124			require.NoError(t, catalogue.Create(sess.persistRecordLocked(sess.createdAt)))
2125
2126			d.afterSessionKillQuarantine = func(*session) {
2127				d.afterSessionKillQuarantine = nil
2128				// A token resume replaces the exact transport after the snapshot
2129				// coordinator was quarantined but before its publication revalidation,
2130				// so the teardown must abort and roll the quarantine back. The
2131				// attachment gate is frozen here, so only the transport incarnation
2132				// changes; the capability is rebound by the next real claim.
2133				fresh := &closeTrackingTransport{}
2134				ac.replaceTransport(fresh)
2135			}
2136			t.Cleanup(func() { d.afterSessionKillQuarantine = nil })
2137
2138			require.ErrorIs(t, tt.kill(t, d, sess), errSessionKillParticipantsChanged)
2139			require.Equal(t, 1, sessionCount(d), "an aborted teardown must leave the session registered")
2140
2141			require.True(t, sess.snapEligible.Load(), "the abort must restore snapshot eligibility")
2142			sess.snapshotMu.Lock()
2143			quarantined := sess.snapshotQuarantined
2144			sess.snapshotMu.Unlock()
2145			require.False(t, quarantined, "the abort must roll back the coordinator quarantine")
2146			require.True(t, sess.snapDirty.Load(), "the abort must keep the surviving session checkpointable")
2147
2148			// The surviving session checkpoints again.
2149			require.True(t, d.scheduleSnapshot(sess), "the surviving session must schedule a checkpoint")
2150			awaitSnapshotClean(t, sess)
2151			_, ok, err := catalogue.Record("work")
2152			require.NoError(t, err)
2153			require.True(t, ok, "the surviving session must keep its durable authority")
2154
2155			// A later teardown of the surviving session completes.
2156			require.NoError(t, d.killSession(sess, protocol.ReasonSessionKilled, true))
2157			require.Equal(t, 0, sessionCount(d), "the later teardown must remove the surviving session")
2158			d.sessWg.Wait()
2159			d.waitNotifies()
2160		})
2161	}
2162}
2163
2164// blockingLiveDeleteRepository blocks its incarnation delete until released,
2165// ignores context cancellation, and signals when the blocked call returns, so a
2166// test can observe a detached live exact delete outliving the bounded purge.
2167type blockingLiveDeleteRepository struct {
2168	noOpSnapshotRepository
2169
2170	entered  chan struct{}
2171	release  chan struct{}
2172	returned chan struct{}
2173	// err is returned once release closes, letting a test model a detached
2174	// delete that fails only after the bounded purge already gave up waiting.
2175	err  error
2176	once sync.Once
2177}
2178
2179func newBlockingLiveDeleteRepository() *blockingLiveDeleteRepository {
2180	return &blockingLiveDeleteRepository{
2181		entered:  make(chan struct{}),
2182		release:  make(chan struct{}),
2183		returned: make(chan struct{}),
2184	}
2185}
2186
2187func (r *blockingLiveDeleteRepository) DeleteIncarnation(context.Context, domain.IncarnationID) error {
2188	r.once.Do(func() { close(r.entered) })
2189	<-r.release
2190	select {
2191	case <-r.returned:
2192	default:
2193		close(r.returned)
2194	}
2195	return r.err
2196}
2197
2198// TestPurgeAllLiveDurableDeleteBoundedDetachesAndReleasesGate proves a live
2199// unit's uncooperative exact durable delete cannot hold purgeAllMu, admission,
2200// or control: the purge returns a typed pending failure at the shared live
2201// budget while the live identity is already removed and its stopped authority
2202// still reserves the name. A second purge is not queued behind the blocked
2203// delete, and once the stale delete returns the name is free for a later
2204// same-name lifecycle.
2205func TestPurgeAllLiveDurableDeleteBoundedDetachesAndReleasesGate(t *testing.T) {
2206	clock := &signalClock{timers: make(chan *signalTimer, 64)}
2207	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
2208	catalogue := newDurableRecoveryCatalogue(nil)
2209	repository := newBlockingLiveDeleteRepository()
2210	WithCatalogue(catalogue, nil)(d)
2211	WithSnapshotRepository(repository)(d)
2212	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, repository, nil))(d)
2213
2214	live := newSnapshotTestSession(t, "work", false, "/work")
2215	require.NoError(t, catalogue.Create(live.persistRecordLocked(live.createdAt)))
2216	livePTY, ok := live.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
2217	require.True(t, ok)
2218	livePTY.EXPECT().Close().Return(nil).Maybe()
2219	d.sessions[live.id] = live
2220
2221	purged := make(chan purgeResult, 1)
2222	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
2223
2224	_ = awaitTestValue(t, clock.timers, "purge did not arm its admission deadline")
2225	liveBudget := awaitTestValue(t, clock.timers, "purge did not arm a separate live-teardown budget")
2226	awaitTestValue(t, repository.entered, "the live purge did not start its uncooperative delete")
2227	liveBudget.ch <- time.Now()
2228
2229	result := awaitTestValue(t, purged, "purge did not return while the live durable delete stayed blocked")
2230	require.True(t, result.Failed(), result.summary())
2231	require.False(t, result.Superseded())
2232	require.Len(t, result.Failures, 1)
2233	require.Equal(t, purgeRecordLive, result.Failures[0].Class)
2234	require.Equal(t, "work", result.Failures[0].Name, "the pending delete must keep its exact name")
2235	require.ErrorIs(t, result.Failures[0].Err, errSessionPurgeDeletePending)
2236
2237	require.Equal(t, 0, sessionCount(d), "the live identity must be removed before its durable delete")
2238	requirePurgeAdmissionBalanced(t, d)
2239	d.mu.Lock()
2240	closing := d.closing
2241	_, stoppedRemains := d.inactive["work"]
2242	d.mu.Unlock()
2243	require.False(t, closing, "a bounded live delete must leave the daemon serving")
2244	require.True(t, stoppedRemains, "a pending exact delete must reserve the name with stopped authority")
2245
2246	// A second purge is not queued behind the blocked delete: it takes the purge
2247	// admission gate and runs while the stale delete still holds the coordinator
2248	// lock, so it is bounded by its own sweep instead of waiting on purgeAllMu.
2249	second := make(chan purgeResult, 1)
2250	go func() { second <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
2251	awaitPurgeGateClosed(t, d)
2252
2253	// Releasing the stale delete lets the detached first finish and the second
2254	// purge settle. The durable record was already gone, so the second purge
2255	// clears the stale stopped authority without a new repository delete.
2256	close(repository.release)
2257	awaitTestValue(t, repository.returned, "the detached exact delete did not return")
2258	secondResult := awaitTestValue(t, second, "the second purge did not complete after the stale delete returned")
2259	require.False(t, secondResult.Failed(), secondResult.summary())
2260	require.False(t, secondResult.Superseded())
2261	d.mu.Lock()
2262	_, stoppedRemains = d.inactive["work"]
2263	taken := d.nameLiveOrStoppedLocked("work")
2264	d.mu.Unlock()
2265	require.False(t, stoppedRemains, "the second purge must clear the stale stopped authority")
2266	require.False(t, taken, "the purged name must be free for a later same-name lifecycle")
2267
2268	// A later same-name lifecycle is safe to adopt and keeps its own authority.
2269	replacement := newSnapshotTestSession(t, "work", false, "/work")
2270	replacement.incarnation = domain.IncarnationID{2}
2271	replacement.createdAt = 43
2272	require.NoError(t, catalogue.Create(replacement.persistRecordLocked(replacement.createdAt)))
2273	replacementPTY, ok := replacement.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
2274	require.True(t, ok)
2275	replacementPTY.EXPECT().Close().Return(nil).Maybe()
2276	d.sessions[replacement.id] = replacement
2277	record, ok, err := catalogue.Record("work")
2278	require.NoError(t, err)
2279	require.True(t, ok, "the replacement must own the same-name durable record")
2280	require.Equal(t, replacement.incarnation, record.IncarnationID)
2281	require.Equal(t, 1, sessionCount(d))
2282	d.sessWg.Wait()
2283	d.waitNotifies()
2284}
2285
2286// TestPurgeAllDetachedLiveDeleteFencesSameNameReplacement proves the detached
2287// live exact delete removes only the stopped authority whose name, incarnation,
2288// and createdAt it captured: a same-name replacement lifecycle that holds the
2289// name when the stale delete returns is never removed.
2290func TestPurgeAllDetachedLiveDeleteFencesSameNameReplacement(t *testing.T) {
2291	clock := &signalClock{timers: make(chan *signalTimer, 64)}
2292	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
2293	catalogue := newDurableRecoveryCatalogue(nil)
2294	repository := newBlockingLiveDeleteRepository()
2295	WithCatalogue(catalogue, nil)(d)
2296	WithSnapshotRepository(repository)(d)
2297	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, repository, nil))(d)
2298
2299	live := newSnapshotTestSession(t, "work", false, "/work")
2300	require.NoError(t, catalogue.Create(live.persistRecordLocked(live.createdAt)))
2301	livePTY, ok := live.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
2302	require.True(t, ok)
2303	livePTY.EXPECT().Close().Return(nil).Maybe()
2304	d.sessions[live.id] = live
2305
2306	purged := make(chan purgeResult, 1)
2307	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
2308
2309	_ = awaitTestValue(t, clock.timers, "purge did not arm its admission deadline")
2310	liveBudget := awaitTestValue(t, clock.timers, "purge did not arm a separate live-teardown budget")
2311	awaitTestValue(t, repository.entered, "the live purge did not start its uncooperative delete")
2312	liveBudget.ch <- time.Now()
2313	result := awaitTestValue(t, purged, "purge did not return while the live durable delete stayed blocked")
2314	require.True(t, result.Failed(), result.summary())
2315	require.Len(t, result.Failures, 1)
2316	require.ErrorIs(t, result.Failures[0].Err, errSessionPurgeDeletePending)
2317
2318	// A same-name replacement lifecycle takes the reserved name while the stale
2319	// delete is still blocked.
2320	replacement := inactiveSession{name: "work", createdAt: 99, incarnation: domain.IncarnationID{9}, purging: true}
2321	d.mu.Lock()
2322	d.inactive["work"] = replacement
2323	d.mu.Unlock()
2324
2325	close(repository.release)
2326	awaitTestValue(t, repository.returned, "the detached exact delete did not return")
2327	d.mu.Lock()
2328	current, ok := d.inactive["work"]
2329	d.mu.Unlock()
2330	require.True(t, ok, "the detached stale delete must not remove a same-name replacement lifecycle")
2331	require.Equal(t, replacement.incarnation, current.incarnation)
2332	d.sessWg.Wait()
2333	d.waitNotifies()
2334}
2335
2336// TestPurgeAllLogsDetachedLiveDeleteLateFailure proves a detached live exact
2337// delete that fails only after the bounded purge already returned a typed
2338// pending verdict is still recorded, using bounded non-sensitive fields instead
2339// of the user-chosen session name.
2340func TestPurgeAllLogsDetachedLiveDeleteLateFailure(t *testing.T) {
2341	clock := &signalClock{timers: make(chan *signalTimer, 64)}
2342	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
2343	var logged syncedLogBuffer
2344	d.log = slog.New(slog.NewTextHandler(&logged, nil))
2345	catalogue := newDurableRecoveryCatalogue(nil)
2346	repository := newBlockingLiveDeleteRepository()
2347	repository.err = errors.New("detached delete failed")
2348	WithCatalogue(catalogue, nil)(d)
2349	WithSnapshotRepository(repository)(d)
2350	WithRecoveryCoordinator(recoveryusecase.NewCoordinator(catalogue, repository, nil))(d)
2351
2352	live := newSnapshotTestSession(t, "work", false, "/work")
2353	require.NoError(t, catalogue.Create(live.persistRecordLocked(live.createdAt)))
2354	livePTY, ok := live.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
2355	require.True(t, ok)
2356	livePTY.EXPECT().Close().Return(nil).Maybe()
2357	d.sessions[live.id] = live
2358
2359	purged := make(chan purgeResult, 1)
2360	go func() { purged <- d.purgeAllSessions(protocol.ReasonSessionKilled) }()
2361
2362	_ = awaitTestValue(t, clock.timers, "purge did not arm its admission deadline")
2363	liveBudget := awaitTestValue(t, clock.timers, "purge did not arm a separate live-teardown budget")
2364	awaitTestValue(t, repository.entered, "the live purge did not start its uncooperative delete")
2365	liveBudget.ch <- time.Now()
2366
2367	result := awaitTestValue(t, purged, "purge did not return while the live durable delete stayed blocked")
2368	require.True(t, result.Failed(), result.summary())
2369	require.Len(t, result.Failures, 1)
2370	require.ErrorIs(t, result.Failures[0].Err, errSessionPurgeDeletePending)
2371
2372	// The stale delete fails after the bounded caller already returned, and the
2373	// detached waiter records the late failure.
2374	close(repository.release)
2375	awaitTestValue(t, repository.returned, "the detached exact delete did not return")
2376	require.Eventually(t, func() bool {
2377		return strings.Contains(logged.String(), "detached session purge delete failed after deadline")
2378	}, testWaitTimeout, time.Millisecond, "a late detached delete failure must be logged")
2379
2380	lateLine := ""
2381	for _, line := range strings.Split(logged.String(), "\n") {
2382		if strings.Contains(line, "detached session purge delete failed after deadline") {
2383			lateLine = line
2384		}
2385	}
2386	require.Contains(t, lateLine, "detached delete failed")
2387	require.NotContains(t, lateLine, "work", "the late failure must not log the user-chosen session name")
2388	d.sessWg.Wait()
2389	d.waitNotifies()
2390}