b87c753 merge(next): integrate history snapshot review fixes · vev

M internal/usecase/copy/mode.go +2 -1


@@ -53,7 +53,8 @@ func NewSnapshotFromRows(rows [][]renderer.Cell, width, height int) Snapshot {
53	53		// Len returns the number of document rows.
54	54		func (s Snapshot) Len() int { return s.history.Len() + s.screen.Height }
55	55
56		-	// Row returns document row i, or nil when i is out of range.
	56	+	// Row returns document row i, or nil when i is out of range. History rows
	57	+	// borrow immutable history storage and must be treated as read-only.
57	58		func (s Snapshot) Row(i int) []renderer.Cell {
58	59		if i < 0 {
59	60		return nil


@@ -53,7 +53,8 @@ func NewSnapshotFromRows(rows [][]renderer.Cell, width, height int) Snapshot {
53		// Len returns the number of document rows.	53		// Len returns the number of document rows.
54		func (s Snapshot) Len() int { return s.history.Len() + s.screen.Height }	54		func (s Snapshot) Len() int { return s.history.Len() + s.screen.Height }
55			55
56	-	// Row returns document row i, or nil when i is out of range.	56	+	// Row returns document row i, or nil when i is out of range. History rows
			57	+	// borrow immutable history storage and must be treated as read-only.
57		func (s Snapshot) Row(i int) []renderer.Cell {	58		func (s Snapshot) Row(i int) []renderer.Cell {
58		if i < 0 {	59		if i < 0 {
59		return nil	60		return nil

M internal/usecase/copy/mode_test.go +3 -3


@@ -610,9 +610,9 @@ func BenchmarkNewSnapshot10KRows(b *testing.B) {
610	610		historyRows = 10_000
611	611		)
612	612
613		-	scrollback := newHistory(historyRows)
	613	+	history := newHistory(historyRows)
614	614		for range historyRows {
615		-	scrollback.Append(row(strings.Repeat("x", width)))
	615	+	history.Append(row(strings.Repeat("x", width)))
616	616		}
617	617		screen := renderer.NewFrame(width, height)
618	618		for y := range height {

@@ -622,7 +622,7 @@ func BenchmarkNewSnapshot10KRows(b *testing.B) {
622	622		b.ReportAllocs()
623	623		b.ResetTimer()
624	624		for b.Loop() {
625		-	benchmarkSnapshotSink = NewSnapshot(scrollback, screen)
	625	+	benchmarkSnapshotSink = NewSnapshot(history, screen)
626	626		}
627	627		}
628	628


@@ -610,9 +610,9 @@ func BenchmarkNewSnapshot10KRows(b *testing.B) {
610		historyRows = 10_000	610		historyRows = 10_000
611		)	611		)
612			612
613	-	scrollback := newHistory(historyRows)	613	+	history := newHistory(historyRows)
614		for range historyRows {	614		for range historyRows {
615	-	scrollback.Append(row(strings.Repeat("x", width)))	615	+	history.Append(row(strings.Repeat("x", width)))
616		}	616		}
617		screen := renderer.NewFrame(width, height)	617		screen := renderer.NewFrame(width, height)
618		for y := range height {	618		for y := range height {

@@ -622,7 +622,7 @@ func BenchmarkNewSnapshot10KRows(b *testing.B) {
622		b.ReportAllocs()	622		b.ReportAllocs()
623		b.ResetTimer()	623		b.ResetTimer()
624		for b.Loop() {	624		for b.Loop() {
625	-	benchmarkSnapshotSink = NewSnapshot(scrollback, screen)	625	+	benchmarkSnapshotSink = NewSnapshot(history, screen)
626		}	626		}
627		}	627		}
628			628

M internal/usecase/daemon/daemon.go +10 -2


@@ -111,6 +111,11 @@ type Daemon struct {
111	111		snapshotWorkerCancel context.CancelFunc
112	112		snapshotWorkerDone chan struct{}
113	113		snapshotWorkerFlush chan struct{}
	114	+	snapshotWorkerFinalWake chan struct{}
	115	+	// snapshotFinalJobs retains immutable terminal captures rejected by the
	116	+	// bounded regular queue. It is only populated during session teardown and
	117	+	// drained by the same worker, so persistence remains outside daemon locks.
	118	+	snapshotFinalJobs []*snapshotCapture
114	119		snapshotWorkerClosing bool
115	120		snapshotWorkerInFlight *snapshotCapture
116	121		restoreDone chan struct{}

@@ -375,7 +380,7 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
375	380		})
376	381		}
377	382		if d.snapsEnabled {
378		-	d.startSnapshotEncodeWorker(d.serveCtx)
	383	+	d.startSnapshotEncodeWorker()
379	384		d.sessWg.Go(func() {
380	385		d.snapshotSaver(d.serveCtx)
381	386		})

@@ -424,8 +429,11 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
424	429		d.waitNotifies()
425	430		d.hardCancel()
426	431		d.serveCancel()
427		-	d.stopSnapshotEncodeWorker()
	432	+	// Wait for handlers before flushing snapshots: a handler that already
	433	+	// entered killSession may still submit its terminal capture after the
	434	+	// registry snapshot has been removed.
428	435		d.connWg.Wait()
	436	+	d.stopSnapshotEncodeWorker()
429	437		d.shutdownAll(ports.ReasonServerShutdown)
430	438		d.sessWg.Wait()
431	439		d.waitNotifies()

 	snapshotWorkerCancel    context.CancelFunc
 	snapshotWorkerDone      chan struct{}
 	snapshotWorkerFlush     chan struct{}
+	snapshotWorkerFinalWake chan struct{}
+	// snapshotFinalJobs retains immutable terminal captures rejected by the
+	// bounded regular queue. It is only populated during session teardown and
+	// drained by the same worker, so persistence remains outside daemon locks.
+	snapshotFinalJobs       []*snapshotCapture
 	snapshotWorkerClosing   bool
 	snapshotWorkerInFlight  *snapshotCapture
 	restoreDone             chan struct{}
 		})
 	}
 	if d.snapsEnabled {
-		d.startSnapshotEncodeWorker(d.serveCtx)
+		d.startSnapshotEncodeWorker()
 		d.sessWg.Go(func() {
 			d.snapshotSaver(d.serveCtx)
 		})
 	d.waitNotifies()
 	d.hardCancel()
 	d.serveCancel()
-	d.stopSnapshotEncodeWorker()
+	// Wait for handlers before flushing snapshots: a handler that already
+	// entered killSession may still submit its terminal capture after the
+	// registry snapshot has been removed.
 	d.connWg.Wait()
+	d.stopSnapshotEncodeWorker()
 	d.shutdownAll(ports.ReasonServerShutdown)
 	d.sessWg.Wait()
 	d.waitNotifies()

M internal/usecase/daemon/performance_benchmark_test.go +2 -7

@@ -1,7 +1,6 @@
1	1		package daemon
2	2
3	3		import (
4		-	"context"
5	4		"fmt"
6	5		"io"
7	6		"runtime"

@@ -736,12 +735,8 @@ func newPerformanceFixture(t testing.TB, config performanceConfig) *performanceF
736	735		resizeSizes: [2]domain.Size{{Cols: 100, Rows: 30}, config.size},
737	736		}
738	737		WithSnapshotStore(fixture.snaps)(d)
739		-	snapshotCtx, cancelSnapshotWorker := context.WithCancel(context.Background())
740		-	d.startSnapshotEncodeWorker(snapshotCtx)
741		-	t.Cleanup(func() {
742		-	cancelSnapshotWorker()
743		-	d.stopSnapshotEncodeWorker()
744		-	})
	738	+	d.startSnapshotEncodeWorker()
	739	+	t.Cleanup(d.stopSnapshotEncodeWorker)
745	740		for tabIndex, tb := range sess.tabs {
746	741		fixture.configureTab(tb, tabIndex, config.panes, config.historyRows, tabSize(config.size))
747	742		}

 package daemon
 import (
-	"context"
 	"fmt"
 	"io"
 	"runtime"
 		resizeSizes: [2]domain.Size{{Cols: 100, Rows: 30}, config.size},
 	}
 	WithSnapshotStore(fixture.snaps)(d)
-	snapshotCtx, cancelSnapshotWorker := context.WithCancel(context.Background())
+	d.startSnapshotEncodeWorker()
-	d.startSnapshotEncodeWorker(snapshotCtx)
+	t.Cleanup(d.stopSnapshotEncodeWorker)
-	t.Cleanup(func() {
-		cancelSnapshotWorker()
-		d.stopSnapshotEncodeWorker()
-	})
 	for tabIndex, tb := range sess.tabs {
 		fixture.configureTab(tb, tabIndex, config.panes, config.historyRows, tabSize(config.size))
 	}

M internal/usecase/daemon/session.go +1 -1


@@ -679,7 +679,7 @@ func (d Daemon) killSession(sess session, reason uint8, purge bool) error {
679	679		}
680	680		} else {
681	681		markSnapshotDirty(sess)
682		-	d.scheduleSnapshot(sess)
	682	+	d.scheduleFinalSnapshot(sess)
683	683		}
684	684		}
685	685		d.mu.Lock()


@@ -679,7 +679,7 @@ func (d Daemon) killSession(sess session, reason uint8, purge bool) error {
679		}	679		}
680		} else {	680		} else {
681		markSnapshotDirty(sess)	681		markSnapshotDirty(sess)
682	-	d.scheduleSnapshot(sess)	682	+	d.scheduleFinalSnapshot(sess)
683		}	683		}
684		}	684		}
685		d.mu.Lock()	685		d.mu.Lock()

M internal/usecase/daemon/snapshot.go +65 -4


@@ -315,6 +315,17 @@ func restorePTYSize(contentRect domain.Rect, tabSize domain.Size) domain.Size {
315	315		// scheduleSnapshot captures a session only once while an immutable capture is
316	316		// queued or being written. It never waits for the bounded worker queue.
317	317		func (d Daemon) scheduleSnapshot(sess session) bool {
	318	+	return d.scheduleSnapshotWithFinalFallback(sess, false)
	319	+	}
	320	+
	321	+	// scheduleFinalSnapshot retains a terminal capture rejected by the regular
	322	+	// bounded queue. A removed session cannot be retried by snapshotSaver, so the
	323	+	// worker drains this fallback before it stops.
	324	+	func (d Daemon) scheduleFinalSnapshot(sess session) bool {
	325	+	return d.scheduleSnapshotWithFinalFallback(sess, true)
	326	+	}
	327	+
	328	+	func (d Daemon) scheduleSnapshotWithFinalFallback(sess session, final bool) bool {
318	329		if d == nil \|\| sess == nil \|\| !d.snapsEnabled \|\| d.snaps == nil \|\| !sess.snapEligible.Load() {
319	330		return true
320	331		}

@@ -333,7 +344,7 @@ func (d Daemon) scheduleSnapshot(sess session) bool {
333	344		d.finishSnapshotCapture(capture, false)
334	345		return false
335	346		}
336		-	if d.enqueueSnapshotCapture(capture) {
	347	+	if d.enqueueSnapshotCapture(capture) \|\| (final && d.enqueueFinalSnapshotCapture(capture)) {
337	348		return true
338	349		}
339	350		// Coalesce under saturation or shutdown: the latest state stays dirty and

@@ -359,6 +370,22 @@ func (d Daemon) enqueueSnapshotCapture(capture snapshotCapture) bool {
359	370		}
360	371		}
361	372
	373	+	// enqueueFinalSnapshotCapture hands terminal captures to the existing worker
	374	+	// without making session teardown wait behind its normal bounded queue.
	375	+	func (d Daemon) enqueueFinalSnapshotCapture(capture snapshotCapture) bool {
	376	+	d.snapshotWorkerMu.Lock()
	377	+	defer d.snapshotWorkerMu.Unlock()
	378	+	if d.snapshotWorkerClosing \|\| d.snapshotWorkerCancel == nil \|\| d.snapshotWorkerCtx == nil \|\| d.snapshotWorkerCtx.Err() != nil {
	379	+	return false
	380	+	}
	381	+	d.snapshotFinalJobs = append(d.snapshotFinalJobs, capture)
	382	+	select {
	383	+	case d.snapshotWorkerFinalWake <- struct{}{}:
	384	+	default:
	385	+	}
	386	+	return true
	387	+	}
	388	+
362	389		// captureSession rotates history tails and clones visible frames while holding
363	390		// each pane lock. The returned capture contains only immutable state; encoding
364	391		// and persistence are deliberately deferred to snapshotEncodeWorker.

@@ -477,8 +504,8 @@ func (d Daemon) finishSnapshotCapture(capture snapshotCapture, succeeded bool)
477	504		})
478	505		}
479	506
480		-	func (d *Daemon) startSnapshotEncodeWorker(ctx context.Context) {
481		-	if d == nil \|\| ctx.Err() != nil {
	507	+	func (d *Daemon) startSnapshotEncodeWorker() {
	508	+	if d == nil {
482	509		return
483	510		}
484	511		d.snapshotWorkerMu.Lock()

@@ -495,10 +522,13 @@ func (d *Daemon) startSnapshotEncodeWorker(ctx context.Context) {
495	522		d.snapshotWorkerCancel = cancel
496	523		d.snapshotWorkerDone = make(chan struct{})
497	524		d.snapshotWorkerFlush = make(chan struct{})
	525	+	d.snapshotWorkerFinalWake = make(chan struct{}, 1)
	526	+	d.snapshotFinalJobs = nil
498	527		d.snapshotWorkerClosing = false
499	528		d.snapshotWorkerInFlight = nil
500	529		done := d.snapshotWorkerDone
501	530		flush := d.snapshotWorkerFlush
	531	+	finalWake := d.snapshotWorkerFinalWake
502	532		d.snapshotWorkerMu.Unlock()
503	533		go func() {
504	534		defer close(done)

@@ -518,6 +548,17 @@ func (d *Daemon) startSnapshotEncodeWorker(ctx context.Context) {
518	548		d.finishSnapshotCapture(capture, err == nil && workerCtx.Err() == nil)
519	549		return workerCtx.Err() == nil
520	550		}
	551	+	drainFinal := func() bool {
	552	+	for {
	553	+	capture := d.takeFinalSnapshotCapture()
	554	+	if capture == nil {
	555	+	return true
	556	+	}
	557	+	if !write(capture) {
	558	+	return false
	559	+	}
	560	+	}
	561	+	}
521	562		flushQueued := func() {
522	563		for {
523	564		select {

@@ -526,6 +567,7 @@ func (d *Daemon) startSnapshotEncodeWorker(ctx context.Context) {
526	567		return
527	568		}
528	569		default:
	570	+	_ = drainFinal()
529	571		return
530	572		}
531	573		}

@@ -537,6 +579,10 @@ func (d *Daemon) startSnapshotEncodeWorker(ctx context.Context) {
537	579		case <-flush:
538	580		flushQueued()
539	581		return
	582	+	case <-finalWake:
	583	+	if !drainFinal() {
	584	+	return
	585	+	}
540	586		case capture := <-d.snapshotJobs:
541	587		if !write(capture) {
542	588		return

@@ -546,6 +592,18 @@ func (d *Daemon) startSnapshotEncodeWorker(ctx context.Context) {
546	592		}()
547	593		}
548	594
	595	+	func (d Daemon) takeFinalSnapshotCapture() snapshotCapture {
	596	+	d.snapshotWorkerMu.Lock()
	597	+	defer d.snapshotWorkerMu.Unlock()
	598	+	if len(d.snapshotFinalJobs) == 0 {
	599	+	return nil
	600	+	}
	601	+	capture := d.snapshotFinalJobs[0]
	602	+	d.snapshotFinalJobs[0] = nil
	603	+	d.snapshotFinalJobs = d.snapshotFinalJobs[1:]
	604	+	return capture
	605	+	}
	606	+
549	607		func (d Daemon) setSnapshotWorkerInFlight(workerID uint64, capture snapshotCapture) bool {
550	608		d.snapshotWorkerMu.Lock()
551	609		defer d.snapshotWorkerMu.Unlock()

@@ -601,9 +659,12 @@ func (d *Daemon) finishStoppedSnapshotWorker(abandoned bool) {
601	659		d.snapshotWorkerCancel = nil
602	660		d.snapshotWorkerDone = nil
603	661		d.snapshotWorkerFlush = nil
	662	+	d.snapshotWorkerFinalWake = nil
604	663		d.snapshotWorkerClosing = false
605	664		d.snapshotWorkerInFlight = nil
606		-	queued := make([]*snapshotCapture, 0, len(d.snapshotJobs))
	665	+	queued := make([]*snapshotCapture, 0, len(d.snapshotJobs)+len(d.snapshotFinalJobs))
	666	+	queued = append(queued, d.snapshotFinalJobs...)
	667	+	d.snapshotFinalJobs = nil
607	668		for {
608	669		select {
609	670		case capture := <-d.snapshotJobs:

 // scheduleSnapshot captures a session only once while an immutable capture is
 // queued or being written. It never waits for the bounded worker queue.
 func (d *Daemon) scheduleSnapshot(sess *session) bool {
+	return d.scheduleSnapshotWithFinalFallback(sess, false)
+}
+// scheduleFinalSnapshot retains a terminal capture rejected by the regular
+// bounded queue. A removed session cannot be retried by snapshotSaver, so the
+// worker drains this fallback before it stops.
+func (d *Daemon) scheduleFinalSnapshot(sess *session) bool {
+	return d.scheduleSnapshotWithFinalFallback(sess, true)
+}
+func (d *Daemon) scheduleSnapshotWithFinalFallback(sess *session, final bool) bool {
 	if d == nil || sess == nil || !d.snapsEnabled || d.snaps == nil || !sess.snapEligible.Load() {
 		return true
 	}
 		d.finishSnapshotCapture(capture, false)
 		return false
 	}
-	if d.enqueueSnapshotCapture(capture) {
+	if d.enqueueSnapshotCapture(capture) || (final && d.enqueueFinalSnapshotCapture(capture)) {
 		return true
 	}
 	// Coalesce under saturation or shutdown: the latest state stays dirty and
 	}
 }
+// enqueueFinalSnapshotCapture hands terminal captures to the existing worker
+// without making session teardown wait behind its normal bounded queue.
+func (d *Daemon) enqueueFinalSnapshotCapture(capture *snapshotCapture) bool {
+	d.snapshotWorkerMu.Lock()
+	defer d.snapshotWorkerMu.Unlock()
+	if d.snapshotWorkerClosing || d.snapshotWorkerCancel == nil || d.snapshotWorkerCtx == nil || d.snapshotWorkerCtx.Err() != nil {
+		return false
+	}
+	d.snapshotFinalJobs = append(d.snapshotFinalJobs, capture)
+	select {
+	case d.snapshotWorkerFinalWake <- struct{}{}:
+	default:
+	}
+	return true
+}
 // captureSession rotates history tails and clones visible frames while holding
 // each pane lock. The returned capture contains only immutable state; encoding
 // and persistence are deliberately deferred to snapshotEncodeWorker.
 	})
 }
-func (d *Daemon) startSnapshotEncodeWorker(ctx context.Context) {
+func (d *Daemon) startSnapshotEncodeWorker() {
-	if d == nil || ctx.Err() != nil {
+	if d == nil {
 		return
 	}
 	d.snapshotWorkerMu.Lock()
 	d.snapshotWorkerCancel = cancel
 	d.snapshotWorkerDone = make(chan struct{})
 	d.snapshotWorkerFlush = make(chan struct{})
+	d.snapshotWorkerFinalWake = make(chan struct{}, 1)
+	d.snapshotFinalJobs = nil
 	d.snapshotWorkerClosing = false
 	d.snapshotWorkerInFlight = nil
 	done := d.snapshotWorkerDone
 	flush := d.snapshotWorkerFlush
+	finalWake := d.snapshotWorkerFinalWake
 	d.snapshotWorkerMu.Unlock()
 	go func() {
 		defer close(done)
 			d.finishSnapshotCapture(capture, err == nil && workerCtx.Err() == nil)
 			return workerCtx.Err() == nil
 		}
+		drainFinal := func() bool {
+			for {
+				capture := d.takeFinalSnapshotCapture()
+				if capture == nil {
+					return true
+				}
+				if !write(capture) {
+					return false
+				}
+			}
+		}
 		flushQueued := func() {
 			for {
 				select {
 						return
 					}
 				default:
+					_ = drainFinal()
 					return
 				}
 			}
 			case <-flush:
 				flushQueued()
 				return
+			case <-finalWake:
+				if !drainFinal() {
+					return
+				}
 			case capture := <-d.snapshotJobs:
 				if !write(capture) {
 					return
 	}()
 }
+func (d *Daemon) takeFinalSnapshotCapture() *snapshotCapture {
+	d.snapshotWorkerMu.Lock()
+	defer d.snapshotWorkerMu.Unlock()
+	if len(d.snapshotFinalJobs) == 0 {
+		return nil
+	}
+	capture := d.snapshotFinalJobs[0]
+	d.snapshotFinalJobs[0] = nil
+	d.snapshotFinalJobs = d.snapshotFinalJobs[1:]
+	return capture
+}
 func (d *Daemon) setSnapshotWorkerInFlight(workerID uint64, capture *snapshotCapture) bool {
 	d.snapshotWorkerMu.Lock()
 	defer d.snapshotWorkerMu.Unlock()
 	d.snapshotWorkerCancel = nil
 	d.snapshotWorkerDone = nil
 	d.snapshotWorkerFlush = nil
+	d.snapshotWorkerFinalWake = nil
 	d.snapshotWorkerClosing = false
 	d.snapshotWorkerInFlight = nil
-	queued := make([]*snapshotCapture, 0, len(d.snapshotJobs))
+	queued := make([]*snapshotCapture, 0, len(d.snapshotJobs)+len(d.snapshotFinalJobs))
+	queued = append(queued, d.snapshotFinalJobs...)
+	d.snapshotFinalJobs = nil
 	for {
 		select {
 		case capture := <-d.snapshotJobs:

M internal/usecase/daemon/snapshot_test.go +81 -25


@@ -25,12 +25,8 @@ func TestSnapshotEncodeWorkerDefersMarshalUntilAfterCapture(t *testing.T) {
25	25		store := &channelSnapshotStore{writes: make(chan []byte, 1)}
26	26		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
27	27		WithSnapshotStore(store)(d)
28		-	ctx, cancel := context.WithCancel(t.Context())
29		-	t.Cleanup(func() {
30		-	cancel()
31		-	d.stopSnapshotEncodeWorker()
32		-	})
33		-	d.startSnapshotEncodeWorker(ctx)
	28	+	t.Cleanup(d.stopSnapshotEncodeWorker)
	29	+	d.startSnapshotEncodeWorker()
34	30
35	31		entered := make(chan struct{})
36	32		release := make(chan struct{})

@@ -72,11 +68,9 @@ func exerciseBlockedSnapshotWorker(t *testing.T, blockEncode bool) {
72	68		store := newGatedSnapshotStore()
73	69		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
74	70		WithSnapshotStore(store)(d)
75		-	ctx, cancel := context.WithCancel(t.Context())
76		-	d.startSnapshotEncodeWorker(ctx)
	71	+	d.startSnapshotEncodeWorker()
77	72		t.Cleanup(func() {
78	73		store.unblock()
79		-	cancel()
80	74		d.stopSnapshotEncodeWorker()
81	75		})
82	76		if blockEncode {

@@ -157,9 +151,7 @@ func TestFinalSnapshotFlushPersistsNewestNamedSessionBeforeWorkerTeardown(t *tes
157	151		store := &channelSnapshotStore{writes: make(chan []byte, 1)}
158	152		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
159	153		WithSnapshotStore(store)(d)
160		-	ctx, cancel := context.WithCancel(t.Context())
161		-	defer cancel()
162		-	d.startSnapshotEncodeWorker(ctx)
	154	+	d.startSnapshotEncodeWorker()
163	155
164	156		entered := make(chan struct{})
165	157		release := make(chan struct{})

@@ -169,7 +161,8 @@ func TestFinalSnapshotFlushPersistsNewestNamedSessionBeforeWorkerTeardown(t *tes
169	161		return snapcodec.Marshal(s)
170	162		}
171	163		sess := newSnapshotTestSession(t, "final", false, "/work")
172		-	pty := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
	164	+	pty, ok := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
	165	+	require.True(t, ok)
173	166		pty.EXPECT().Close().Return(nil).Once()
174	167		sess.tabs[0].panes["pane-1"].screen.Write([]byte("\rnewest"))
175	168		d.sessions[sess.id] = sess

@@ -202,9 +195,7 @@ func TestStopSnapshotEncodeWorkerDoesNotWaitForUncancellableWrite(t *testing.T)
202	195		clock := newFinalFlushClock()
203	196		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
204	197		WithSnapshotStore(store)(d)
205		-	ctx, cancel := context.WithCancel(t.Context())
206		-	defer cancel()
207		-	d.startSnapshotEncodeWorker(ctx)
	198	+	d.startSnapshotEncodeWorker()
208	199
209	200		sess := newSnapshotTestSession(t, "stuck-write", false, "/work")
210	201		require.True(t, d.captureSession(sess))

@@ -240,10 +231,8 @@ func TestStoppedSnapshotWorkerLeavesCaptureRetryable(t *testing.T) {
240	231		store := &channelSnapshotStore{writes: make(chan []byte, 1)}
241	232		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
242	233		WithSnapshotStore(store)(d)
243		-	ctx, cancel := context.WithCancel(t.Context())
244		-	d.startSnapshotEncodeWorker(ctx)
	234	+	d.startSnapshotEncodeWorker()
245	235		d.stopSnapshotEncodeWorker()
246		-	t.Cleanup(cancel)
247	236
248	237		sess := newSnapshotTestSession(t, "stopped", false, "/work")
249	238		markSnapshotDirty(sess)

@@ -287,6 +276,77 @@ func TestSnapshotQueueSaturationLeavesCaptureRetryable(t *testing.T) {
287	276		awaitSnapshotClean(t, third)
288	277		}
289	278
	279	+	func TestNamedFinalSnapshotSurvivesSaturatedQueue(t *testing.T) {
	280	+	for _, tc := range []struct {
	281	+	name string
	282	+	stop func(t testing.T, d Daemon, sess *session)
	283	+	}{
	284	+	{
	285	+	name: "kill session",
	286	+	stop: func(t testing.T, d Daemon, sess *session) {
	287	+	t.Helper()
	288	+	require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, false))
	289	+	},
	290	+	},
	291	+	{
	292	+	name: "shutdown all",
	293	+	stop: func(t testing.T, d Daemon, _ *session) {
	294	+	t.Helper()
	295	+	d.shutdownAll(ports.ReasonServerShutdown)
	296	+	},
	297	+	},
	298	+	} {
	299	+	t.Run(tc.name, func(t *testing.T) {
	300	+	store := &channelSnapshotStore{writes: make(chan []byte, 3)}
	301	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	302	+	WithSnapshotStore(store)(d)
	303	+	startSnapshotEncodeWorker(t, d)
	304	+
	305	+	entered := make(chan struct{})
	306	+	release := make(chan struct{})
	307	+	var calls atomic.Int32
	308	+	d.snapshotMarshal = func(s snapcodec.Session) ([]byte, error) {
	309	+	if calls.Add(1) == 1 {
	310	+	close(entered)
	311	+	<-release
	312	+	}
	313	+	return snapcodec.Marshal(s)
	314	+	}
	315	+	first := newSnapshotTestSession(t, "first", false, "/work")
	316	+	second := newSnapshotTestSession(t, "second", false, "/work")
	317	+	final := newSnapshotTestSession(t, "final", false, "/work")
	318	+	finalPTY, ok := final.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
	319	+	require.True(t, ok)
	320	+	finalPTY.EXPECT().Close().Return(nil).Once()
	321	+	markSnapshotDirty(first)
	322	+	markSnapshotDirty(second)
	323	+	require.True(t, d.scheduleSnapshot(first))
	324	+	<-entered
	325	+	require.True(t, d.scheduleSnapshot(second))
	326	+	d.sessions[final.id] = final
	327	+
	328	+	tc.stop(t, d, final)
	329	+	require.NotContains(t, d.sessions, final.id, "final capture must survive session removal")
	330	+	close(release)
	331	+
	332	+	written := make(map[string]bool, 3)
	333	+	deadline := time.NewTimer(time.Second)
	334	+	defer deadline.Stop()
	335	+	for len(written) < 3 {
	336	+	select {
	337	+	case data := <-store.writes:
	338	+	snap, err := snapcodec.Unmarshal(data)
	339	+	require.NoError(t, err)
	340	+	written[snap.Name] = true
	341	+	case <-deadline.C:
	342	+	t.Fatalf("final snapshot was dropped: wrote %v", written)
	343	+	}
	344	+	}
	345	+	require.True(t, written[final.name])
	346	+	})
	347	+	}
	348	+	}
	349	+
290	350		func TestSnapshotCompletionSignalsWaiters(t *testing.T) {
291	351		store := &channelSnapshotStore{writes: make(chan []byte, 1)}
292	352		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})

@@ -846,12 +906,8 @@ func awaitSnapshotClean(t testing.T, sess session) {
846	906
847	907		func startSnapshotEncodeWorker(t testing.T, d Daemon) {
848	908		t.Helper()
849		-	ctx, cancel := context.WithCancel(t.Context())
850		-	d.startSnapshotEncodeWorker(ctx)
851		-	t.Cleanup(func() {
852		-	cancel()
853		-	d.stopSnapshotEncodeWorker()
854		-	})
	909	+	d.startSnapshotEncodeWorker()
	910	+	t.Cleanup(d.stopSnapshotEncodeWorker)
855	911		}
856	912
857	913		type gatedSnapshotStore struct {

 	store := &channelSnapshotStore{writes: make(chan []byte, 1)}
 	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
 	WithSnapshotStore(store)(d)
-	ctx, cancel := context.WithCancel(t.Context())
+	t.Cleanup(d.stopSnapshotEncodeWorker)
-	t.Cleanup(func() {
+	d.startSnapshotEncodeWorker()
-		cancel()
-		d.stopSnapshotEncodeWorker()
-	})
-	d.startSnapshotEncodeWorker(ctx)
 	entered := make(chan struct{})
 	release := make(chan struct{})
 	store := newGatedSnapshotStore()
 	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
 	WithSnapshotStore(store)(d)
-	ctx, cancel := context.WithCancel(t.Context())
+	d.startSnapshotEncodeWorker()
-	d.startSnapshotEncodeWorker(ctx)
 	t.Cleanup(func() {
 		store.unblock()
-		cancel()
 		d.stopSnapshotEncodeWorker()
 	})
 	if blockEncode {
 	store := &channelSnapshotStore{writes: make(chan []byte, 1)}
 	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
 	WithSnapshotStore(store)(d)
-	ctx, cancel := context.WithCancel(t.Context())
+	d.startSnapshotEncodeWorker()
-	defer cancel()
-	d.startSnapshotEncodeWorker(ctx)
 	entered := make(chan struct{})
 	release := make(chan struct{})
 		return snapcodec.Marshal(s)
 	}
 	sess := newSnapshotTestSession(t, "final", false, "/work")
-	pty := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
+	pty, ok := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
+	require.True(t, ok)
 	pty.EXPECT().Close().Return(nil).Once()
 	sess.tabs[0].panes["pane-1"].screen.Write([]byte("\rnewest"))
 	d.sessions[sess.id] = sess
 	clock := newFinalFlushClock()
 	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
 	WithSnapshotStore(store)(d)
-	ctx, cancel := context.WithCancel(t.Context())
+	d.startSnapshotEncodeWorker()
-	defer cancel()
-	d.startSnapshotEncodeWorker(ctx)
 	sess := newSnapshotTestSession(t, "stuck-write", false, "/work")
 	require.True(t, d.captureSession(sess))
 	store := &channelSnapshotStore{writes: make(chan []byte, 1)}
 	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
 	WithSnapshotStore(store)(d)
-	ctx, cancel := context.WithCancel(t.Context())
+	d.startSnapshotEncodeWorker()
-	d.startSnapshotEncodeWorker(ctx)
 	d.stopSnapshotEncodeWorker()
-	t.Cleanup(cancel)
 	sess := newSnapshotTestSession(t, "stopped", false, "/work")
 	markSnapshotDirty(sess)
 	awaitSnapshotClean(t, third)
 }
+func TestNamedFinalSnapshotSurvivesSaturatedQueue(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		stop func(t *testing.T, d *Daemon, sess *session)
+	}{
+		{
+			name: "kill session",
+			stop: func(t *testing.T, d *Daemon, sess *session) {
+				t.Helper()
+				require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, false))
+			},
+		},
+		{
+			name: "shutdown all",
+			stop: func(t *testing.T, d *Daemon, _ *session) {
+				t.Helper()
+				d.shutdownAll(ports.ReasonServerShutdown)
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			store := &channelSnapshotStore{writes: make(chan []byte, 3)}
+			d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
+			WithSnapshotStore(store)(d)
+			startSnapshotEncodeWorker(t, d)
+			entered := make(chan struct{})
+			release := make(chan struct{})
+			var calls atomic.Int32
+			d.snapshotMarshal = func(s snapcodec.Session) ([]byte, error) {
+				if calls.Add(1) == 1 {
+					close(entered)
+					<-release
+				}
+				return snapcodec.Marshal(s)
+			}
+			first := newSnapshotTestSession(t, "first", false, "/work")
+			second := newSnapshotTestSession(t, "second", false, "/work")
+			final := newSnapshotTestSession(t, "final", false, "/work")
+			finalPTY, ok := final.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
+			require.True(t, ok)
+			finalPTY.EXPECT().Close().Return(nil).Once()
+			markSnapshotDirty(first)
+			markSnapshotDirty(second)
+			require.True(t, d.scheduleSnapshot(first))
+			<-entered
+			require.True(t, d.scheduleSnapshot(second))
+			d.sessions[final.id] = final
+			tc.stop(t, d, final)
+			require.NotContains(t, d.sessions, final.id, "final capture must survive session removal")
+			close(release)
+			written := make(map[string]bool, 3)
+			deadline := time.NewTimer(time.Second)
+			defer deadline.Stop()
+			for len(written) < 3 {
+				select {
+				case data := <-store.writes:
+					snap, err := snapcodec.Unmarshal(data)
+					require.NoError(t, err)
+					written[snap.Name] = true
+				case <-deadline.C:
+					t.Fatalf("final snapshot was dropped: wrote %v", written)
+				}
+			}
+			require.True(t, written[final.name])
+		})
+	}
+}
 func TestSnapshotCompletionSignalsWaiters(t *testing.T) {
 	store := &channelSnapshotStore{writes: make(chan []byte, 1)}
 	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
 func startSnapshotEncodeWorker(t *testing.T, d *Daemon) {
 	t.Helper()
-	ctx, cancel := context.WithCancel(t.Context())
+	d.startSnapshotEncodeWorker()
-	d.startSnapshotEncodeWorker(ctx)
+	t.Cleanup(d.stopSnapshotEncodeWorker)
-	t.Cleanup(func() {
-		cancel()
-		d.stopSnapshotEncodeWorker()
-	})
 }
 type gatedSnapshotStore struct {