refactor(daemon): encode snapshots asynchronously

commit: 8151d785d3ee3593aa73512066cc1ee2fcd4569b
author: brice · 3mo ago · 2026-07-12 16:23 UTC
parent: 89d8fa0
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

7 files · +792 -60

M internal/usecase/copy/mode_test.go +85 -0


@@ -23,6 +23,14 @@ func row(text string) []renderer.Cell {
23	23
24	24		func newHistory(rows int) *vt.History { return vt.NewHistory(vt.HistoryConfig{MaxRows: rows}) }
25	25
	26	+	func rowText(cells []renderer.Cell) string {
	27	+	runes := make([]rune, len(cells))
	28	+	for i, cell := range cells {
	29	+	runes[i] = cell.Rune
	30	+	}
	31	+	return string(runes)
	32	+	}
	33	+
26	34		func snapshot(lines []string, height int) Snapshot {
27	35		rows := make([][]renderer.Cell, len(lines))
28	36		for i, line := range lines {

@@ -489,6 +497,83 @@ func TestOSC52FromBase64(t *testing.T) {
489	497		}
490	498		}
491	499
	500	+	func TestNewSnapshotFreezesHistoryAndVisibleScreen(t *testing.T) {
	501	+	history := newHistory(1)
	502	+	history.Append(row("old"))
	503	+	screen := renderer.NewFrame(3, 1)
	504	+	copy(screen.Row(0), row("one"))
	505	+
	506	+	document := NewSnapshot(history, screen)
	507	+	history.Append(row("new"))
	508	+	screen.Set(0, 0, renderer.Cell{Rune: 'x'})
	509	+
	510	+	require.Equal(t, 2, document.Len())
	511	+	require.Equal(t, "old", rowText(document.Row(0)))
	512	+	require.Equal(t, "one", rowText(document.Row(1)))
	513	+	require.Nil(t, document.Row(-1))
	514	+	require.Nil(t, document.Row(2))
	515	+	}
	516	+
	517	+	func TestNewSnapshotFromRowsOwnsCallerRows(t *testing.T) {
	518	+	rows := [][]renderer.Cell{row("one")}
	519	+	document := NewSnapshotFromRows(rows, 3, 1)
	520	+	rows[0][0].Rune = 'x'
	521	+	require.Equal(t, "one", rowText(document.Row(0)))
	522	+	}
	523	+
	524	+	// This scaling gate deliberately compares equal row counts rather than an
	525	+	// absolute budget, which varies between Go releases. A HistoryView shares
	526	+	// sealed chunks, so its allocation must not grow with history-row width.
	527	+	func TestNewSnapshotAllocationIsWidthIndependent(t *testing.T) {
	528	+	if testing.Short() {
	529	+	t.Skip("skipping benchmark-driven allocation gate in short mode")
	530	+	}
	531	+	narrow := benchmarkNewSnapshotBytes(16)
	532	+	wide := benchmarkNewSnapshotBytes(512)
	533	+	const conservativeWidthTolerance = 2
	534	+	require.LessOrEqual(t, wide, narrow*conservativeWidthTolerance,
	535	+	"NewSnapshot B/op scaled with row width: narrow=%d wide=%d", narrow, wide)
	536	+	}
	537	+
	538	+	func benchmarkNewSnapshotBytes(width int) int64 {
	539	+	const historyRows = 256
	540	+	history := newHistory(historyRows)
	541	+	for range historyRows {
	542	+	history.Append(make([]renderer.Cell, width))
	543	+	}
	544	+	// Keep visible-frame work constant so the comparison isolates history rows.
	545	+	screen := renderer.NewFrame(8, 4)
	546	+	result := testing.Benchmark(func(b *testing.B) {
	547	+	b.ReportAllocs()
	548	+	b.ResetTimer()
	549	+	for b.Loop() {
	550	+	benchmarkSnapshotSink = NewSnapshot(history, screen)
	551	+	}
	552	+	})
	553	+	return result.AllocedBytesPerOp()
	554	+	}
	555	+
	556	+	// This regression replaces the removed Scrollback ring tests: the VT history
	557	+	// remains bounded oldest-first and a captured copy document survives eviction.
	558	+	func TestNewSnapshotRetainsBoundedHistoryAcrossLaterEviction(t *testing.T) {
	559	+	history := newHistory(2)
	560	+	history.Append(row("one"))
	561	+	history.Append(row("two"))
	562	+	document := NewSnapshot(history, renderer.NewFrame(3, 1))
	563	+	history.Append(row("three"))
	564	+	history.Append(row("four"))
	565	+
	566	+	// A copy document is history followed by the visible frame. Its total is
	567	+	// therefore the bounded retained history plus the one visible row.
	568	+	require.Equal(t, 3, document.Len())
	569	+	require.Equal(t, "one", rowText(document.Row(0)))
	570	+	require.Equal(t, "two", rowText(document.Row(1)))
	571	+	require.Equal(t, " ", rowText(document.Row(2)), "the visible frame remains part of the document")
	572	+	view := history.View()
	573	+	require.Equal(t, "three", rowText(view.Row(0)))
	574	+	require.Equal(t, "four", rowText(view.Row(1)))
	575	+	}
	576	+
492	577		func BenchmarkNewSnapshot10KRows(b *testing.B) {
493	578		const (
494	579		width = 120


@@ -23,6 +23,14 @@ func row(text string) []renderer.Cell {
23		23
24	func newHistory(rows int) *vt.History { return vt.NewHistory(vt.HistoryConfig{MaxRows: rows}) }	24		func newHistory(rows int) *vt.History { return vt.NewHistory(vt.HistoryConfig{MaxRows: rows}) }
25		25
		26	+	func rowText(cells []renderer.Cell) string {
		27	+	runes := make([]rune, len(cells))
		28	+	for i, cell := range cells {
		29	+	runes[i] = cell.Rune
		30	+	}
		31	+	return string(runes)
		32	+	}
		33	+
26	func snapshot(lines []string, height int) Snapshot {	34		func snapshot(lines []string, height int) Snapshot {
27	rows := make([][]renderer.Cell, len(lines))	35		rows := make([][]renderer.Cell, len(lines))
28	for i, line := range lines {	36		for i, line := range lines {

@@ -489,6 +497,83 @@ func TestOSC52FromBase64(t *testing.T) {
489	}	497		}
490	}	498		}
491		499
		500	+	func TestNewSnapshotFreezesHistoryAndVisibleScreen(t *testing.T) {
		501	+	history := newHistory(1)
		502	+	history.Append(row("old"))
		503	+	screen := renderer.NewFrame(3, 1)
		504	+	copy(screen.Row(0), row("one"))
		505	+
		506	+	document := NewSnapshot(history, screen)
		507	+	history.Append(row("new"))
		508	+	screen.Set(0, 0, renderer.Cell{Rune: 'x'})
		509	+
		510	+	require.Equal(t, 2, document.Len())
		511	+	require.Equal(t, "old", rowText(document.Row(0)))
		512	+	require.Equal(t, "one", rowText(document.Row(1)))
		513	+	require.Nil(t, document.Row(-1))
		514	+	require.Nil(t, document.Row(2))
		515	+	}
		516	+
		517	+	func TestNewSnapshotFromRowsOwnsCallerRows(t *testing.T) {
		518	+	rows := [][]renderer.Cell{row("one")}
		519	+	document := NewSnapshotFromRows(rows, 3, 1)
		520	+	rows[0][0].Rune = 'x'
		521	+	require.Equal(t, "one", rowText(document.Row(0)))
		522	+	}
		523	+
		524	+	// This scaling gate deliberately compares equal row counts rather than an
		525	+	// absolute budget, which varies between Go releases. A HistoryView shares
		526	+	// sealed chunks, so its allocation must not grow with history-row width.
		527	+	func TestNewSnapshotAllocationIsWidthIndependent(t *testing.T) {
		528	+	if testing.Short() {
		529	+	t.Skip("skipping benchmark-driven allocation gate in short mode")
		530	+	}
		531	+	narrow := benchmarkNewSnapshotBytes(16)
		532	+	wide := benchmarkNewSnapshotBytes(512)
		533	+	const conservativeWidthTolerance = 2
		534	+	require.LessOrEqual(t, wide, narrow*conservativeWidthTolerance,
		535	+	"NewSnapshot B/op scaled with row width: narrow=%d wide=%d", narrow, wide)
		536	+	}
		537	+
		538	+	func benchmarkNewSnapshotBytes(width int) int64 {
		539	+	const historyRows = 256
		540	+	history := newHistory(historyRows)
		541	+	for range historyRows {
		542	+	history.Append(make([]renderer.Cell, width))
		543	+	}
		544	+	// Keep visible-frame work constant so the comparison isolates history rows.
		545	+	screen := renderer.NewFrame(8, 4)
		546	+	result := testing.Benchmark(func(b *testing.B) {
		547	+	b.ReportAllocs()
		548	+	b.ResetTimer()
		549	+	for b.Loop() {
		550	+	benchmarkSnapshotSink = NewSnapshot(history, screen)
		551	+	}
		552	+	})
		553	+	return result.AllocedBytesPerOp()
		554	+	}
		555	+
		556	+	// This regression replaces the removed Scrollback ring tests: the VT history
		557	+	// remains bounded oldest-first and a captured copy document survives eviction.
		558	+	func TestNewSnapshotRetainsBoundedHistoryAcrossLaterEviction(t *testing.T) {
		559	+	history := newHistory(2)
		560	+	history.Append(row("one"))
		561	+	history.Append(row("two"))
		562	+	document := NewSnapshot(history, renderer.NewFrame(3, 1))
		563	+	history.Append(row("three"))
		564	+	history.Append(row("four"))
		565	+
		566	+	// A copy document is history followed by the visible frame. Its total is
		567	+	// therefore the bounded retained history plus the one visible row.
		568	+	require.Equal(t, 3, document.Len())
		569	+	require.Equal(t, "one", rowText(document.Row(0)))
		570	+	require.Equal(t, "two", rowText(document.Row(1)))
		571	+	require.Equal(t, " ", rowText(document.Row(2)), "the visible frame remains part of the document")
		572	+	view := history.View()
		573	+	require.Equal(t, "three", rowText(view.Row(0)))
		574	+	require.Equal(t, "four", rowText(view.Row(1)))
		575	+	}
		576	+
492	func BenchmarkNewSnapshot10KRows(b *testing.B) {	577		func BenchmarkNewSnapshot10KRows(b *testing.B) {
493	const (	578		const (
494	width = 120	579		width = 120

M internal/usecase/daemon/daemon.go +11 -0


@@ -16,6 +16,7 @@ import (
16	16		"github.com/bnema/vev/internal/persist"
17	17		"github.com/bnema/vev/internal/ports"
18	18		"github.com/bnema/vev/internal/usecase/keys"
	19	+	snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
19	20		)
20	21
21	22		// Scheduler debounce bounds. Idle updates use the minimum for low latency;

@@ -101,6 +102,12 @@ type Daemon struct {
101	102		persistEnabled bool
102	103		snaps ports.SnapshotStore
103	104		snapsEnabled bool
	105	+	snapshotMarshal func(snapcodec.Session) ([]byte, error)
	106	+	snapshotJobs chan snapshotCapture
	107	+	snapshotWorkerMu sync.Mutex
	108	+	snapshotWorkerCtx context.Context
	109	+	snapshotWorkerCancel context.CancelFunc
	110	+	snapshotWorkerDone chan struct{}
104	111		restoreDone chan struct{}
105	112		restoreOnce sync.Once
106	113		procCwd func(int) (string, error)

@@ -285,6 +292,8 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt
285	292		done: make(chan struct{}),
286	293		restoreDone: make(chan struct{}),
287	294		animWake: make(chan struct{}, 1),
	295	+	snapshotMarshal: snapcodec.Marshal,
	296	+	snapshotJobs: make(chan snapshotCapture, snapshotQueueCapacity),
288	297		resumeParkGrace: defaultResumeParkGrace,
289	298		barScripts: &barScriptState{
290	299		cfg: barConfigFromDomain(domain.Defaults().Bar),

@@ -355,6 +364,7 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
355	364		})
356	365		}
357	366		if d.snapsEnabled {
	367	+	d.startSnapshotEncodeWorker(d.serveCtx)
358	368		d.sessWg.Go(func() {
359	369		d.snapshotSaver(d.serveCtx)
360	370		})

@@ -403,6 +413,7 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
403	413		d.waitNotifies()
404	414		d.hardCancel()
405	415		d.serveCancel()
	416	+	d.stopSnapshotEncodeWorker()
406	417		d.connWg.Wait()
407	418		d.shutdownAll(ports.ReasonServerShutdown)
408	419		d.sessWg.Wait()


@@ -16,6 +16,7 @@ import (
16	"github.com/bnema/vev/internal/persist"	16		"github.com/bnema/vev/internal/persist"
17	"github.com/bnema/vev/internal/ports"	17		"github.com/bnema/vev/internal/ports"
18	"github.com/bnema/vev/internal/usecase/keys"	18		"github.com/bnema/vev/internal/usecase/keys"
		19	+	snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
19	)	20		)
20		21
21	// Scheduler debounce bounds. Idle updates use the minimum for low latency;	22		// Scheduler debounce bounds. Idle updates use the minimum for low latency;

@@ -101,6 +102,12 @@ type Daemon struct {
101	persistEnabled bool	102		persistEnabled bool
102	snaps ports.SnapshotStore	103		snaps ports.SnapshotStore
103	snapsEnabled bool	104		snapsEnabled bool
		105	+	snapshotMarshal func(snapcodec.Session) ([]byte, error)
		106	+	snapshotJobs chan snapshotCapture
		107	+	snapshotWorkerMu sync.Mutex
		108	+	snapshotWorkerCtx context.Context
		109	+	snapshotWorkerCancel context.CancelFunc
		110	+	snapshotWorkerDone chan struct{}
104	restoreDone chan struct{}	111		restoreDone chan struct{}
105	restoreOnce sync.Once	112		restoreOnce sync.Once
106	procCwd func(int) (string, error)	113		procCwd func(int) (string, error)

@@ -285,6 +292,8 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt
285	done: make(chan struct{}),	292		done: make(chan struct{}),
286	restoreDone: make(chan struct{}),	293		restoreDone: make(chan struct{}),
287	animWake: make(chan struct{}, 1),	294		animWake: make(chan struct{}, 1),
		295	+	snapshotMarshal: snapcodec.Marshal,
		296	+	snapshotJobs: make(chan snapshotCapture, snapshotQueueCapacity),
288	resumeParkGrace: defaultResumeParkGrace,	297		resumeParkGrace: defaultResumeParkGrace,
289	barScripts: &barScriptState{	298		barScripts: &barScriptState{
290	cfg: barConfigFromDomain(domain.Defaults().Bar),	299		cfg: barConfigFromDomain(domain.Defaults().Bar),

@@ -355,6 +364,7 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
355	})	364		})
356	}	365		}
357	if d.snapsEnabled {	366		if d.snapsEnabled {
		367	+	d.startSnapshotEncodeWorker(d.serveCtx)
358	d.sessWg.Go(func() {	368		d.sessWg.Go(func() {
359	d.snapshotSaver(d.serveCtx)	369		d.snapshotSaver(d.serveCtx)
360	})	370		})

@@ -403,6 +413,7 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
403	d.waitNotifies()	413		d.waitNotifies()
404	d.hardCancel()	414		d.hardCancel()
405	d.serveCancel()	415		d.serveCancel()
		416	+	d.stopSnapshotEncodeWorker()
406	d.connWg.Wait()	417		d.connWg.Wait()
407	d.shutdownAll(ports.ReasonServerShutdown)	418		d.shutdownAll(ports.ReasonServerShutdown)
408	d.sessWg.Wait()	419		d.sessWg.Wait()

M internal/usecase/daemon/performance_benchmark_test.go +24 -1

@@ -1,8 +1,10 @@
1	1		package daemon
2	2
3	3		import (
	4	+	"context"
4	5		"fmt"
5	6		"io"
	7	+	"runtime"
6	8		"sync"
7	9		"testing"
8	10		"time"

@@ -501,13 +503,19 @@ func newPerformanceFixture(t testing.TB, config performanceConfig) *performanceF
501	503		sess: sess,
502	504		ac: ac,
503	505		output: output,
504		-	snaps: &countingSnapshotStore{},
	506	+	snaps: &countingSnapshotStore{done: make(chan struct{}, 1)},
505	507		pty: pty,
506	508		clock: clock,
507	509		liveWrites: [][]byte{[]byte("\x1b[1;1HA\x1b[2;2HA"), []byte("\x1b[1;1HB\x1b[2;2HB")},
508	510		resizeSizes: [2]domain.Size{{Cols: 100, Rows: 30}, config.size},
509	511		}
510	512		WithSnapshotStore(fixture.snaps)(d)
	513	+	snapshotCtx, cancelSnapshotWorker := context.WithCancel(context.Background())
	514	+	d.startSnapshotEncodeWorker(snapshotCtx)
	515	+	t.Cleanup(func() {
	516	+	cancelSnapshotWorker()
	517	+	d.stopSnapshotEncodeWorker()
	518	+	})
511	519		for tabIndex, tb := range sess.tabs {
512	520		fixture.configureTab(tb, tabIndex, config.panes, config.historyRows, tabSize(config.size))
513	521		}

@@ -594,6 +602,17 @@ func (f *performanceFixture) paintLive() {
594	602		func (f *performanceFixture) captureSnapshot() {
595	603		f.t.Helper()
596	604		require.True(f.t, f.d.captureSession(f.sess))
	605	+	<-f.snaps.done
	606	+	for range 4096 {
	607	+	f.sess.snapshotMu.Lock()
	608	+	pending := f.sess.snapshotPending
	609	+	f.sess.snapshotMu.Unlock()
	610	+	if !pending {
	611	+	return
	612	+	}
	613	+	runtime.Gosched()
	614	+	}
	615	+	f.t.Fatal("snapshot worker did not finish performance capture")
597	616		}
598	617
599	618		func (f *performanceFixture) enterCopySearch() {

@@ -754,6 +773,7 @@ type countingSnapshotStore struct {
754	773		mu sync.Mutex
755	774		countingSnapshotMetrics
756	775		last []byte
	776	+	done chan struct{}
757	777		}
758	778
759	779		func (s *countingSnapshotStore) Write(_ string, data []byte) error {

@@ -762,6 +782,9 @@ func (s *countingSnapshotStore) Write(_ string, data []byte) error {
762	782		s.bytes += uint64(len(data))
763	783		s.last = data
764	784		s.mu.Unlock()
	785	+	if s.done != nil {
	786	+	s.done <- struct{}{}
	787	+	}
765	788		return nil
766	789		}
767	790		func (*countingSnapshotStore) Load() ([]ports.SnapshotBlob, error) { return nil, nil }

 package daemon
 import (
+	"context"
 	"fmt"
 	"io"
+	"runtime"
 	"sync"
 	"testing"
 	"time"
 		sess:        sess,
 		ac:          ac,
 		output:      output,
-		snaps:       &countingSnapshotStore{},
+		snaps:       &countingSnapshotStore{done: make(chan struct{}, 1)},
 		pty:         pty,
 		clock:       clock,
 		liveWrites:  [][]byte{[]byte("\x1b[1;1HA\x1b[2;2HA"), []byte("\x1b[1;1HB\x1b[2;2HB")},
 		resizeSizes: [2]domain.Size{{Cols: 100, Rows: 30}, config.size},
 	}
 	WithSnapshotStore(fixture.snaps)(d)
+	snapshotCtx, cancelSnapshotWorker := context.WithCancel(context.Background())
+	d.startSnapshotEncodeWorker(snapshotCtx)
+	t.Cleanup(func() {
+		cancelSnapshotWorker()
+		d.stopSnapshotEncodeWorker()
+	})
 	for tabIndex, tb := range sess.tabs {
 		fixture.configureTab(tb, tabIndex, config.panes, config.historyRows, tabSize(config.size))
 	}
 func (f *performanceFixture) captureSnapshot() {
 	f.t.Helper()
 	require.True(f.t, f.d.captureSession(f.sess))
+	<-f.snaps.done
+	for range 4096 {
+		f.sess.snapshotMu.Lock()
+		pending := f.sess.snapshotPending
+		f.sess.snapshotMu.Unlock()
+		if !pending {
+			return
+		}
+		runtime.Gosched()
+	}
+	f.t.Fatal("snapshot worker did not finish performance capture")
 }
 func (f *performanceFixture) enterCopySearch() {
 	mu sync.Mutex
 	countingSnapshotMetrics
 	last []byte
+	done chan struct{}
 }
 func (s *countingSnapshotStore) Write(_ string, data []byte) error {
 	s.bytes += uint64(len(data))
 	s.last = data
 	s.mu.Unlock()
+	if s.done != nil {
+		s.done <- struct{}{}
+	}
 	return nil
 }
 func (*countingSnapshotStore) Load() ([]ports.SnapshotBlob, error) { return nil, nil }

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


@@ -44,6 +44,12 @@ type session struct {
44	44		mruAt atomic.Uint64
45	45		snapDirty atomic.Bool
46	46		snapEligible atomic.Bool
	47	+	// snapshotMu serializes the dirty generation with worker completion. It is
	48	+	// intentionally independent from mu: persistence never holds session state
	49	+	// locks while encoding or writing.
	50	+	snapshotMu sync.Mutex
	51	+	snapshotGeneration uint64
	52	+	snapshotPending bool
47	53		// syncGen makes synchronized-output watchdog generations unique across all
48	54		// panes in this session.
49	55		syncGen atomic.Uint64

@@ -671,7 +677,8 @@ func (d Daemon) killSession(sess session, reason uint8, purge bool) error {
671	677		d.log.Warn("deleting session snapshot failed", "err", err, "session", name)
672	678		}
673	679		} else {
674		-	d.captureSession(sess)
	680	+	markSnapshotDirty(sess)
	681	+	d.scheduleSnapshot(sess)
675	682		}
676	683		}
677	684		d.mu.Lock()

 	mruAt                  atomic.Uint64
 	snapDirty              atomic.Bool
 	snapEligible           atomic.Bool
+	// snapshotMu serializes the dirty generation with worker completion. It is
+	// intentionally independent from mu: persistence never holds session state
+	// locks while encoding or writing.
+	snapshotMu         sync.Mutex
+	snapshotGeneration uint64
+	snapshotPending    bool
 	// syncGen makes synchronized-output watchdog generations unique across all
 	// panes in this session.
 	syncGen atomic.Uint64
 				d.log.Warn("deleting session snapshot failed", "err", err, "session", name)
 			}
 		} else {
-			d.captureSession(sess)
+			markSnapshotDirty(sess)
+			d.scheduleSnapshot(sess)
 		}
 	}
 	d.mu.Lock()

M internal/usecase/daemon/snapshot.go +222 -52


@@ -9,9 +9,42 @@ import (
9	9		"github.com/bnema/vev/internal/domain"
10	10		"github.com/bnema/vev/internal/usecase/layout"
11	11		snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
	12	+	"github.com/bnema/vev/pkg/renderer"
12	13		"github.com/bnema/vev/pkg/vt"
13	14		)
14	15
	16	+	// snapshotQueueCapacity bounds retained immutable captures. A full queue never
	17	+	// stalls session producers; the session remains dirty for the next interval.
	18	+	const snapshotQueueCapacity = 1
	19	+
	20	+	type snapshotCapture struct {
	21	+	session *session
	22	+	generation uint64
	23	+	name string
	24	+	createdAt uint64
	25	+	active uint16
	26	+	tabs []snapshotCaptureTab
	27	+	}
	28	+
	29	+	type snapshotCaptureTab struct {
	30	+	stableID string
	31	+	cols uint16
	32	+	rows uint16
	33	+	nextPaneID uint64
	34	+	focus layout.PaneID
	35	+	tree *layout.Tree
	36	+	panes []snapshotCapturePane
	37	+	}
	38	+
	39	+	type snapshotCapturePane struct {
	40	+	id layout.PaneID
	41	+	stableID string
	42	+	cwd string
	43	+	history vt.HistoryView
	44	+	visible renderer.Frame
	45	+	process *snapcodec.Process
	46	+	}
	47	+
15	48		const snapshotInterval = 30 * time.Second
16	49
17	50		func (d *Daemon) closeRestoreDone() {

@@ -202,12 +235,13 @@ func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session) err
202	235		}
203	236
204	237		func markSnapshotDirty(sess *session) {
205		-	if sess == nil {
	238	+	if sess == nil \|\| !sess.snapEligible.Load() {
206	239		return
207	240		}
208		-	if sess.snapEligible.Load() {
209		-	sess.snapDirty.Store(true)
210		-	}
	241	+	sess.snapshotMu.Lock()
	242	+	sess.snapshotGeneration++
	243	+	sess.snapDirty.Store(true)
	244	+	sess.snapshotMu.Unlock()
211	245		}
212	246
213	247		func restorePaneTerminal(p *pane, snap snapcodec.Pane) error {

@@ -254,35 +288,84 @@ func restorePTYSize(contentRect domain.Rect, tabSize domain.Size) domain.Size {
254	288		return domain.Size{Cols: cols, Rows: rows}
255	289		}
256	290
257		-	func (d Daemon) captureSession(sess session) bool {
258		-	if d == nil \|\| sess == nil \|\| !d.snapsEnabled \|\| d.snaps == nil {
	291	+	// scheduleSnapshot captures a session only once while an immutable capture is
	292	+	// queued or being written. It never waits for the bounded worker queue.
	293	+	func (d Daemon) scheduleSnapshot(sess session) bool {
	294	+	if d == nil \|\| sess == nil \|\| !d.snapsEnabled \|\| d.snaps == nil \|\| !sess.snapEligible.Load() {
259	295		return true
260	296		}
	297	+	sess.snapshotMu.Lock()
	298	+	if !sess.snapDirty.Load() \|\| sess.snapshotPending {
	299	+	sess.snapshotMu.Unlock()
	300	+	return true
	301	+	}
	302	+	generation := sess.snapshotGeneration
	303	+	sess.snapshotPending = true
	304	+	sess.snapshotMu.Unlock()
261	305
	306	+	capture, ok := d.captureSnapshotState(sess, generation)
	307	+	if !ok {
	308	+	d.finishSnapshotCapture(sess, generation, false)
	309	+	return false
	310	+	}
	311	+	if d.enqueueSnapshotCapture(capture) {
	312	+	return true
	313	+	}
	314	+	// Coalesce under saturation or shutdown: the latest state stays dirty and
	315	+	// will be captured on a later tick once an active worker has room.
	316	+	d.finishSnapshotCapture(sess, generation, false)
	317	+	return false
	318	+	}
	319	+
	320	+	// enqueueSnapshotCapture serializes worker shutdown with the non-blocking
	321	+	// queue send. The queue is deliberately never closed: producers can race
	322	+	// shutdown without risking a send-on-closed panic.
	323	+	func (d *Daemon) enqueueSnapshotCapture(capture snapshotCapture) bool {
	324	+	d.snapshotWorkerMu.Lock()
	325	+	defer d.snapshotWorkerMu.Unlock()
	326	+	if d.snapshotWorkerCancel == nil \|\| d.snapshotWorkerCtx == nil \|\| d.snapshotWorkerCtx.Err() != nil {
	327	+	return false
	328	+	}
	329	+	select {
	330	+	case d.snapshotJobs <- capture:
	331	+	return true
	332	+	default:
	333	+	return false
	334	+	}
	335	+	}
	336	+
	337	+	// captureSession rotates history tails and clones visible frames while holding
	338	+	// each pane lock. The returned capture contains only immutable state; encoding
	339	+	// and persistence are deliberately deferred to snapshotEncodeWorker.
	340	+	func (d Daemon) captureSnapshotState(sess session, generation uint64) (snapshotCapture, bool) {
262	341		sess.mu.Lock()
263		-	name := sess.name
	342	+	capture := snapshotCapture{
	343	+	session: sess,
	344	+	generation: generation,
	345	+	name: sess.name,
	346	+	createdAt: uint64(sess.createdAt),
	347	+	active: uint16(max(sess.active, 0)),
	348	+	}
264	349		ephemeral := sess.ephemeral
265		-	createdAt := sess.createdAt
266		-	active := sess.active
267	350		fallbackCwd := sess.cwd
268	351		tabs := append([]*tab(nil), sess.tabs...)
269	352		sess.mu.Unlock()
270		-	if ephemeral \|\| name == "" {
271		-	return true
	353	+	if ephemeral \|\| capture.name == "" {
	354	+	return snapshotCapture{}, false
272	355		}
273	356
274		-	snap := snapcodec.Session{Name: name, CreatedAt: uint64(createdAt), Active: uint16(max(active, 0))}
	357	+	capture.tabs = make([]snapshotCaptureTab, 0, len(tabs))
275	358		for _, tb := range tabs {
276	359		tb.mu.Lock()
277		-	tabSnap := snapcodec.Tab{
278		-	StableID: tb.stableID,
279		-	Cols: uint16(max(tb.size.Cols, 0)),
280		-	Rows: uint16(max(tb.size.Rows, 0)),
281		-	NextPaneID: uint64(max(tb.nextPaneID, 0)),
282		-	Tree: tb.tree.Clone(),
	360	+	tabCapture := snapshotCaptureTab{
	361	+	stableID: tb.stableID,
	362	+	cols: uint16(max(tb.size.Cols, 0)),
	363	+	rows: uint16(max(tb.size.Rows, 0)),
	364	+	nextPaneID: uint64(max(tb.nextPaneID, 0)),
	365	+	tree: tb.tree.Clone(),
283	366		}
284	367		if tb.tree != nil {
285		-	tabSnap.Focus = tb.tree.Focus
	368	+	tabCapture.focus = tb.tree.Focus
286	369		}
287	370		panes := make([]*pane, 0, len(tb.panes))
288	371		for _, p := range tb.panes {

@@ -290,55 +373,144 @@ func (d Daemon) captureSession(sess session) bool {
290	373		}
291	374		tb.mu.Unlock()
292	375
293		-	paneIDs := make(map[layout.PaneID]struct{}, len(panes))
294		-	paneSnaps := make([]snapcodec.Pane, 0, len(panes))
	376	+	tabCapture.panes = make([]snapshotCapturePane, 0, len(panes))
295	377		for _, p := range panes {
296	378		p.mu.Lock()
297		-	pid := 0
298	379		pty := p.pty
	380	+	pid := 0
299	381		if pty != nil {
300	382		pid = pty.Pid()
301	383		}
302		-	historyView := p.history.SealAndView()
303		-	visibleFrame := p.screen.PrimaryVisibleFrame()
304		-	paneID, paneStableID := p.id, p.stableID
305		-	p.mu.Unlock()
306		-	// Sealed VT chunks are immutable and visibleFrame is the one required
307		-	// screen clone, so both encoders run after pane.mu is released.
308		-	sealed, tail, historyErr := vt.MarshalSealedHistory(historyView)
309		-	visible, visibleErr := vt.MarshalVisible(visibleFrame)
310		-	if historyErr != nil \|\| visibleErr != nil {
311		-	d.log.Warn("capturing terminal snapshot failed", "session", name, "pane", paneID, "history_err", historyErr, "visible_err", visibleErr)
312		-	return false
	384	+	paneCapture := snapshotCapturePane{
	385	+	id: p.id,
	386	+	stableID: p.stableID,
	387	+	history: p.history.SealAndView(),
	388	+	visible: p.screen.PrimaryVisibleFrame(),
313	389		}
314		-	paneSnap := snapcodec.Pane{ID: paneID, StableID: paneStableID, SealedChunks: sealed, Tail: tail, Visible: visible}
315		-	paneSnap.Cwd = fallbackCwd
	390	+	p.mu.Unlock()
	391	+	paneCapture.cwd = fallbackCwd
316	392		if d.procCwd != nil && pid > 0 {
317	393		if cwd, err := d.procCwd(pid); err == nil && cwd != "" {
318		-	paneSnap.Cwd = cwd
	394	+	paneCapture.cwd = cwd
319	395		}
320	396		}
321		-	paneSnap.Process = d.capturePaneProcess(pty, pid)
322		-	paneIDs[paneSnap.ID] = struct{}{}
323		-	paneSnaps = append(paneSnaps, paneSnap)
	397	+	paneCapture.process = d.capturePaneProcess(pty, pid)
	398	+	tabCapture.panes = append(tabCapture.panes, paneCapture)
	399	+	}
	400	+	capture.tabs = append(capture.tabs, tabCapture)
	401	+	}
	402	+	return capture, true
	403	+	}
	404	+
	405	+	// captureSession remains the synchronous producer-facing trigger for callers
	406	+	// such as teardown and benchmarks; the actual encoding and Write stay async.
	407	+	func (d Daemon) captureSession(sess session) bool {
	408	+	markSnapshotDirty(sess)
	409	+	return d.scheduleSnapshot(sess)
	410	+	}
	411	+
	412	+	func (d *Daemon) encodeSnapshotCapture(capture snapshotCapture) ([]byte, error) {
	413	+	snap := snapcodec.Session{Name: capture.name, CreatedAt: capture.createdAt, Active: capture.active, Tabs: make([]snapcodec.Tab, 0, len(capture.tabs))}
	414	+	for _, tabCapture := range capture.tabs {
	415	+	tabSnap := snapcodec.Tab{
	416	+	StableID: tabCapture.stableID, Cols: tabCapture.cols, Rows: tabCapture.rows,
	417	+	NextPaneID: tabCapture.nextPaneID, Focus: tabCapture.focus, Tree: tabCapture.tree,
	418	+	Panes: make([]snapcodec.Pane, 0, len(tabCapture.panes)),
	419	+	}
	420	+	paneIDs := make(map[layout.PaneID]struct{}, len(tabCapture.panes))
	421	+	for _, paneCapture := range tabCapture.panes {
	422	+	sealed, tail, historyErr := vt.MarshalSealedHistory(paneCapture.history)
	423	+	visible, visibleErr := vt.MarshalVisible(paneCapture.visible)
	424	+	if historyErr != nil \|\| visibleErr != nil {
	425	+	return nil, fmt.Errorf("terminal snapshot pane %s: history=%w visible=%w", paneCapture.id, historyErr, visibleErr)
	426	+	}
	427	+	tabSnap.Panes = append(tabSnap.Panes, snapcodec.Pane{ID: paneCapture.id, StableID: paneCapture.stableID, Cwd: paneCapture.cwd, SealedChunks: sealed, Tail: tail, Visible: visible, Process: paneCapture.process})
	428	+	paneIDs[paneCapture.id] = struct{}{}
324	429		}
325	430		if tabSnap.Tree != nil {
326	431		pruneTreeToPanes(tabSnap.Tree.Root, paneIDs)
327	432		}
328		-	tabSnap.Panes = paneSnaps
329	433		snap.Tabs = append(snap.Tabs, tabSnap)
330	434		}
	435	+	return d.snapshotMarshal(snap)
	436	+	}
331	437
332		-	data, err := snapcodec.Marshal(snap)
333		-	if err != nil {
334		-	d.log.Warn("marshaling session snapshot failed", "err", err, "session", name)
335		-	return false
	438	+	func (d Daemon) finishSnapshotCapture(sess session, generation uint64, succeeded bool) {
	439	+	sess.snapshotMu.Lock()
	440	+	sess.snapshotPending = false
	441	+	if succeeded && sess.snapshotGeneration == generation {
	442	+	sess.snapDirty.Store(false)
	443	+	} else if !succeeded {
	444	+	sess.snapDirty.Store(true)
336	445		}
337		-	if err := d.snaps.Write(name, data); err != nil {
338		-	d.log.Warn("writing session snapshot failed", "err", err, "session", name)
339		-	return false
	446	+	sess.snapshotMu.Unlock()
	447	+	}
	448	+
	449	+	func (d *Daemon) startSnapshotEncodeWorker(ctx context.Context) {
	450	+	if d == nil \|\| ctx.Err() != nil {
	451	+	return
	452	+	}
	453	+	d.snapshotWorkerMu.Lock()
	454	+	if d.snapshotWorkerCancel != nil {
	455	+	d.snapshotWorkerMu.Unlock()
	456	+	return
	457	+	}
	458	+	workerCtx, cancel := context.WithCancel(ctx)
	459	+	d.snapshotWorkerCtx = workerCtx
	460	+	d.snapshotWorkerCancel = cancel
	461	+	d.snapshotWorkerDone = make(chan struct{})
	462	+	done := d.snapshotWorkerDone
	463	+	d.snapshotWorkerMu.Unlock()
	464	+	go func() {
	465	+	defer close(done)
	466	+	for {
	467	+	select {
	468	+	case <-workerCtx.Done():
	469	+	return
	470	+	case capture := <-d.snapshotJobs:
	471	+	// A queued capture must not begin persistence after shutdown.
	472	+	if workerCtx.Err() != nil {
	473	+	d.finishSnapshotCapture(capture.session, capture.generation, false)
	474	+	return
	475	+	}
	476	+	data, err := d.encodeSnapshotCapture(capture)
	477	+	if err == nil {
	478	+	err = d.snaps.Write(capture.name, data)
	479	+	}
	480	+	if err != nil {
	481	+	d.log.Warn("writing session snapshot failed", "err", err, "session", capture.name)
	482	+	}
	483	+	d.finishSnapshotCapture(capture.session, capture.generation, err == nil)
	484	+	}
	485	+	}
	486	+	}()
	487	+	}
	488	+
	489	+	func (d *Daemon) stopSnapshotEncodeWorker() {
	490	+	if d == nil {
	491	+	return
	492	+	}
	493	+	d.snapshotWorkerMu.Lock()
	494	+	defer d.snapshotWorkerMu.Unlock()
	495	+	cancel, done := d.snapshotWorkerCancel, d.snapshotWorkerDone
	496	+	if cancel == nil {
	497	+	return
	498	+	}
	499	+	// Keep workerMu held until the old worker has exited and its queue has been
	500	+	// drained, so a concurrent restart cannot consume a stale capture.
	501	+	d.snapshotWorkerCtx = nil
	502	+	d.snapshotWorkerCancel = nil
	503	+	d.snapshotWorkerDone = nil
	504	+	cancel()
	505	+	<-done
	506	+	for {
	507	+	select {
	508	+	case capture := <-d.snapshotJobs:
	509	+	d.finishSnapshotCapture(capture.session, capture.generation, false)
	510	+	default:
	511	+	return
	512	+	}
340	513		}
341		-	return true
342	514		}
343	515
344	516		func (d Daemon) capturePaneProcess(pty interface{ ForegroundPgid() (int, error) }, shellPid int) snapcodec.Process {

@@ -377,10 +549,8 @@ func (d *Daemon) snapshotSaver(ctx context.Context) {
377	549		sessions := d.sessionsSnapshotLocked()
378	550		d.mu.Unlock()
379	551		for _, sess := range sessions {
380		-	if sess.snapEligible.Load() && sess.snapDirty.Swap(false) {
381		-	if !d.captureSession(sess) {
382		-	sess.snapDirty.Store(true)
383		-	}
	552	+	if sess.snapEligible.Load() && sess.snapDirty.Load() {
	553	+	d.scheduleSnapshot(sess)
384	554		}
385	555		}
386	556		t.Reset(snapshotInterval)


@@ -9,9 +9,42 @@ import (
9		"github.com/bnema/vev/internal/domain"	9		"github.com/bnema/vev/internal/domain"
10		"github.com/bnema/vev/internal/usecase/layout"	10		"github.com/bnema/vev/internal/usecase/layout"
11		snapcodec "github.com/bnema/vev/internal/usecase/snapshot"	11		snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
			12	+	"github.com/bnema/vev/pkg/renderer"
12		"github.com/bnema/vev/pkg/vt"	13		"github.com/bnema/vev/pkg/vt"
13		)	14		)
14			15
			16	+	// snapshotQueueCapacity bounds retained immutable captures. A full queue never
			17	+	// stalls session producers; the session remains dirty for the next interval.
			18	+	const snapshotQueueCapacity = 1
			19	+
			20	+	type snapshotCapture struct {
			21	+	session *session
			22	+	generation uint64
			23	+	name string
			24	+	createdAt uint64
			25	+	active uint16
			26	+	tabs []snapshotCaptureTab
			27	+	}
			28	+
			29	+	type snapshotCaptureTab struct {
			30	+	stableID string
			31	+	cols uint16
			32	+	rows uint16
			33	+	nextPaneID uint64
			34	+	focus layout.PaneID
			35	+	tree *layout.Tree
			36	+	panes []snapshotCapturePane
			37	+	}
			38	+
			39	+	type snapshotCapturePane struct {
			40	+	id layout.PaneID
			41	+	stableID string
			42	+	cwd string
			43	+	history vt.HistoryView
			44	+	visible renderer.Frame
			45	+	process *snapcodec.Process
			46	+	}
			47	+
15		const snapshotInterval = 30 * time.Second	48		const snapshotInterval = 30 * time.Second
16			49
17		func (d *Daemon) closeRestoreDone() {	50		func (d *Daemon) closeRestoreDone() {

@@ -202,12 +235,13 @@ func (d *Daemon) restoreSession(ctx context.Context, snap snapcodec.Session) err
202		}	235		}
203			236
204		func markSnapshotDirty(sess *session) {	237		func markSnapshotDirty(sess *session) {
205	-	if sess == nil {	238	+	if sess == nil \|\| !sess.snapEligible.Load() {
206		return	239		return
207		}	240		}
208	-	if sess.snapEligible.Load() {	241	+	sess.snapshotMu.Lock()
209	-	sess.snapDirty.Store(true)	242	+	sess.snapshotGeneration++
210	-	}	243	+	sess.snapDirty.Store(true)
			244	+	sess.snapshotMu.Unlock()
211		}	245		}
212			246
213		func restorePaneTerminal(p *pane, snap snapcodec.Pane) error {	247		func restorePaneTerminal(p *pane, snap snapcodec.Pane) error {

@@ -254,35 +288,84 @@ func restorePTYSize(contentRect domain.Rect, tabSize domain.Size) domain.Size {
254		return domain.Size{Cols: cols, Rows: rows}	288		return domain.Size{Cols: cols, Rows: rows}
255		}	289		}
256			290
257	-	func (d Daemon) captureSession(sess session) bool {	291	+	// scheduleSnapshot captures a session only once while an immutable capture is
258	-	if d == nil \|\| sess == nil \|\| !d.snapsEnabled \|\| d.snaps == nil {	292	+	// queued or being written. It never waits for the bounded worker queue.
			293	+	func (d Daemon) scheduleSnapshot(sess session) bool {
			294	+	if d == nil \|\| sess == nil \|\| !d.snapsEnabled \|\| d.snaps == nil \|\| !sess.snapEligible.Load() {
259		return true	295		return true
260		}	296		}
			297	+	sess.snapshotMu.Lock()
			298	+	if !sess.snapDirty.Load() \|\| sess.snapshotPending {
			299	+	sess.snapshotMu.Unlock()
			300	+	return true
			301	+	}
			302	+	generation := sess.snapshotGeneration
			303	+	sess.snapshotPending = true
			304	+	sess.snapshotMu.Unlock()
261			305
			306	+	capture, ok := d.captureSnapshotState(sess, generation)
			307	+	if !ok {
			308	+	d.finishSnapshotCapture(sess, generation, false)
			309	+	return false
			310	+	}
			311	+	if d.enqueueSnapshotCapture(capture) {
			312	+	return true
			313	+	}
			314	+	// Coalesce under saturation or shutdown: the latest state stays dirty and
			315	+	// will be captured on a later tick once an active worker has room.
			316	+	d.finishSnapshotCapture(sess, generation, false)
			317	+	return false
			318	+	}
			319	+
			320	+	// enqueueSnapshotCapture serializes worker shutdown with the non-blocking
			321	+	// queue send. The queue is deliberately never closed: producers can race
			322	+	// shutdown without risking a send-on-closed panic.
			323	+	func (d *Daemon) enqueueSnapshotCapture(capture snapshotCapture) bool {
			324	+	d.snapshotWorkerMu.Lock()
			325	+	defer d.snapshotWorkerMu.Unlock()
			326	+	if d.snapshotWorkerCancel == nil \|\| d.snapshotWorkerCtx == nil \|\| d.snapshotWorkerCtx.Err() != nil {
			327	+	return false
			328	+	}
			329	+	select {
			330	+	case d.snapshotJobs <- capture:
			331	+	return true
			332	+	default:
			333	+	return false
			334	+	}
			335	+	}
			336	+
			337	+	// captureSession rotates history tails and clones visible frames while holding
			338	+	// each pane lock. The returned capture contains only immutable state; encoding
			339	+	// and persistence are deliberately deferred to snapshotEncodeWorker.
			340	+	func (d Daemon) captureSnapshotState(sess session, generation uint64) (snapshotCapture, bool) {
262		sess.mu.Lock()	341		sess.mu.Lock()
263	-	name := sess.name	342	+	capture := snapshotCapture{
			343	+	session: sess,
			344	+	generation: generation,
			345	+	name: sess.name,
			346	+	createdAt: uint64(sess.createdAt),
			347	+	active: uint16(max(sess.active, 0)),
			348	+	}
264		ephemeral := sess.ephemeral	349		ephemeral := sess.ephemeral
265	-	createdAt := sess.createdAt
266	-	active := sess.active
267		fallbackCwd := sess.cwd	350		fallbackCwd := sess.cwd
268		tabs := append([]*tab(nil), sess.tabs...)	351		tabs := append([]*tab(nil), sess.tabs...)
269		sess.mu.Unlock()	352		sess.mu.Unlock()
270	-	if ephemeral \|\| name == "" {	353	+	if ephemeral \|\| capture.name == "" {
271	-	return true	354	+	return snapshotCapture{}, false
272		}	355		}
273			356
274	-	snap := snapcodec.Session{Name: name, CreatedAt: uint64(createdAt), Active: uint16(max(active, 0))}	357	+	capture.tabs = make([]snapshotCaptureTab, 0, len(tabs))
275		for _, tb := range tabs {	358		for _, tb := range tabs {
276		tb.mu.Lock()	359		tb.mu.Lock()
277	-	tabSnap := snapcodec.Tab{	360	+	tabCapture := snapshotCaptureTab{
278	-	StableID: tb.stableID,	361	+	stableID: tb.stableID,
279	-	Cols: uint16(max(tb.size.Cols, 0)),	362	+	cols: uint16(max(tb.size.Cols, 0)),
280	-	Rows: uint16(max(tb.size.Rows, 0)),	363	+	rows: uint16(max(tb.size.Rows, 0)),
281	-	NextPaneID: uint64(max(tb.nextPaneID, 0)),	364	+	nextPaneID: uint64(max(tb.nextPaneID, 0)),
282	-	Tree: tb.tree.Clone(),	365	+	tree: tb.tree.Clone(),
283		}	366		}
284		if tb.tree != nil {	367		if tb.tree != nil {
285	-	tabSnap.Focus = tb.tree.Focus	368	+	tabCapture.focus = tb.tree.Focus
286		}	369		}
287		panes := make([]*pane, 0, len(tb.panes))	370		panes := make([]*pane, 0, len(tb.panes))
288		for _, p := range tb.panes {	371		for _, p := range tb.panes {

@@ -290,55 +373,144 @@ func (d Daemon) captureSession(sess session) bool {
290		}	373		}
291		tb.mu.Unlock()	374		tb.mu.Unlock()
292			375
293	-	paneIDs := make(map[layout.PaneID]struct{}, len(panes))	376	+	tabCapture.panes = make([]snapshotCapturePane, 0, len(panes))
294	-	paneSnaps := make([]snapcodec.Pane, 0, len(panes))
295		for _, p := range panes {	377		for _, p := range panes {
296		p.mu.Lock()	378		p.mu.Lock()
297	-	pid := 0
298		pty := p.pty	379		pty := p.pty
			380	+	pid := 0
299		if pty != nil {	381		if pty != nil {
300		pid = pty.Pid()	382		pid = pty.Pid()
301		}	383		}
302	-	historyView := p.history.SealAndView()	384	+	paneCapture := snapshotCapturePane{
303	-	visibleFrame := p.screen.PrimaryVisibleFrame()	385	+	id: p.id,
304	-	paneID, paneStableID := p.id, p.stableID	386	+	stableID: p.stableID,
305	-	p.mu.Unlock()	387	+	history: p.history.SealAndView(),
306	-	// Sealed VT chunks are immutable and visibleFrame is the one required	388	+	visible: p.screen.PrimaryVisibleFrame(),
307	-	// screen clone, so both encoders run after pane.mu is released.
308	-	sealed, tail, historyErr := vt.MarshalSealedHistory(historyView)
309	-	visible, visibleErr := vt.MarshalVisible(visibleFrame)
310	-	if historyErr != nil \|\| visibleErr != nil {
311	-	d.log.Warn("capturing terminal snapshot failed", "session", name, "pane", paneID, "history_err", historyErr, "visible_err", visibleErr)
312	-	return false
313		}	389		}
314	-	paneSnap := snapcodec.Pane{ID: paneID, StableID: paneStableID, SealedChunks: sealed, Tail: tail, Visible: visible}	390	+	p.mu.Unlock()
315	-	paneSnap.Cwd = fallbackCwd	391	+	paneCapture.cwd = fallbackCwd
316		if d.procCwd != nil && pid > 0 {	392		if d.procCwd != nil && pid > 0 {
317		if cwd, err := d.procCwd(pid); err == nil && cwd != "" {	393		if cwd, err := d.procCwd(pid); err == nil && cwd != "" {
318	-	paneSnap.Cwd = cwd	394	+	paneCapture.cwd = cwd
319		}	395		}
320		}	396		}
321	-	paneSnap.Process = d.capturePaneProcess(pty, pid)	397	+	paneCapture.process = d.capturePaneProcess(pty, pid)
322	-	paneIDs[paneSnap.ID] = struct{}{}	398	+	tabCapture.panes = append(tabCapture.panes, paneCapture)
323	-	paneSnaps = append(paneSnaps, paneSnap)	399	+	}
			400	+	capture.tabs = append(capture.tabs, tabCapture)
			401	+	}
			402	+	return capture, true
			403	+	}
			404	+
			405	+	// captureSession remains the synchronous producer-facing trigger for callers
			406	+	// such as teardown and benchmarks; the actual encoding and Write stay async.
			407	+	func (d Daemon) captureSession(sess session) bool {
			408	+	markSnapshotDirty(sess)
			409	+	return d.scheduleSnapshot(sess)
			410	+	}
			411	+
			412	+	func (d *Daemon) encodeSnapshotCapture(capture snapshotCapture) ([]byte, error) {
			413	+	snap := snapcodec.Session{Name: capture.name, CreatedAt: capture.createdAt, Active: capture.active, Tabs: make([]snapcodec.Tab, 0, len(capture.tabs))}
			414	+	for _, tabCapture := range capture.tabs {
			415	+	tabSnap := snapcodec.Tab{
			416	+	StableID: tabCapture.stableID, Cols: tabCapture.cols, Rows: tabCapture.rows,
			417	+	NextPaneID: tabCapture.nextPaneID, Focus: tabCapture.focus, Tree: tabCapture.tree,
			418	+	Panes: make([]snapcodec.Pane, 0, len(tabCapture.panes)),
			419	+	}
			420	+	paneIDs := make(map[layout.PaneID]struct{}, len(tabCapture.panes))
			421	+	for _, paneCapture := range tabCapture.panes {
			422	+	sealed, tail, historyErr := vt.MarshalSealedHistory(paneCapture.history)
			423	+	visible, visibleErr := vt.MarshalVisible(paneCapture.visible)
			424	+	if historyErr != nil \|\| visibleErr != nil {
			425	+	return nil, fmt.Errorf("terminal snapshot pane %s: history=%w visible=%w", paneCapture.id, historyErr, visibleErr)
			426	+	}
			427	+	tabSnap.Panes = append(tabSnap.Panes, snapcodec.Pane{ID: paneCapture.id, StableID: paneCapture.stableID, Cwd: paneCapture.cwd, SealedChunks: sealed, Tail: tail, Visible: visible, Process: paneCapture.process})
			428	+	paneIDs[paneCapture.id] = struct{}{}
324		}	429		}
325		if tabSnap.Tree != nil {	430		if tabSnap.Tree != nil {
326		pruneTreeToPanes(tabSnap.Tree.Root, paneIDs)	431		pruneTreeToPanes(tabSnap.Tree.Root, paneIDs)
327		}	432		}
328	-	tabSnap.Panes = paneSnaps
329		snap.Tabs = append(snap.Tabs, tabSnap)	433		snap.Tabs = append(snap.Tabs, tabSnap)
330		}	434		}
			435	+	return d.snapshotMarshal(snap)
			436	+	}
331			437
332	-	data, err := snapcodec.Marshal(snap)	438	+	func (d Daemon) finishSnapshotCapture(sess session, generation uint64, succeeded bool) {
333	-	if err != nil {	439	+	sess.snapshotMu.Lock()
334	-	d.log.Warn("marshaling session snapshot failed", "err", err, "session", name)	440	+	sess.snapshotPending = false
335	-	return false	441	+	if succeeded && sess.snapshotGeneration == generation {
			442	+	sess.snapDirty.Store(false)
			443	+	} else if !succeeded {
			444	+	sess.snapDirty.Store(true)
336		}	445		}
337	-	if err := d.snaps.Write(name, data); err != nil {	446	+	sess.snapshotMu.Unlock()
338	-	d.log.Warn("writing session snapshot failed", "err", err, "session", name)	447	+	}
339	-	return false	448	+
			449	+	func (d *Daemon) startSnapshotEncodeWorker(ctx context.Context) {
			450	+	if d == nil \|\| ctx.Err() != nil {
			451	+	return
			452	+	}
			453	+	d.snapshotWorkerMu.Lock()
			454	+	if d.snapshotWorkerCancel != nil {
			455	+	d.snapshotWorkerMu.Unlock()
			456	+	return
			457	+	}
			458	+	workerCtx, cancel := context.WithCancel(ctx)
			459	+	d.snapshotWorkerCtx = workerCtx
			460	+	d.snapshotWorkerCancel = cancel
			461	+	d.snapshotWorkerDone = make(chan struct{})
			462	+	done := d.snapshotWorkerDone
			463	+	d.snapshotWorkerMu.Unlock()
			464	+	go func() {
			465	+	defer close(done)
			466	+	for {
			467	+	select {
			468	+	case <-workerCtx.Done():
			469	+	return
			470	+	case capture := <-d.snapshotJobs:
			471	+	// A queued capture must not begin persistence after shutdown.
			472	+	if workerCtx.Err() != nil {
			473	+	d.finishSnapshotCapture(capture.session, capture.generation, false)
			474	+	return
			475	+	}
			476	+	data, err := d.encodeSnapshotCapture(capture)
			477	+	if err == nil {
			478	+	err = d.snaps.Write(capture.name, data)
			479	+	}
			480	+	if err != nil {
			481	+	d.log.Warn("writing session snapshot failed", "err", err, "session", capture.name)
			482	+	}
			483	+	d.finishSnapshotCapture(capture.session, capture.generation, err == nil)
			484	+	}
			485	+	}
			486	+	}()
			487	+	}
			488	+
			489	+	func (d *Daemon) stopSnapshotEncodeWorker() {
			490	+	if d == nil {
			491	+	return
			492	+	}
			493	+	d.snapshotWorkerMu.Lock()
			494	+	defer d.snapshotWorkerMu.Unlock()
			495	+	cancel, done := d.snapshotWorkerCancel, d.snapshotWorkerDone
			496	+	if cancel == nil {
			497	+	return
			498	+	}
			499	+	// Keep workerMu held until the old worker has exited and its queue has been
			500	+	// drained, so a concurrent restart cannot consume a stale capture.
			501	+	d.snapshotWorkerCtx = nil
			502	+	d.snapshotWorkerCancel = nil
			503	+	d.snapshotWorkerDone = nil
			504	+	cancel()
			505	+	<-done
			506	+	for {
			507	+	select {
			508	+	case capture := <-d.snapshotJobs:
			509	+	d.finishSnapshotCapture(capture.session, capture.generation, false)
			510	+	default:
			511	+	return
			512	+	}
340		}	513		}
341	-	return true
342		}	514		}
343			515
344		func (d Daemon) capturePaneProcess(pty interface{ ForegroundPgid() (int, error) }, shellPid int) snapcodec.Process {	516		func (d Daemon) capturePaneProcess(pty interface{ ForegroundPgid() (int, error) }, shellPid int) snapcodec.Process {

@@ -377,10 +549,8 @@ func (d *Daemon) snapshotSaver(ctx context.Context) {
377		sessions := d.sessionsSnapshotLocked()	549		sessions := d.sessionsSnapshotLocked()
378		d.mu.Unlock()	550		d.mu.Unlock()
379		for _, sess := range sessions {	551		for _, sess := range sessions {
380	-	if sess.snapEligible.Load() && sess.snapDirty.Swap(false) {	552	+	if sess.snapEligible.Load() && sess.snapDirty.Load() {
381	-	if !d.captureSession(sess) {	553	+	d.scheduleSnapshot(sess)
382	-	sess.snapDirty.Store(true)
383	-	}
384		}	554		}
385		}	555		}
386		t.Reset(snapshotInterval)	556		t.Reset(snapshotInterval)

M internal/usecase/daemon/snapshot_test.go +325 -6


@@ -4,6 +4,7 @@ import (
4	4		"context"
5	5		"errors"
6	6		"io"
	7	+	"runtime"
7	8		"sync"
8	9		"sync/atomic"
9	10		"testing"

@@ -20,12 +21,245 @@ import (
20	21		"github.com/bnema/vev/pkg/renderer"
21	22		)
22	23
	24	+	func TestSnapshotEncodeWorkerDefersMarshalUntilAfterCapture(t *testing.T) {
	25	+	store := &channelSnapshotStore{writes: make(chan []byte, 1)}
	26	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	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)
	34	+
	35	+	entered := make(chan struct{})
	36	+	release := make(chan struct{})
	37	+	d.snapshotMarshal = func(s snapcodec.Session) ([]byte, error) {
	38	+	close(entered)
	39	+	<-release
	40	+	return snapcodec.Marshal(s)
	41	+	}
	42	+	sess := newSnapshotTestSession(t, "work", false, "/work")
	43	+	markSnapshotDirty(sess)
	44	+	require.True(t, d.scheduleSnapshot(sess))
	45	+	<-entered
	46	+
	47	+	paneUnlocked := make(chan struct{})
	48	+	go func() {
	49	+	p := sess.tabs[0].panes["pane-1"]
	50	+	p.mu.Lock()
	51	+	p.mu.Unlock()
	52	+	close(paneUnlocked)
	53	+	}()
	54	+	<-paneUnlocked
	55	+	close(release)
	56	+	<-store.writes
	57	+	}
	58	+
	59	+	func TestSnapshotEncodeWorkerDoesNotBlockPTYOrCoordinatorRender(t *testing.T) {
	60	+	exerciseBlockedSnapshotWorker(t, true)
	61	+	}
	62	+
	63	+	func TestSnapshotStoreWorkerDoesNotBlockPTYOrCoordinatorRender(t *testing.T) {
	64	+	exerciseBlockedSnapshotWorker(t, false)
	65	+	}
	66	+
	67	+	func exerciseBlockedSnapshotWorker(t *testing.T, blockEncode bool) {
	68	+	t.Helper()
	69	+	store := newGatedSnapshotStore()
	70	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	71	+	WithSnapshotStore(store)(d)
	72	+	ctx, cancel := context.WithCancel(t.Context())
	73	+	d.startSnapshotEncodeWorker(ctx)
	74	+	t.Cleanup(func() {
	75	+	store.unblock()
	76	+	cancel()
	77	+	d.stopSnapshotEncodeWorker()
	78	+	})
	79	+	if blockEncode {
	80	+	d.snapshotMarshal = func(s snapcodec.Session) ([]byte, error) {
	81	+	store.enter()
	82	+	<-store.release
	83	+	return snapcodec.Marshal(s)
	84	+	}
	85	+	}
	86	+
	87	+	sess := newSnapshotTestSession(t, "live", false, "/live")
	88	+	d.sessions[sess.id] = sess
	89	+	sess.tabs[0].size = domain.Size{Cols: 80, Rows: 24}
	90	+	p := sess.tabs[0].panes["pane-1"]
	91	+	tr, sends, _ := newConn(t, ports.Frame{})
	92	+	ac := &attachedClient{tr: tr, output: newOutputStateStream(), size: domain.Size{Cols: 8, Rows: 3}}
	93	+	ac.setSession(sess)
	94	+	sess.mu.Lock()
	95	+	sess.client = ac
	96	+	sess.mu.Unlock()
	97	+	rendered := make(chan struct{}, 1)
	98	+	rc := newRenderCoordinator(renderCoordinatorOptions{
	99	+	clock: stubClock{},
	100	+	onInvalidate: func(renderInvalidation) {
	101	+	d.paint(sess, ac, true)
	102	+	rendered <- struct{}{}
	103	+	},
	104	+	})
	105	+	sess.installRenderCoordinator(rc)
	106	+	rc.attach(ac)
	107	+	require.True(t, rc.markAttachmentReady(ac))
	108	+	t.Cleanup(rc.noteSessionTeardown)
	109	+
	110	+	markSnapshotDirty(sess)
	111	+	require.True(t, d.scheduleSnapshot(sess))
	112	+	<-store.entered
	113	+
	114	+	paneUnlocked := make(chan struct{})
	115	+	go func() {
	116	+	p.mu.Lock()
	117	+	p.mu.Unlock()
	118	+	close(paneUnlocked)
	119	+	}()
	120	+	<-paneUnlocked
	121	+
	122	+	reader := portsmocks.NewMockPTY(t)
	123	+	var reads atomic.Int32
	124	+	reader.EXPECT().Read(mock.Anything).RunAndReturn(func(buf []byte) (int, error) {
	125	+	if reads.Add(1) == 1 {
	126	+	return copy(buf, []byte("ok")), nil
	127	+	}
	128	+	return 0, io.EOF
	129	+	}).Twice()
	130	+	reader.EXPECT().Write(mock.Anything).Return(0, nil).Maybe()
	131	+	p.mu.Lock()
	132	+	p.pty = reader
	133	+	p.onExit = func() {}
	134	+	p.mu.Unlock()
	135	+	d.sessWg.Add(1)
	136	+	go d.ptyReader(sess, sess.tabs[0], p)
	137	+	d.sessWg.Wait()
	138	+	<-rendered
	139	+	_ = awaitFrame(t, sends, ports.MsgOutput)
	140	+	p.mu.Lock()
	141	+	require.Contains(t, screenLineText(p.screen, 0), "ok")
	142	+	p.mu.Unlock()
	143	+
	144	+	store.unblock()
	145	+	awaitSnapshotIdle(t, sess)
	146	+	require.True(t, sess.snapDirty.Load(), "PTY output published while blocked must remain retryable")
	147	+	}
	148	+
	149	+	func TestStoppedSnapshotWorkerLeavesCaptureRetryable(t *testing.T) {
	150	+	store := &channelSnapshotStore{writes: make(chan []byte, 1)}
	151	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	152	+	WithSnapshotStore(store)(d)
	153	+	ctx, cancel := context.WithCancel(t.Context())
	154	+	d.startSnapshotEncodeWorker(ctx)
	155	+	d.stopSnapshotEncodeWorker()
	156	+	t.Cleanup(cancel)
	157	+
	158	+	sess := newSnapshotTestSession(t, "stopped", false, "/work")
	159	+	markSnapshotDirty(sess)
	160	+	require.False(t, d.scheduleSnapshot(sess), "a stopped worker must not accept a capture")
	161	+	awaitSnapshotIdle(t, sess)
	162	+	require.True(t, sess.snapDirty.Load(), "an unsubmitted capture must remain retryable")
	163	+	}
	164	+
	165	+	func TestSnapshotQueueSaturationLeavesCaptureRetryable(t *testing.T) {
	166	+	store := &channelSnapshotStore{writes: make(chan []byte, 3)}
	167	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	168	+	WithSnapshotStore(store)(d)
	169	+	startSnapshotEncodeWorker(t, d)
	170	+	entered := make(chan struct{})
	171	+	release := make(chan struct{})
	172	+	var calls atomic.Int32
	173	+	d.snapshotMarshal = func(s snapcodec.Session) ([]byte, error) {
	174	+	if calls.Add(1) == 1 {
	175	+	close(entered)
	176	+	<-release
	177	+	}
	178	+	return snapcodec.Marshal(s)
	179	+	}
	180	+	first := newSnapshotTestSession(t, "first", false, "/work")
	181	+	second := newSnapshotTestSession(t, "second", false, "/work")
	182	+	third := newSnapshotTestSession(t, "third", false, "/work")
	183	+	markSnapshotDirty(first)
	184	+	markSnapshotDirty(second)
	185	+	markSnapshotDirty(third)
	186	+	require.True(t, d.scheduleSnapshot(first))
	187	+	<-entered
	188	+	require.True(t, d.scheduleSnapshot(second), "one capture may wait in the bounded queue")
	189	+	require.False(t, d.scheduleSnapshot(third), "full queue must not block the producer")
	190	+	require.True(t, third.snapDirty.Load())
	191	+
	192	+	close(release)
	193	+	awaitSnapshotClean(t, first)
	194	+	awaitSnapshotClean(t, second)
	195	+	require.True(t, d.scheduleSnapshot(third))
	196	+	<-store.writes
	197	+	awaitSnapshotClean(t, third)
	198	+	}
	199	+
	200	+	func TestSnapshotSuccessDoesNotClearNewerGeneration(t *testing.T) {
	201	+	store := &channelSnapshotStore{writes: make(chan []byte, 2)}
	202	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	203	+	WithSnapshotStore(store)(d)
	204	+	startSnapshotEncodeWorker(t, d)
	205	+	entered := make(chan struct{})
	206	+	release := make(chan struct{})
	207	+	var calls atomic.Int32
	208	+	d.snapshotMarshal = func(s snapcodec.Session) ([]byte, error) {
	209	+	if calls.Add(1) == 1 {
	210	+	close(entered)
	211	+	<-release
	212	+	}
	213	+	return snapcodec.Marshal(s)
	214	+	}
	215	+	sess := newSnapshotTestSession(t, "generation", false, "/work")
	216	+
	217	+	markSnapshotDirty(sess)
	218	+	require.True(t, d.scheduleSnapshot(sess))
	219	+	<-entered
	220	+	markSnapshotDirty(sess)
	221	+	close(release)
	222	+	<-store.writes
	223	+	awaitSnapshotIdle(t, sess)
	224	+	require.True(t, sess.snapDirty.Load(), "a stale successful capture must not clear newer state")
	225	+
	226	+	require.True(t, d.scheduleSnapshot(sess))
	227	+	<-store.writes
	228	+	awaitSnapshotClean(t, sess)
	229	+	}
	230	+
	231	+	func TestSnapshotEncodeFailureLeavesSessionDirtyAndRetries(t *testing.T) {
	232	+	store := &channelSnapshotStore{writes: make(chan []byte, 1)}
	233	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	234	+	WithSnapshotStore(store)(d)
	235	+	startSnapshotEncodeWorker(t, d)
	236	+	sess := newSnapshotTestSession(t, "retry-encode", false, "/work")
	237	+	failed := make(chan struct{}, 1)
	238	+	var calls atomic.Int32
	239	+	d.snapshotMarshal = func(s snapcodec.Session) ([]byte, error) {
	240	+	if calls.Add(1) == 1 {
	241	+	failed <- struct{}{}
	242	+	return nil, errors.New("encode failed")
	243	+	}
	244	+	return snapcodec.Marshal(s)
	245	+	}
	246	+
	247	+	require.True(t, d.captureSession(sess))
	248	+	<-failed
	249	+	awaitSnapshotIdle(t, sess)
	250	+	require.True(t, sess.snapDirty.Load())
	251	+	require.True(t, d.scheduleSnapshot(sess))
	252	+	<-store.writes
	253	+	awaitSnapshotClean(t, sess)
	254	+	}
	255	+
23	256		func TestSnapshotSaverWritesDirtyNamedSessionsOnly(t *testing.T) {
24	257		clock := newManualSnapshotClock()
25	258		store := portsmocks.NewMockSnapshotStore(t)
26	259		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
27	260		WithSnapshotStore(store)(d)
28	261		d.procCwd = func(int) (string, error) { return "/pane", nil }
	262	+	startSnapshotEncodeWorker(t, d)
29	263
30	264		dirty := newSnapshotTestSession(t, "dirty", false, "/fallback")
31	265		clean := newSnapshotTestSession(t, "clean", false, "/fallback")

@@ -67,6 +301,7 @@ func TestSnapshotCaptureExcludesFloatingRuntime(t *testing.T) {
67	301		store := portsmocks.NewMockSnapshotStore(t)
68	302		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
69	303		WithSnapshotStore(store)(d)
	304	+	startSnapshotEncodeWorker(t, d)
70	305		sess := newSnapshotTestSession(t, "work", false, "/fallback")
71	306		tb := sess.tabs[0]
72	307		floatingPTY, _ := newBlockingPTY(t)

@@ -85,12 +320,14 @@ func TestSnapshotCaptureExcludesFloatingRuntime(t *testing.T) {
85	320		return nil
86	321		}).Once()
87	322		require.True(t, d.captureSession(sess))
	323	+	awaitSnapshotClean(t, sess)
88	324		}
89	325
90	326		func TestSnapshotCaptureStoresForegroundProcessMetadata(t *testing.T) {
91	327		store := portsmocks.NewMockSnapshotStore(t)
92	328		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
93	329		WithSnapshotStore(store)(d)
	330	+	startSnapshotEncodeWorker(t, d)
94	331		d.procGroupArgv = func(pgid int, shellPid int) ([]string, error) {
95	332		require.Equal(t, 4321, pgid)
96	333		require.Equal(t, 1234, shellPid)

@@ -113,12 +350,14 @@ func TestSnapshotCaptureStoresForegroundProcessMetadata(t *testing.T) {
113	350		}).Once()
114	351
115	352		require.True(t, d.captureSession(sess))
	353	+	awaitSnapshotClean(t, sess)
116	354		}
117	355
118	356		func TestSnapshotCaptureProcessFailureDoesNotBlockSnapshot(t *testing.T) {
119	357		store := portsmocks.NewMockSnapshotStore(t)
120	358		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
121	359		WithSnapshotStore(store)(d)
	360	+	startSnapshotEncodeWorker(t, d)
122	361		d.procGroupArgv = func(int, int) ([]string, error) { return nil, errors.New("inspect failed") }
123	362
124	363		sess := newSnapshotTestSession(t, "work", false, "/fallback")

@@ -133,12 +372,14 @@ func TestSnapshotCaptureProcessFailureDoesNotBlockSnapshot(t *testing.T) {
133	372		}).Once()
134	373
135	374		require.True(t, d.captureSession(sess))
	375	+	awaitSnapshotClean(t, sess)
136	376		}
137	377
138	378		func TestSnapshotCaptureSkipsBareShellProcess(t *testing.T) {
139	379		store := portsmocks.NewMockSnapshotStore(t)
140	380		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
141	381		WithSnapshotStore(store)(d)
	382	+	startSnapshotEncodeWorker(t, d)
142	383		d.procGroupArgv = func(int, int) ([]string, error) {
143	384		t.Fatal("bare shell should not inspect argv")
144	385		return nil, nil

@@ -156,35 +397,37 @@ func TestSnapshotCaptureSkipsBareShellProcess(t *testing.T) {
156	397		}).Once()
157	398
158	399		require.True(t, d.captureSession(sess))
	400	+	awaitSnapshotClean(t, sess)
159	401		}
160	402
161	403		func TestKillSessionSnapshotsNamedSessionBeforeClosingPanes(t *testing.T) {
162	404		store := portsmocks.NewMockSnapshotStore(t)
163	405		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
164	406		WithSnapshotStore(store)(d)
	407	+	startSnapshotEncodeWorker(t, d)
165	408		d.procCwd = func(int) (string, error) { return "/live", nil }
166	409
167	410		sess := newSnapshotTestSession(t, "work", false, "/fallback")
168	411		d.sessions[sess.id] = sess
169	412
170		-	var wrote atomic.Bool
	413	+	var closed atomic.Bool
171	414		store.EXPECT().Write("work", mock.Anything).RunAndReturn(func(_ string, data []byte) error {
172	415		snap, err := snapcodec.Unmarshal(data)
173	416		require.NoError(t, err)
174	417		require.Equal(t, "work", snap.Name)
175	418		require.Equal(t, "/live", snap.Tabs[0].Panes[0].Cwd)
176		-	wrote.Store(true)
	419	+	require.True(t, closed.Load(), "snapshot persistence must not delay pane close")
177	420		return nil
178	421		}).Once()
179	422		mockPTY, ok := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
180	423		require.True(t, ok)
181	424		mockPTY.EXPECT().Close().RunAndReturn(func() error {
182		-	require.True(t, wrote.Load(), "snapshot Write must happen before pane Close")
	425	+	closed.Store(true)
183	426		return nil
184	427		}).Once()
185	428
186	429		require.NoError(t, d.killSession(sess, 0, false))
187		-	require.True(t, wrote.Load())
	430	+	awaitSnapshotClean(t, sess)
188	431		}
189	432
190	433		func TestPTYReaderSynchronizedUpdateFlushMarksSnapshotDirty(t *testing.T) {

@@ -241,12 +484,17 @@ func TestSnapshotSaverKeepsDirtyWhenWriteFails(t *testing.T) {
241	484		store := portsmocks.NewMockSnapshotStore(t)
242	485		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
243	486		WithSnapshotStore(store)(d)
	487	+	startSnapshotEncodeWorker(t, d)
244	488
245	489		sess := newSnapshotTestSession(t, "retry", false, "/fallback")
246	490		sess.snapDirty.Store(true)
247	491		d.sessions[sess.id] = sess
248	492
249		-	store.EXPECT().Write("retry", mock.Anything).Return(errors.New("boom")).Once()
	493	+	failed := make(chan struct{}, 1)
	494	+	store.EXPECT().Write("retry", mock.Anything).RunAndReturn(func(string, []byte) error {
	495	+	failed <- struct{}{}
	496	+	return errors.New("boom")
	497	+	}).Once()
250	498		wrote := make(chan struct{}, 1)
251	499		store.EXPECT().Write("retry", mock.Anything).RunAndReturn(func(string, []byte) error {
252	500		wrote <- struct{}{}

@@ -257,7 +505,9 @@ func TestSnapshotSaverKeepsDirtyWhenWriteFails(t *testing.T) {
257	505		defer cancel()
258	506		go d.snapshotSaver(ctx)
259	507		clock.fireNext(t)
260		-	require.Eventually(t, func() bool { return sess.snapDirty.Load() }, time.Second, time.Millisecond)
	508	+	<-failed
	509	+	awaitSnapshotIdle(t, sess)
	510	+	require.True(t, sess.snapDirty.Load())
261	511		clock.fireNext(t)
262	512		select {
263	513		case <-wrote:

@@ -365,6 +615,7 @@ func TestSnapshotSaverCapturesLayoutOnlyDirtySave(t *testing.T) {
365	615		factory := portsmocks.NewMockPTYFactory(t)
366	616		d := newTestDaemon(t, factory, clock)
367	617		WithSnapshotStore(store)(d)
	618	+	startSnapshotEncodeWorker(t, d)
368	619
369	620		sess := newSnapshotTestSession(t, "work", false, "/work")
370	621		tctx, tcancel := context.WithCancel(context.Background())

@@ -450,6 +701,74 @@ func newSnapshotTestSession(t *testing.T, name string, ephemeral bool, cwd strin
450	701		return sess
451	702		}
452	703
	704	+	func awaitSnapshotIdle(t testing.T, sess session) {
	705	+	t.Helper()
	706	+	for range 4096 {
	707	+	sess.snapshotMu.Lock()
	708	+	pending := sess.snapshotPending
	709	+	sess.snapshotMu.Unlock()
	710	+	if !pending {
	711	+	return
	712	+	}
	713	+	runtime.Gosched()
	714	+	}
	715	+	t.Fatal("snapshot worker did not become idle")
	716	+	}
	717	+
	718	+	func awaitSnapshotClean(t testing.T, sess session) {
	719	+	t.Helper()
	720	+	for range 4096 {
	721	+	if !sess.snapDirty.Load() {
	722	+	return
	723	+	}
	724	+	runtime.Gosched()
	725	+	}
	726	+	t.Fatal("snapshot worker did not finish")
	727	+	}
	728	+
	729	+	func startSnapshotEncodeWorker(t testing.T, d Daemon) {
	730	+	t.Helper()
	731	+	ctx, cancel := context.WithCancel(t.Context())
	732	+	d.startSnapshotEncodeWorker(ctx)
	733	+	t.Cleanup(func() {
	734	+	cancel()
	735	+	d.stopSnapshotEncodeWorker()
	736	+	})
	737	+	}
	738	+
	739	+	type gatedSnapshotStore struct {
	740	+	entered chan struct{}
	741	+	release chan struct{}
	742	+	once sync.Once
	743	+	releaseOnce sync.Once
	744	+	}
	745	+
	746	+	func newGatedSnapshotStore() *gatedSnapshotStore {
	747	+	return &gatedSnapshotStore{entered: make(chan struct{}), release: make(chan struct{})}
	748	+	}
	749	+
	750	+	func (s *gatedSnapshotStore) enter() { s.once.Do(func() { close(s.entered) }) }
	751	+	func (s *gatedSnapshotStore) unblock() { s.releaseOnce.Do(func() { close(s.release) }) }
	752	+
	753	+	func (s *gatedSnapshotStore) Write(_ string, _ []byte) error {
	754	+	s.enter()
	755	+	<-s.release
	756	+	return nil
	757	+	}
	758	+	func (*gatedSnapshotStore) Load() ([]ports.SnapshotBlob, error) { return nil, nil }
	759	+	func (*gatedSnapshotStore) Delete(string) error { return nil }
	760	+
	761	+	type channelSnapshotStore struct {
	762	+	writes chan []byte
	763	+	}
	764	+
	765	+	func (s *channelSnapshotStore) Write(_ string, data []byte) error {
	766	+	s.writes <- append([]byte(nil), data...)
	767	+	return nil
	768	+	}
	769	+	func (*channelSnapshotStore) Load() ([]ports.SnapshotBlob, error) { return nil, nil }
	770	+	func (*channelSnapshotStore) Delete(string) error { return nil }
	771	+
453	772		type manualSnapshotClock struct {
454	773		mu sync.Mutex
455	774		timers []*manualSnapshotTimer


@@ -4,6 +4,7 @@ import (
4		"context"	4		"context"
5		"errors"	5		"errors"
6		"io"	6		"io"
			7	+	"runtime"
7		"sync"	8		"sync"
8		"sync/atomic"	9		"sync/atomic"
9		"testing"	10		"testing"

@@ -20,12 +21,245 @@ import (
20		"github.com/bnema/vev/pkg/renderer"	21		"github.com/bnema/vev/pkg/renderer"
21		)	22		)
22			23
			24	+	func TestSnapshotEncodeWorkerDefersMarshalUntilAfterCapture(t *testing.T) {
			25	+	store := &channelSnapshotStore{writes: make(chan []byte, 1)}
			26	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
			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)
			34	+
			35	+	entered := make(chan struct{})
			36	+	release := make(chan struct{})
			37	+	d.snapshotMarshal = func(s snapcodec.Session) ([]byte, error) {
			38	+	close(entered)
			39	+	<-release
			40	+	return snapcodec.Marshal(s)
			41	+	}
			42	+	sess := newSnapshotTestSession(t, "work", false, "/work")
			43	+	markSnapshotDirty(sess)
			44	+	require.True(t, d.scheduleSnapshot(sess))
			45	+	<-entered
			46	+
			47	+	paneUnlocked := make(chan struct{})
			48	+	go func() {
			49	+	p := sess.tabs[0].panes["pane-1"]
			50	+	p.mu.Lock()
			51	+	p.mu.Unlock()
			52	+	close(paneUnlocked)
			53	+	}()
			54	+	<-paneUnlocked
			55	+	close(release)
			56	+	<-store.writes
			57	+	}
			58	+
			59	+	func TestSnapshotEncodeWorkerDoesNotBlockPTYOrCoordinatorRender(t *testing.T) {
			60	+	exerciseBlockedSnapshotWorker(t, true)
			61	+	}
			62	+
			63	+	func TestSnapshotStoreWorkerDoesNotBlockPTYOrCoordinatorRender(t *testing.T) {
			64	+	exerciseBlockedSnapshotWorker(t, false)
			65	+	}
			66	+
			67	+	func exerciseBlockedSnapshotWorker(t *testing.T, blockEncode bool) {
			68	+	t.Helper()
			69	+	store := newGatedSnapshotStore()
			70	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
			71	+	WithSnapshotStore(store)(d)
			72	+	ctx, cancel := context.WithCancel(t.Context())
			73	+	d.startSnapshotEncodeWorker(ctx)
			74	+	t.Cleanup(func() {
			75	+	store.unblock()
			76	+	cancel()
			77	+	d.stopSnapshotEncodeWorker()
			78	+	})
			79	+	if blockEncode {
			80	+	d.snapshotMarshal = func(s snapcodec.Session) ([]byte, error) {
			81	+	store.enter()
			82	+	<-store.release
			83	+	return snapcodec.Marshal(s)
			84	+	}
			85	+	}
			86	+
			87	+	sess := newSnapshotTestSession(t, "live", false, "/live")
			88	+	d.sessions[sess.id] = sess
			89	+	sess.tabs[0].size = domain.Size{Cols: 80, Rows: 24}
			90	+	p := sess.tabs[0].panes["pane-1"]
			91	+	tr, sends, _ := newConn(t, ports.Frame{})
			92	+	ac := &attachedClient{tr: tr, output: newOutputStateStream(), size: domain.Size{Cols: 8, Rows: 3}}
			93	+	ac.setSession(sess)
			94	+	sess.mu.Lock()
			95	+	sess.client = ac
			96	+	sess.mu.Unlock()
			97	+	rendered := make(chan struct{}, 1)
			98	+	rc := newRenderCoordinator(renderCoordinatorOptions{
			99	+	clock: stubClock{},
			100	+	onInvalidate: func(renderInvalidation) {
			101	+	d.paint(sess, ac, true)
			102	+	rendered <- struct{}{}
			103	+	},
			104	+	})
			105	+	sess.installRenderCoordinator(rc)
			106	+	rc.attach(ac)
			107	+	require.True(t, rc.markAttachmentReady(ac))
			108	+	t.Cleanup(rc.noteSessionTeardown)
			109	+
			110	+	markSnapshotDirty(sess)
			111	+	require.True(t, d.scheduleSnapshot(sess))
			112	+	<-store.entered
			113	+
			114	+	paneUnlocked := make(chan struct{})
			115	+	go func() {
			116	+	p.mu.Lock()
			117	+	p.mu.Unlock()
			118	+	close(paneUnlocked)
			119	+	}()
			120	+	<-paneUnlocked
			121	+
			122	+	reader := portsmocks.NewMockPTY(t)
			123	+	var reads atomic.Int32
			124	+	reader.EXPECT().Read(mock.Anything).RunAndReturn(func(buf []byte) (int, error) {
			125	+	if reads.Add(1) == 1 {
			126	+	return copy(buf, []byte("ok")), nil
			127	+	}
			128	+	return 0, io.EOF
			129	+	}).Twice()
			130	+	reader.EXPECT().Write(mock.Anything).Return(0, nil).Maybe()
			131	+	p.mu.Lock()
			132	+	p.pty = reader
			133	+	p.onExit = func() {}
			134	+	p.mu.Unlock()
			135	+	d.sessWg.Add(1)
			136	+	go d.ptyReader(sess, sess.tabs[0], p)
			137	+	d.sessWg.Wait()
			138	+	<-rendered
			139	+	_ = awaitFrame(t, sends, ports.MsgOutput)
			140	+	p.mu.Lock()
			141	+	require.Contains(t, screenLineText(p.screen, 0), "ok")
			142	+	p.mu.Unlock()
			143	+
			144	+	store.unblock()
			145	+	awaitSnapshotIdle(t, sess)
			146	+	require.True(t, sess.snapDirty.Load(), "PTY output published while blocked must remain retryable")
			147	+	}
			148	+
			149	+	func TestStoppedSnapshotWorkerLeavesCaptureRetryable(t *testing.T) {
			150	+	store := &channelSnapshotStore{writes: make(chan []byte, 1)}
			151	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
			152	+	WithSnapshotStore(store)(d)
			153	+	ctx, cancel := context.WithCancel(t.Context())
			154	+	d.startSnapshotEncodeWorker(ctx)
			155	+	d.stopSnapshotEncodeWorker()
			156	+	t.Cleanup(cancel)
			157	+
			158	+	sess := newSnapshotTestSession(t, "stopped", false, "/work")
			159	+	markSnapshotDirty(sess)
			160	+	require.False(t, d.scheduleSnapshot(sess), "a stopped worker must not accept a capture")
			161	+	awaitSnapshotIdle(t, sess)
			162	+	require.True(t, sess.snapDirty.Load(), "an unsubmitted capture must remain retryable")
			163	+	}
			164	+
			165	+	func TestSnapshotQueueSaturationLeavesCaptureRetryable(t *testing.T) {
			166	+	store := &channelSnapshotStore{writes: make(chan []byte, 3)}
			167	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
			168	+	WithSnapshotStore(store)(d)
			169	+	startSnapshotEncodeWorker(t, d)
			170	+	entered := make(chan struct{})
			171	+	release := make(chan struct{})
			172	+	var calls atomic.Int32
			173	+	d.snapshotMarshal = func(s snapcodec.Session) ([]byte, error) {
			174	+	if calls.Add(1) == 1 {
			175	+	close(entered)
			176	+	<-release
			177	+	}
			178	+	return snapcodec.Marshal(s)
			179	+	}
			180	+	first := newSnapshotTestSession(t, "first", false, "/work")
			181	+	second := newSnapshotTestSession(t, "second", false, "/work")
			182	+	third := newSnapshotTestSession(t, "third", false, "/work")
			183	+	markSnapshotDirty(first)
			184	+	markSnapshotDirty(second)
			185	+	markSnapshotDirty(third)
			186	+	require.True(t, d.scheduleSnapshot(first))
			187	+	<-entered
			188	+	require.True(t, d.scheduleSnapshot(second), "one capture may wait in the bounded queue")
			189	+	require.False(t, d.scheduleSnapshot(third), "full queue must not block the producer")
			190	+	require.True(t, third.snapDirty.Load())
			191	+
			192	+	close(release)
			193	+	awaitSnapshotClean(t, first)
			194	+	awaitSnapshotClean(t, second)
			195	+	require.True(t, d.scheduleSnapshot(third))
			196	+	<-store.writes
			197	+	awaitSnapshotClean(t, third)
			198	+	}
			199	+
			200	+	func TestSnapshotSuccessDoesNotClearNewerGeneration(t *testing.T) {
			201	+	store := &channelSnapshotStore{writes: make(chan []byte, 2)}
			202	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
			203	+	WithSnapshotStore(store)(d)
			204	+	startSnapshotEncodeWorker(t, d)
			205	+	entered := make(chan struct{})
			206	+	release := make(chan struct{})
			207	+	var calls atomic.Int32
			208	+	d.snapshotMarshal = func(s snapcodec.Session) ([]byte, error) {
			209	+	if calls.Add(1) == 1 {
			210	+	close(entered)
			211	+	<-release
			212	+	}
			213	+	return snapcodec.Marshal(s)
			214	+	}
			215	+	sess := newSnapshotTestSession(t, "generation", false, "/work")
			216	+
			217	+	markSnapshotDirty(sess)
			218	+	require.True(t, d.scheduleSnapshot(sess))
			219	+	<-entered
			220	+	markSnapshotDirty(sess)
			221	+	close(release)
			222	+	<-store.writes
			223	+	awaitSnapshotIdle(t, sess)
			224	+	require.True(t, sess.snapDirty.Load(), "a stale successful capture must not clear newer state")
			225	+
			226	+	require.True(t, d.scheduleSnapshot(sess))
			227	+	<-store.writes
			228	+	awaitSnapshotClean(t, sess)
			229	+	}
			230	+
			231	+	func TestSnapshotEncodeFailureLeavesSessionDirtyAndRetries(t *testing.T) {
			232	+	store := &channelSnapshotStore{writes: make(chan []byte, 1)}
			233	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
			234	+	WithSnapshotStore(store)(d)
			235	+	startSnapshotEncodeWorker(t, d)
			236	+	sess := newSnapshotTestSession(t, "retry-encode", false, "/work")
			237	+	failed := make(chan struct{}, 1)
			238	+	var calls atomic.Int32
			239	+	d.snapshotMarshal = func(s snapcodec.Session) ([]byte, error) {
			240	+	if calls.Add(1) == 1 {
			241	+	failed <- struct{}{}
			242	+	return nil, errors.New("encode failed")
			243	+	}
			244	+	return snapcodec.Marshal(s)
			245	+	}
			246	+
			247	+	require.True(t, d.captureSession(sess))
			248	+	<-failed
			249	+	awaitSnapshotIdle(t, sess)
			250	+	require.True(t, sess.snapDirty.Load())
			251	+	require.True(t, d.scheduleSnapshot(sess))
			252	+	<-store.writes
			253	+	awaitSnapshotClean(t, sess)
			254	+	}
			255	+
23		func TestSnapshotSaverWritesDirtyNamedSessionsOnly(t *testing.T) {	256		func TestSnapshotSaverWritesDirtyNamedSessionsOnly(t *testing.T) {
24		clock := newManualSnapshotClock()	257		clock := newManualSnapshotClock()
25		store := portsmocks.NewMockSnapshotStore(t)	258		store := portsmocks.NewMockSnapshotStore(t)
26		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)	259		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
27		WithSnapshotStore(store)(d)	260		WithSnapshotStore(store)(d)
28		d.procCwd = func(int) (string, error) { return "/pane", nil }	261		d.procCwd = func(int) (string, error) { return "/pane", nil }
			262	+	startSnapshotEncodeWorker(t, d)
29			263
30		dirty := newSnapshotTestSession(t, "dirty", false, "/fallback")	264		dirty := newSnapshotTestSession(t, "dirty", false, "/fallback")
31		clean := newSnapshotTestSession(t, "clean", false, "/fallback")	265		clean := newSnapshotTestSession(t, "clean", false, "/fallback")

@@ -67,6 +301,7 @@ func TestSnapshotCaptureExcludesFloatingRuntime(t *testing.T) {
67		store := portsmocks.NewMockSnapshotStore(t)	301		store := portsmocks.NewMockSnapshotStore(t)
68		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})	302		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
69		WithSnapshotStore(store)(d)	303		WithSnapshotStore(store)(d)
			304	+	startSnapshotEncodeWorker(t, d)
70		sess := newSnapshotTestSession(t, "work", false, "/fallback")	305		sess := newSnapshotTestSession(t, "work", false, "/fallback")
71		tb := sess.tabs[0]	306		tb := sess.tabs[0]
72		floatingPTY, _ := newBlockingPTY(t)	307		floatingPTY, _ := newBlockingPTY(t)

@@ -85,12 +320,14 @@ func TestSnapshotCaptureExcludesFloatingRuntime(t *testing.T) {
85		return nil	320		return nil
86		}).Once()	321		}).Once()
87		require.True(t, d.captureSession(sess))	322		require.True(t, d.captureSession(sess))
			323	+	awaitSnapshotClean(t, sess)
88		}	324		}
89			325
90		func TestSnapshotCaptureStoresForegroundProcessMetadata(t *testing.T) {	326		func TestSnapshotCaptureStoresForegroundProcessMetadata(t *testing.T) {
91		store := portsmocks.NewMockSnapshotStore(t)	327		store := portsmocks.NewMockSnapshotStore(t)
92		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})	328		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
93		WithSnapshotStore(store)(d)	329		WithSnapshotStore(store)(d)
			330	+	startSnapshotEncodeWorker(t, d)
94		d.procGroupArgv = func(pgid int, shellPid int) ([]string, error) {	331		d.procGroupArgv = func(pgid int, shellPid int) ([]string, error) {
95		require.Equal(t, 4321, pgid)	332		require.Equal(t, 4321, pgid)
96		require.Equal(t, 1234, shellPid)	333		require.Equal(t, 1234, shellPid)

@@ -113,12 +350,14 @@ func TestSnapshotCaptureStoresForegroundProcessMetadata(t *testing.T) {
113		}).Once()	350		}).Once()
114			351
115		require.True(t, d.captureSession(sess))	352		require.True(t, d.captureSession(sess))
			353	+	awaitSnapshotClean(t, sess)
116		}	354		}
117			355
118		func TestSnapshotCaptureProcessFailureDoesNotBlockSnapshot(t *testing.T) {	356		func TestSnapshotCaptureProcessFailureDoesNotBlockSnapshot(t *testing.T) {
119		store := portsmocks.NewMockSnapshotStore(t)	357		store := portsmocks.NewMockSnapshotStore(t)
120		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})	358		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
121		WithSnapshotStore(store)(d)	359		WithSnapshotStore(store)(d)
			360	+	startSnapshotEncodeWorker(t, d)
122		d.procGroupArgv = func(int, int) ([]string, error) { return nil, errors.New("inspect failed") }	361		d.procGroupArgv = func(int, int) ([]string, error) { return nil, errors.New("inspect failed") }
123			362
124		sess := newSnapshotTestSession(t, "work", false, "/fallback")	363		sess := newSnapshotTestSession(t, "work", false, "/fallback")

@@ -133,12 +372,14 @@ func TestSnapshotCaptureProcessFailureDoesNotBlockSnapshot(t *testing.T) {
133		}).Once()	372		}).Once()
134			373
135		require.True(t, d.captureSession(sess))	374		require.True(t, d.captureSession(sess))
			375	+	awaitSnapshotClean(t, sess)
136		}	376		}
137			377
138		func TestSnapshotCaptureSkipsBareShellProcess(t *testing.T) {	378		func TestSnapshotCaptureSkipsBareShellProcess(t *testing.T) {
139		store := portsmocks.NewMockSnapshotStore(t)	379		store := portsmocks.NewMockSnapshotStore(t)
140		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})	380		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
141		WithSnapshotStore(store)(d)	381		WithSnapshotStore(store)(d)
			382	+	startSnapshotEncodeWorker(t, d)
142		d.procGroupArgv = func(int, int) ([]string, error) {	383		d.procGroupArgv = func(int, int) ([]string, error) {
143		t.Fatal("bare shell should not inspect argv")	384		t.Fatal("bare shell should not inspect argv")
144		return nil, nil	385		return nil, nil

@@ -156,35 +397,37 @@ func TestSnapshotCaptureSkipsBareShellProcess(t *testing.T) {
156		}).Once()	397		}).Once()
157			398
158		require.True(t, d.captureSession(sess))	399		require.True(t, d.captureSession(sess))
			400	+	awaitSnapshotClean(t, sess)
159		}	401		}
160			402
161		func TestKillSessionSnapshotsNamedSessionBeforeClosingPanes(t *testing.T) {	403		func TestKillSessionSnapshotsNamedSessionBeforeClosingPanes(t *testing.T) {
162		store := portsmocks.NewMockSnapshotStore(t)	404		store := portsmocks.NewMockSnapshotStore(t)
163		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})	405		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
164		WithSnapshotStore(store)(d)	406		WithSnapshotStore(store)(d)
			407	+	startSnapshotEncodeWorker(t, d)
165		d.procCwd = func(int) (string, error) { return "/live", nil }	408		d.procCwd = func(int) (string, error) { return "/live", nil }
166			409
167		sess := newSnapshotTestSession(t, "work", false, "/fallback")	410		sess := newSnapshotTestSession(t, "work", false, "/fallback")
168		d.sessions[sess.id] = sess	411		d.sessions[sess.id] = sess
169			412
170	-	var wrote atomic.Bool	413	+	var closed atomic.Bool
171		store.EXPECT().Write("work", mock.Anything).RunAndReturn(func(_ string, data []byte) error {	414		store.EXPECT().Write("work", mock.Anything).RunAndReturn(func(_ string, data []byte) error {
172		snap, err := snapcodec.Unmarshal(data)	415		snap, err := snapcodec.Unmarshal(data)
173		require.NoError(t, err)	416		require.NoError(t, err)
174		require.Equal(t, "work", snap.Name)	417		require.Equal(t, "work", snap.Name)
175		require.Equal(t, "/live", snap.Tabs[0].Panes[0].Cwd)	418		require.Equal(t, "/live", snap.Tabs[0].Panes[0].Cwd)
176	-	wrote.Store(true)	419	+	require.True(t, closed.Load(), "snapshot persistence must not delay pane close")
177		return nil	420		return nil
178		}).Once()	421		}).Once()
179		mockPTY, ok := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)	422		mockPTY, ok := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
180		require.True(t, ok)	423		require.True(t, ok)
181		mockPTY.EXPECT().Close().RunAndReturn(func() error {	424		mockPTY.EXPECT().Close().RunAndReturn(func() error {
182	-	require.True(t, wrote.Load(), "snapshot Write must happen before pane Close")	425	+	closed.Store(true)
183		return nil	426		return nil
184		}).Once()	427		}).Once()
185			428
186		require.NoError(t, d.killSession(sess, 0, false))	429		require.NoError(t, d.killSession(sess, 0, false))
187	-	require.True(t, wrote.Load())	430	+	awaitSnapshotClean(t, sess)
188		}	431		}
189			432
190		func TestPTYReaderSynchronizedUpdateFlushMarksSnapshotDirty(t *testing.T) {	433		func TestPTYReaderSynchronizedUpdateFlushMarksSnapshotDirty(t *testing.T) {

@@ -241,12 +484,17 @@ func TestSnapshotSaverKeepsDirtyWhenWriteFails(t *testing.T) {
241		store := portsmocks.NewMockSnapshotStore(t)	484		store := portsmocks.NewMockSnapshotStore(t)
242		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)	485		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
243		WithSnapshotStore(store)(d)	486		WithSnapshotStore(store)(d)
			487	+	startSnapshotEncodeWorker(t, d)
244			488
245		sess := newSnapshotTestSession(t, "retry", false, "/fallback")	489		sess := newSnapshotTestSession(t, "retry", false, "/fallback")
246		sess.snapDirty.Store(true)	490		sess.snapDirty.Store(true)
247		d.sessions[sess.id] = sess	491		d.sessions[sess.id] = sess
248			492
249	-	store.EXPECT().Write("retry", mock.Anything).Return(errors.New("boom")).Once()	493	+	failed := make(chan struct{}, 1)
			494	+	store.EXPECT().Write("retry", mock.Anything).RunAndReturn(func(string, []byte) error {
			495	+	failed <- struct{}{}
			496	+	return errors.New("boom")
			497	+	}).Once()
250		wrote := make(chan struct{}, 1)	498		wrote := make(chan struct{}, 1)
251		store.EXPECT().Write("retry", mock.Anything).RunAndReturn(func(string, []byte) error {	499		store.EXPECT().Write("retry", mock.Anything).RunAndReturn(func(string, []byte) error {
252		wrote <- struct{}{}	500		wrote <- struct{}{}

@@ -257,7 +505,9 @@ func TestSnapshotSaverKeepsDirtyWhenWriteFails(t *testing.T) {
257		defer cancel()	505		defer cancel()
258		go d.snapshotSaver(ctx)	506		go d.snapshotSaver(ctx)
259		clock.fireNext(t)	507		clock.fireNext(t)
260	-	require.Eventually(t, func() bool { return sess.snapDirty.Load() }, time.Second, time.Millisecond)	508	+	<-failed
			509	+	awaitSnapshotIdle(t, sess)
			510	+	require.True(t, sess.snapDirty.Load())
261		clock.fireNext(t)	511		clock.fireNext(t)
262		select {	512		select {
263		case <-wrote:	513		case <-wrote:

@@ -365,6 +615,7 @@ func TestSnapshotSaverCapturesLayoutOnlyDirtySave(t *testing.T) {
365		factory := portsmocks.NewMockPTYFactory(t)	615		factory := portsmocks.NewMockPTYFactory(t)
366		d := newTestDaemon(t, factory, clock)	616		d := newTestDaemon(t, factory, clock)
367		WithSnapshotStore(store)(d)	617		WithSnapshotStore(store)(d)
			618	+	startSnapshotEncodeWorker(t, d)
368			619
369		sess := newSnapshotTestSession(t, "work", false, "/work")	620		sess := newSnapshotTestSession(t, "work", false, "/work")
370		tctx, tcancel := context.WithCancel(context.Background())	621		tctx, tcancel := context.WithCancel(context.Background())

@@ -450,6 +701,74 @@ func newSnapshotTestSession(t *testing.T, name string, ephemeral bool, cwd strin
450		return sess	701		return sess
451		}	702		}
452			703
			704	+	func awaitSnapshotIdle(t testing.T, sess session) {
			705	+	t.Helper()
			706	+	for range 4096 {
			707	+	sess.snapshotMu.Lock()
			708	+	pending := sess.snapshotPending
			709	+	sess.snapshotMu.Unlock()
			710	+	if !pending {
			711	+	return
			712	+	}
			713	+	runtime.Gosched()
			714	+	}
			715	+	t.Fatal("snapshot worker did not become idle")
			716	+	}
			717	+
			718	+	func awaitSnapshotClean(t testing.T, sess session) {
			719	+	t.Helper()
			720	+	for range 4096 {
			721	+	if !sess.snapDirty.Load() {
			722	+	return
			723	+	}
			724	+	runtime.Gosched()
			725	+	}
			726	+	t.Fatal("snapshot worker did not finish")
			727	+	}
			728	+
			729	+	func startSnapshotEncodeWorker(t testing.T, d Daemon) {
			730	+	t.Helper()
			731	+	ctx, cancel := context.WithCancel(t.Context())
			732	+	d.startSnapshotEncodeWorker(ctx)
			733	+	t.Cleanup(func() {
			734	+	cancel()
			735	+	d.stopSnapshotEncodeWorker()
			736	+	})
			737	+	}
			738	+
			739	+	type gatedSnapshotStore struct {
			740	+	entered chan struct{}
			741	+	release chan struct{}
			742	+	once sync.Once
			743	+	releaseOnce sync.Once
			744	+	}
			745	+
			746	+	func newGatedSnapshotStore() *gatedSnapshotStore {
			747	+	return &gatedSnapshotStore{entered: make(chan struct{}), release: make(chan struct{})}
			748	+	}
			749	+
			750	+	func (s *gatedSnapshotStore) enter() { s.once.Do(func() { close(s.entered) }) }
			751	+	func (s *gatedSnapshotStore) unblock() { s.releaseOnce.Do(func() { close(s.release) }) }
			752	+
			753	+	func (s *gatedSnapshotStore) Write(_ string, _ []byte) error {
			754	+	s.enter()
			755	+	<-s.release
			756	+	return nil
			757	+	}
			758	+	func (*gatedSnapshotStore) Load() ([]ports.SnapshotBlob, error) { return nil, nil }
			759	+	func (*gatedSnapshotStore) Delete(string) error { return nil }
			760	+
			761	+	type channelSnapshotStore struct {
			762	+	writes chan []byte
			763	+	}
			764	+
			765	+	func (s *channelSnapshotStore) Write(_ string, data []byte) error {
			766	+	s.writes <- append([]byte(nil), data...)
			767	+	return nil
			768	+	}
			769	+	func (*channelSnapshotStore) Load() ([]ports.SnapshotBlob, error) { return nil, nil }
			770	+	func (*channelSnapshotStore) Delete(string) error { return nil }
			771	+
453		type manualSnapshotClock struct {	772		type manualSnapshotClock struct {
454		mu sync.Mutex	773		mu sync.Mutex
455		timers []*manualSnapshotTimer	774		timers []*manualSnapshotTimer

M internal/usecase/snapshot/codec_test.go +117 -0


@@ -114,6 +114,123 @@ func TestUnmarshalRejectsInvalidTreeReference(t *testing.T) {
114	114		}
115	115		}
116	116
	117	+	func TestUnmarshalRejectsMalformedTerminalBlobWithoutPanic(t *testing.T) {
	118	+	tail := mustHistoryTail(t)
	119	+	visible := visibleBlob(t, [][]renderer.Cell{{{Rune: 'v'}}})
	120	+	// The visible cell's underline-style byte is outside the renderer enum.
	121	+	visible[13+29] = 0xff
	122	+	body := v3Manifest(0, nil, nil, uint32(len(tail)), tail, uint32(len(visible)), visible)
	123	+	reject(t, v3Envelope(version, 0, body), ErrInvalidData)
	124	+	}
	125	+
	126	+	func TestRoundTripNilTreeAndRootDoNotMaterializeLeaf(t *testing.T) {
	127	+	for _, tc := range []struct {
	128	+	name string
	129	+	tab Tab
	130	+	}{
	131	+	{name: "nil tree", tab: Tab{Tree: nil}},
	132	+	{name: "nil root", tab: Tab{Tree: &layout.Tree{Root: nil}}},
	133	+	} {
	134	+	t.Run(tc.name, func(t *testing.T) {
	135	+	data, err := Marshal(Session{Name: "s", Tabs: []Tab{tc.tab}})
	136	+	requireNoError(t, err)
	137	+	got, err := Unmarshal(data)
	138	+	requireNoError(t, err)
	139	+	if got.Tabs[0].Tree == nil \|\| got.Tabs[0].Tree.Root != nil {
	140	+	t.Fatalf("tree = %#v, want nonnil tree with nil root", got.Tabs[0].Tree)
	141	+	}
	142	+	})
	143	+	}
	144	+	}
	145	+
	146	+	func TestRoundTripSessionMetadataAndProcess(t *testing.T) {
	147	+	sealed, tail := historyBlobs(t, [][]renderer.Cell{{{Rune: 'h'}}, {{Rune: 'i'}}})
	148	+	visible := visibleBlob(t, [][]renderer.Cell{{{Rune: 'v'}}})
	149	+	want := Session{Name: "named", CreatedAt: 42, Active: 1, Tabs: []Tab{
	150	+	{
	151	+	StableID: "t1", Cols: 100, Rows: 40, NextPaneID: 9, Focus: "2",
	152	+	Tree: &layout.Tree{Focus: "2", Root: &layout.Node{Kind: layout.Split, Dir: layout.Horizontal, Children: []*layout.Node{
	153	+	layout.NewLeaf("1"), layout.NewLeaf("2"),
	154	+	}}},
	155	+	Panes: []Pane{
	156	+	{ID: "1", StableID: "p1", Cwd: "/one", SealedChunks: sealed, Tail: tail, Visible: visible, Process: &Process{Argv: []string{"pi", "--resume"}, Strategy: "pi", Opts: ProcessOpts{AgentSessionID: "agent-123"}}},
	157	+	{ID: "2", StableID: "p2", Cwd: "/two", SealedChunks: sealed, Tail: tail, Visible: visible},
	158	+	},
	159	+	},
	160	+	{StableID: "t2", Cols: 80, Rows: 24, NextPaneID: 2, Focus: "a", Tree: layout.NewTree("a"), Panes: []Pane{{ID: "a", Cwd: "/tmp", SealedChunks: sealed, Tail: tail, Visible: visible}}},
	161	+	}}
	162	+	data, err := Marshal(want)
	163	+	requireNoError(t, err)
	164	+	got, err := Unmarshal(data)
	165	+	requireNoError(t, err)
	166	+	if got.Name != want.Name \|\| got.CreatedAt != want.CreatedAt \|\| got.Active != want.Active \|\| len(got.Tabs) != 2 {
	167	+	t.Fatalf("session metadata round trip = %#v", got)
	168	+	}
	169	+	pane := got.Tabs[0].Panes[0]
	170	+	if pane.ID != "1" \|\| pane.StableID != "p1" \|\| pane.Cwd != "/one" \|\| pane.Process == nil \|\| pane.Process.Opts.AgentSessionID != "agent-123" \|\| got.Tabs[0].Tree.Root.Kind != layout.Split {
	171	+	t.Fatalf("pane metadata round trip = %#v", pane)
	172	+	}
	173	+	}
	174	+
	175	+	func TestMarshalRejectsProcessWithoutArgv(t *testing.T) {
	176	+	_, err := Marshal(Session{Name: "s", Tabs: []Tab{
	177	+	{Tree: layout.NewTree("p"), Focus: "p", Panes: []Pane{{ID: "p", Process: &Process{Strategy: "generic"}}}},
	178	+	}})
	179	+	if !errors.Is(err, ErrInvalidData) {
	180	+	t.Fatalf("Marshal() error = %v, want %v", err, ErrInvalidData)
	181	+	}
	182	+	}
	183	+
	184	+	func TestUnmarshalRejectsProcessWithoutArgv(t *testing.T) {
	185	+	tail := mustHistoryTail(t)
	186	+	visible := visibleBlob(t, [][]renderer.Cell{{{Rune: 'v'}}})
	187	+	var w payloadWriter
	188	+	_ = w.putString("s")
	189	+	w.putUint64(0)
	190	+	w.putUint16(0)
	191	+	w.putUint16(1)
	192	+	_ = w.putString("t")
	193	+	w.putUint16(80)
	194	+	w.putUint16(24)
	195	+	w.putUint64(1)
	196	+	_ = w.putString("p")
	197	+	_ = writeNode(&w, layout.NewLeaf("p"))
	198	+	w.putUint16(1)
	199	+	_ = w.putString("p")
	200	+	_ = w.putString("")
	201	+	_ = w.putString("")
	202	+	w.putUint32(0)
	203	+	w.putUint32(uint32(len(tail)))
	204	+	w.b = append(w.b, tail...)
	205	+	w.putUint32(uint32(len(visible)))
	206	+	w.b = append(w.b, visible...)
	207	+	w.putUint8(1)
	208	+	w.putUint16(0) // argv count
	209	+	_ = w.putString("generic")
	210	+	_ = w.putString("")
	211	+	if _, err := Unmarshal(v3Envelope(version, 0, w.b)); !errors.Is(err, ErrInvalidData) {
	212	+	t.Fatalf("Unmarshal() error = %v, want %v", err, ErrInvalidData)
	213	+	}
	214	+	}
	215	+
	216	+	func TestUnmarshalRejectsAggregateOpaqueBlobDeclarations(t *testing.T) {
	217	+	body := v3Manifest(math.MaxUint32, nil, nil, math.MaxUint32, nil, math.MaxUint32, nil)
	218	+	reject(t, v3Envelope(version, 0, body), ErrInvalidData)
	219	+	}
	220	+
	221	+	func mustHistoryTail(t *testing.T) []byte {
	222	+	t.Helper()
	223	+	_, tail := historyBlobs(t, nil)
	224	+	return tail
	225	+	}
	226	+
	227	+	func requireNoError(t *testing.T, err error) {
	228	+	t.Helper()
	229	+	if err != nil {
	230	+	t.Fatal(err)
	231	+	}
	232	+	}
	233	+
117	234		func historyBlobs(t *testing.T, rows [][]renderer.Cell) ([][]byte, []byte) {
118	235		t.Helper()
119	236		h := vt.NewHistory(vt.HistoryConfig{MaxRows: 128, ChunkRows: 2})


@@ -114,6 +114,123 @@ func TestUnmarshalRejectsInvalidTreeReference(t *testing.T) {
114	}	114		}
115	}	115		}
116		116
		117	+	func TestUnmarshalRejectsMalformedTerminalBlobWithoutPanic(t *testing.T) {
		118	+	tail := mustHistoryTail(t)
		119	+	visible := visibleBlob(t, [][]renderer.Cell{{{Rune: 'v'}}})
		120	+	// The visible cell's underline-style byte is outside the renderer enum.
		121	+	visible[13+29] = 0xff
		122	+	body := v3Manifest(0, nil, nil, uint32(len(tail)), tail, uint32(len(visible)), visible)
		123	+	reject(t, v3Envelope(version, 0, body), ErrInvalidData)
		124	+	}
		125	+
		126	+	func TestRoundTripNilTreeAndRootDoNotMaterializeLeaf(t *testing.T) {
		127	+	for _, tc := range []struct {
		128	+	name string
		129	+	tab Tab
		130	+	}{
		131	+	{name: "nil tree", tab: Tab{Tree: nil}},
		132	+	{name: "nil root", tab: Tab{Tree: &layout.Tree{Root: nil}}},
		133	+	} {
		134	+	t.Run(tc.name, func(t *testing.T) {
		135	+	data, err := Marshal(Session{Name: "s", Tabs: []Tab{tc.tab}})
		136	+	requireNoError(t, err)
		137	+	got, err := Unmarshal(data)
		138	+	requireNoError(t, err)
		139	+	if got.Tabs[0].Tree == nil \|\| got.Tabs[0].Tree.Root != nil {
		140	+	t.Fatalf("tree = %#v, want nonnil tree with nil root", got.Tabs[0].Tree)
		141	+	}
		142	+	})
		143	+	}
		144	+	}
		145	+
		146	+	func TestRoundTripSessionMetadataAndProcess(t *testing.T) {
		147	+	sealed, tail := historyBlobs(t, [][]renderer.Cell{{{Rune: 'h'}}, {{Rune: 'i'}}})
		148	+	visible := visibleBlob(t, [][]renderer.Cell{{{Rune: 'v'}}})
		149	+	want := Session{Name: "named", CreatedAt: 42, Active: 1, Tabs: []Tab{
		150	+	{
		151	+	StableID: "t1", Cols: 100, Rows: 40, NextPaneID: 9, Focus: "2",
		152	+	Tree: &layout.Tree{Focus: "2", Root: &layout.Node{Kind: layout.Split, Dir: layout.Horizontal, Children: []*layout.Node{
		153	+	layout.NewLeaf("1"), layout.NewLeaf("2"),
		154	+	}}},
		155	+	Panes: []Pane{
		156	+	{ID: "1", StableID: "p1", Cwd: "/one", SealedChunks: sealed, Tail: tail, Visible: visible, Process: &Process{Argv: []string{"pi", "--resume"}, Strategy: "pi", Opts: ProcessOpts{AgentSessionID: "agent-123"}}},
		157	+	{ID: "2", StableID: "p2", Cwd: "/two", SealedChunks: sealed, Tail: tail, Visible: visible},
		158	+	},
		159	+	},
		160	+	{StableID: "t2", Cols: 80, Rows: 24, NextPaneID: 2, Focus: "a", Tree: layout.NewTree("a"), Panes: []Pane{{ID: "a", Cwd: "/tmp", SealedChunks: sealed, Tail: tail, Visible: visible}}},
		161	+	}}
		162	+	data, err := Marshal(want)
		163	+	requireNoError(t, err)
		164	+	got, err := Unmarshal(data)
		165	+	requireNoError(t, err)
		166	+	if got.Name != want.Name \|\| got.CreatedAt != want.CreatedAt \|\| got.Active != want.Active \|\| len(got.Tabs) != 2 {
		167	+	t.Fatalf("session metadata round trip = %#v", got)
		168	+	}
		169	+	pane := got.Tabs[0].Panes[0]
		170	+	if pane.ID != "1" \|\| pane.StableID != "p1" \|\| pane.Cwd != "/one" \|\| pane.Process == nil \|\| pane.Process.Opts.AgentSessionID != "agent-123" \|\| got.Tabs[0].Tree.Root.Kind != layout.Split {
		171	+	t.Fatalf("pane metadata round trip = %#v", pane)
		172	+	}
		173	+	}
		174	+
		175	+	func TestMarshalRejectsProcessWithoutArgv(t *testing.T) {
		176	+	_, err := Marshal(Session{Name: "s", Tabs: []Tab{
		177	+	{Tree: layout.NewTree("p"), Focus: "p", Panes: []Pane{{ID: "p", Process: &Process{Strategy: "generic"}}}},
		178	+	}})
		179	+	if !errors.Is(err, ErrInvalidData) {
		180	+	t.Fatalf("Marshal() error = %v, want %v", err, ErrInvalidData)
		181	+	}
		182	+	}
		183	+
		184	+	func TestUnmarshalRejectsProcessWithoutArgv(t *testing.T) {
		185	+	tail := mustHistoryTail(t)
		186	+	visible := visibleBlob(t, [][]renderer.Cell{{{Rune: 'v'}}})
		187	+	var w payloadWriter
		188	+	_ = w.putString("s")
		189	+	w.putUint64(0)
		190	+	w.putUint16(0)
		191	+	w.putUint16(1)
		192	+	_ = w.putString("t")
		193	+	w.putUint16(80)
		194	+	w.putUint16(24)
		195	+	w.putUint64(1)
		196	+	_ = w.putString("p")
		197	+	_ = writeNode(&w, layout.NewLeaf("p"))
		198	+	w.putUint16(1)
		199	+	_ = w.putString("p")
		200	+	_ = w.putString("")
		201	+	_ = w.putString("")
		202	+	w.putUint32(0)
		203	+	w.putUint32(uint32(len(tail)))
		204	+	w.b = append(w.b, tail...)
		205	+	w.putUint32(uint32(len(visible)))
		206	+	w.b = append(w.b, visible...)
		207	+	w.putUint8(1)
		208	+	w.putUint16(0) // argv count
		209	+	_ = w.putString("generic")
		210	+	_ = w.putString("")
		211	+	if _, err := Unmarshal(v3Envelope(version, 0, w.b)); !errors.Is(err, ErrInvalidData) {
		212	+	t.Fatalf("Unmarshal() error = %v, want %v", err, ErrInvalidData)
		213	+	}
		214	+	}
		215	+
		216	+	func TestUnmarshalRejectsAggregateOpaqueBlobDeclarations(t *testing.T) {
		217	+	body := v3Manifest(math.MaxUint32, nil, nil, math.MaxUint32, nil, math.MaxUint32, nil)
		218	+	reject(t, v3Envelope(version, 0, body), ErrInvalidData)
		219	+	}
		220	+
		221	+	func mustHistoryTail(t *testing.T) []byte {
		222	+	t.Helper()
		223	+	_, tail := historyBlobs(t, nil)
		224	+	return tail
		225	+	}
		226	+
		227	+	func requireNoError(t *testing.T, err error) {
		228	+	t.Helper()
		229	+	if err != nil {
		230	+	t.Fatal(err)
		231	+	}
		232	+	}
		233	+
117	func historyBlobs(t *testing.T, rows [][]renderer.Cell) ([][]byte, []byte) {	234		func historyBlobs(t *testing.T, rows [][]renderer.Cell) ([][]byte, []byte) {
118	t.Helper()	235		t.Helper()
119	h := vt.NewHistory(vt.HistoryConfig{MaxRows: 128, ChunkRows: 2})	236		h := vt.NewHistory(vt.HistoryConfig{MaxRows: 128, ChunkRows: 2})