33d8242 test(daemon): align recovery integration fixtures · vev

M internal/usecase/daemon/legacy_snapshot_test.go +1 -14

@@ -1,21 +1,8 @@
1	1		package daemon
2	2
3		-	import (
4		-	"bytes"
5		-	"context"
6		-	"fmt"
7		-
8		-	"github.com/bnema/vev/internal/ports"
9		-	)
	3	+	import "context"
10	4
11	5		// importLegacySnapshots is retained only for tests covering the retired
12	6		// pre-migration bridge. Normal runtime legacy decoding lives exclusively in
13	7		// the snapshot migration adapter.
14	8		func (d *Daemon) importLegacySnapshots(context.Context) {}
15		-
16		-	func verifyLegacyImportGeneration(generation ports.SnapshotGeneration, publication ports.SnapshotPublication) error {
17		-	if generation.Name != publication.Name \|\| generation.Generation != publication.Generation \|\| !bytes.Equal(generation.Manifest, publication.Manifest) {
18		-	return fmt.Errorf("snapshot: legacy import verification mismatch")
19		-	}
20		-	return nil
21		-	}

@@ -1,21 +1,8 @@
1		package daemon	1		package daemon
2			2
3	-	import (	3	+	import "context"
4	-	"bytes"
5	-	"context"
6	-	"fmt"
7	-
8	-	"github.com/bnema/vev/internal/ports"
9	-	)
10			4
11		// importLegacySnapshots is retained only for tests covering the retired	5		// importLegacySnapshots is retained only for tests covering the retired
12		// pre-migration bridge. Normal runtime legacy decoding lives exclusively in	6		// pre-migration bridge. Normal runtime legacy decoding lives exclusively in
13		// the snapshot migration adapter.	7		// the snapshot migration adapter.
14		func (d *Daemon) importLegacySnapshots(context.Context) {}	8		func (d *Daemon) importLegacySnapshots(context.Context) {}
15	-
16	-	func verifyLegacyImportGeneration(generation ports.SnapshotGeneration, publication ports.SnapshotPublication) error {
17	-	if generation.Name != publication.Name \|\| generation.Generation != publication.Generation \|\| !bytes.Equal(generation.Manifest, publication.Manifest) {
18	-	return fmt.Errorf("snapshot: legacy import verification mismatch")
19	-	}
20	-	return nil
21	-	}

M internal/usecase/daemon/performance_benchmark_test.go +8 -8


@@ -627,6 +627,12 @@ func publishBenchmarkSnapshot(b testing.TB, fixture *performanceFixture, generat
627	627		if err := fixture.snapshots.Publish(context.Background(), publication); err != nil {
628	628		b.Fatal(err)
629	629		}
	630	+	checkpoint := domain.CheckpointRef{Generation: generation, ManifestDigest: snapcodec.ManifestDigest(publication.Manifest)}
	631	+	capture.checkpoint = checkpoint
	632	+	fixture.sess.snapshotMu.Lock()
	633	+	fixture.sess.snapshotPublishedGeneration = generation
	634	+	fixture.sess.snapshotPublishedCheckpoint = &checkpoint
	635	+	fixture.sess.snapshotMu.Unlock()
630	636		markSnapshotCaptureObjectsPublished(capture)
631	637		}
632	638

@@ -1036,6 +1042,7 @@ func newPerformanceFixtureWithCleanup(t testing.TB, config performanceConfig, re
1036	1042		ac.size = config.size
1037	1043		sess.name = "performance"
1038	1044		sess.ephemeral = false
	1045	+	sess.incarnation = domain.IncarnationID{1}
1039	1046		sess.snapEligible.Store(true)
1040	1047
1041	1048		fixture := &performanceFixture{

@@ -1376,6 +1383,7 @@ type countingSnapshotMetrics struct {
1376	1383		writes, objectBytes, historyBlobBytes, suppliedObjectBytes, suppliedHistoryBytes, manifestBytes, headBytes uint64
1377	1384		}
1378	1385		type countingSnapshotRepository struct {
	1386	+	noOpSnapshotRepository
1379	1387		mu sync.Mutex
1380	1388		countingSnapshotMetrics
1381	1389		objects map[string]map[ports.SnapshotDigest]struct{}

@@ -1418,14 +1426,6 @@ func (s *countingSnapshotRepository) Publish(_ context.Context, publication port
1418	1426		s.last = publication.Manifest
1419	1427		return nil
1420	1428		}
1421		-	func (*countingSnapshotRepository) List(context.Context) ([]string, error) { return nil, nil }
1422		-	func (*countingSnapshotRepository) Load(context.Context, string) (ports.SnapshotGeneration, error) {
1423		-	return ports.SnapshotGeneration{}, nil
1424		-	}
1425		-	func (*countingSnapshotRepository) Delete(context.Context, string) error { return nil }
1426		-	func (*countingSnapshotRepository) Tombstone(context.Context, string) error { return nil }
1427		-	func (*countingSnapshotRepository) DeleteTombstone(context.Context, string) error { return nil }
1428		-	func (*countingSnapshotRepository) Maintain(context.Context) error { return nil }
1429	1429		func (s *countingSnapshotRepository) reset() {
1430	1430		s.mu.Lock()
1431	1431		s.countingSnapshotMetrics = countingSnapshotMetrics{}

 	if err := fixture.snapshots.Publish(context.Background(), publication); err != nil {
 		b.Fatal(err)
 	}
+	checkpoint := domain.CheckpointRef{Generation: generation, ManifestDigest: snapcodec.ManifestDigest(publication.Manifest)}
+	capture.checkpoint = checkpoint
+	fixture.sess.snapshotMu.Lock()
+	fixture.sess.snapshotPublishedGeneration = generation
+	fixture.sess.snapshotPublishedCheckpoint = &checkpoint
+	fixture.sess.snapshotMu.Unlock()
 	markSnapshotCaptureObjectsPublished(capture)
 }
 	ac.size = config.size
 	sess.name = "performance"
 	sess.ephemeral = false
+	sess.incarnation = domain.IncarnationID{1}
 	sess.snapEligible.Store(true)
 	fixture := &performanceFixture{
 	writes, objectBytes, historyBlobBytes, suppliedObjectBytes, suppliedHistoryBytes, manifestBytes, headBytes uint64
 }
 type countingSnapshotRepository struct {
+	noOpSnapshotRepository
 	mu sync.Mutex
 	countingSnapshotMetrics
 	objects map[string]map[ports.SnapshotDigest]struct{}
 	s.last = publication.Manifest
 	return nil
 }
-func (*countingSnapshotRepository) List(context.Context) ([]string, error) { return nil, nil }
-func (*countingSnapshotRepository) Load(context.Context, string) (ports.SnapshotGeneration, error) {
-	return ports.SnapshotGeneration{}, nil
-}
-func (*countingSnapshotRepository) Delete(context.Context, string) error          { return nil }
-func (*countingSnapshotRepository) Tombstone(context.Context, string) error       { return nil }
-func (*countingSnapshotRepository) DeleteTombstone(context.Context, string) error { return nil }
-func (*countingSnapshotRepository) Maintain(context.Context) error                { return nil }
 func (s *countingSnapshotRepository) reset() {
 	s.mu.Lock()
 	s.countingSnapshotMetrics = countingSnapshotMetrics{}

M internal/usecase/daemon/picker_test.go +6 -10


@@ -45,21 +45,17 @@ func (failingDeleteStore) Range(func(k, v []byte) bool) {}
45	45		func (failingDeleteStore) Sync() error { return nil }
46	46		func (failingDeleteStore) Close() error { return nil }
47	47
48		-	type refusingSnapshotDeleteRepository struct{ err error }
	48	+	type refusingSnapshotDeleteRepository struct {
	49	+	noOpSnapshotRepository
	50	+	err error
	51	+	}
49	52
50	53		func (refusingSnapshotDeleteRepository) Publish(context.Context, ports.SnapshotPublication) error {
51	54		return nil
52	55		}
53		-	func (refusingSnapshotDeleteRepository) List(context.Context) ([]string, error) { return nil, nil }
54		-	func (refusingSnapshotDeleteRepository) Load(context.Context, string) (ports.SnapshotGeneration, error) {
55		-	return ports.SnapshotGeneration{}, nil
56		-	}
57		-	func (s refusingSnapshotDeleteRepository) Delete(context.Context, string) error { return s.err }
58		-	func (refusingSnapshotDeleteRepository) Tombstone(context.Context, string) error { return nil }
59		-	func (refusingSnapshotDeleteRepository) DeleteTombstone(context.Context, string) error {
60		-	return nil
	56	+	func (s refusingSnapshotDeleteRepository) QuarantineDeletionSources(context.Context, domain.DeletionTombstone, bool) error {
	57	+	return s.err
61	58		}
62		-	func (refusingSnapshotDeleteRepository) Maintain(context.Context) error { return nil }
63	59
64	60		func newTestTabWithContext(p ports.PTY, ctx context.Context, cancel context.CancelFunc) *tab {
65	61		tb := newTab(p, domain.Size{Cols: 80, Rows: 23})

 func (failingDeleteStore) Sync() error                  { return nil }
 func (failingDeleteStore) Close() error                 { return nil }
-type refusingSnapshotDeleteRepository struct{ err error }
+type refusingSnapshotDeleteRepository struct {
+	noOpSnapshotRepository
+	err error
+}
 func (refusingSnapshotDeleteRepository) Publish(context.Context, ports.SnapshotPublication) error {
 	return nil
 }
-func (refusingSnapshotDeleteRepository) List(context.Context) ([]string, error) { return nil, nil }
+func (s refusingSnapshotDeleteRepository) QuarantineDeletionSources(context.Context, domain.DeletionTombstone, bool) error {
-func (refusingSnapshotDeleteRepository) Load(context.Context, string) (ports.SnapshotGeneration, error) {
+	return s.err
-	return ports.SnapshotGeneration{}, nil
-}
-func (s refusingSnapshotDeleteRepository) Delete(context.Context, string) error  { return s.err }
-func (refusingSnapshotDeleteRepository) Tombstone(context.Context, string) error { return nil }
-func (refusingSnapshotDeleteRepository) DeleteTombstone(context.Context, string) error {
-	return nil
 }
-func (refusingSnapshotDeleteRepository) Maintain(context.Context) error { return nil }
 func newTestTabWithContext(p ports.PTY, ctx context.Context, cancel context.CancelFunc) *tab {
 	tb := newTab(p, domain.Size{Cols: 80, Rows: 23})

M internal/usecase/daemon/session_purge_retry_test.go +21 -23


@@ -13,6 +13,7 @@ import (
13	13		)
14	14
15	15		type retryablePurgeRepository struct {
	16	+	noOpSnapshotRepository
16	17		incrementalErr error
17	18		legacyErr error
18	19		tombstoned map[string]bool

@@ -22,38 +23,35 @@ type retryablePurgeRepository struct {
22	23		func (r *retryablePurgeRepository) Publish(context.Context, ports.SnapshotPublication) error {
23	24		return nil
24	25		}
25		-	func (r *retryablePurgeRepository) List(context.Context) ([]string, error) { return nil, nil }
26		-	func (r *retryablePurgeRepository) Load(context.Context, string) (ports.SnapshotGeneration, error) {
27		-	return ports.SnapshotGeneration{}, errors.New("unused")
28		-	}
29		-	func (r *retryablePurgeRepository) Maintain(context.Context) error { return nil }
30		-	func (r *retryablePurgeRepository) Delete(_ context.Context, name string) error {
31		-	r.calls = append(r.calls, "incremental")
32		-	return r.incrementalErr
33		-	}
34		-	func (r *retryablePurgeRepository) LoadLegacy(context.Context) ([]ports.LegacySnapshot, error) {
35		-	return nil, nil
36		-	}
37		-	func (r *retryablePurgeRepository) DeleteVerifiedLegacy(ctx context.Context, blob ports.LegacySnapshot) error {
38		-	return r.DeleteLegacy(ctx, blob.Name)
39		-	}
40		-	func (r *retryablePurgeRepository) DeleteLegacy(_ context.Context, name string) error {
41		-	r.calls = append(r.calls, "legacy")
42		-	return r.legacyErr
43		-	}
44		-	func (r *retryablePurgeRepository) Tombstone(_ context.Context, name string) error {
	26	+	func (r *retryablePurgeRepository) WriteDeletionTombstone(_ context.Context, tombstone domain.DeletionTombstone) error {
45	27		if r.tombstoned == nil {
46	28		r.tombstoned = make(map[string]bool)
47	29		}
48	30		r.calls = append(r.calls, "tombstone")
49		-	r.tombstoned[name] = true
	31	+	r.tombstoned[tombstone.Name] = true
50	32		return nil
51	33		}
52		-	func (r *retryablePurgeRepository) DeleteTombstone(_ context.Context, name string) error {
	34	+	func (r *retryablePurgeRepository) QuarantineDeletionSources(_ context.Context, tombstone domain.DeletionTombstone, _ bool) error {
	35	+	r.calls = append(r.calls, "incremental", "legacy")
	36	+	if r.incrementalErr != nil {
	37	+	return r.incrementalErr
	38	+	}
	39	+	return r.legacyErr
	40	+	}
	41	+	func (r *retryablePurgeRepository) DeleteDeletionTombstone(_ context.Context, _ domain.IncarnationID) error {
53	42		r.calls = append(r.calls, "clear tombstone")
54		-	delete(r.tombstoned, name)
	43	+	for name := range r.tombstoned {
	44	+	delete(r.tombstoned, name)
	45	+	}
	46	+	return nil
	47	+	}
	48	+	func (*retryablePurgeRepository) LoadLegacy(context.Context) ([]ports.LegacySnapshot, error) {
	49	+	return nil, nil
	50	+	}
	51	+	func (*retryablePurgeRepository) DeleteVerifiedLegacy(context.Context, ports.LegacySnapshot) error {
55	52		return nil
56	53		}
	54	+	func (*retryablePurgeRepository) DeleteLegacy(context.Context, string) error { return nil }
57	55
58	56		func TestLivePurgeRetainsTombstoneAcrossPartialSourceDeletion(t *testing.T) {
59	57		for _, tt := range []struct {

 )
 type retryablePurgeRepository struct {
+	noOpSnapshotRepository
 	incrementalErr error
 	legacyErr      error
 	tombstoned     map[string]bool
 func (r *retryablePurgeRepository) Publish(context.Context, ports.SnapshotPublication) error {
 	return nil
 }
-func (r *retryablePurgeRepository) List(context.Context) ([]string, error) { return nil, nil }
+func (r *retryablePurgeRepository) WriteDeletionTombstone(_ context.Context, tombstone domain.DeletionTombstone) error {
-func (r *retryablePurgeRepository) Load(context.Context, string) (ports.SnapshotGeneration, error) {
-	return ports.SnapshotGeneration{}, errors.New("unused")
-}
-func (r *retryablePurgeRepository) Maintain(context.Context) error { return nil }
-func (r *retryablePurgeRepository) Delete(_ context.Context, name string) error {
-	r.calls = append(r.calls, "incremental")
-	return r.incrementalErr
-}
-func (r *retryablePurgeRepository) LoadLegacy(context.Context) ([]ports.LegacySnapshot, error) {
-	return nil, nil
-}
-func (r *retryablePurgeRepository) DeleteVerifiedLegacy(ctx context.Context, blob ports.LegacySnapshot) error {
-	return r.DeleteLegacy(ctx, blob.Name)
-}
-func (r *retryablePurgeRepository) DeleteLegacy(_ context.Context, name string) error {
-	r.calls = append(r.calls, "legacy")
-	return r.legacyErr
-}
-func (r *retryablePurgeRepository) Tombstone(_ context.Context, name string) error {
 	if r.tombstoned == nil {
 		r.tombstoned = make(map[string]bool)
 	}
 	r.calls = append(r.calls, "tombstone")
-	r.tombstoned[name] = true
+	r.tombstoned[tombstone.Name] = true
 	return nil
 }
-func (r *retryablePurgeRepository) DeleteTombstone(_ context.Context, name string) error {
+func (r *retryablePurgeRepository) QuarantineDeletionSources(_ context.Context, tombstone domain.DeletionTombstone, _ bool) error {
+	r.calls = append(r.calls, "incremental", "legacy")
+	if r.incrementalErr != nil {
+		return r.incrementalErr
+	}
+	return r.legacyErr
+}
+func (r *retryablePurgeRepository) DeleteDeletionTombstone(_ context.Context, _ domain.IncarnationID) error {
 	r.calls = append(r.calls, "clear tombstone")
-	delete(r.tombstoned, name)
+	for name := range r.tombstoned {
+		delete(r.tombstoned, name)
+	}
+	return nil
+}
+func (*retryablePurgeRepository) LoadLegacy(context.Context) ([]ports.LegacySnapshot, error) {
+	return nil, nil
+}
+func (*retryablePurgeRepository) DeleteVerifiedLegacy(context.Context, ports.LegacySnapshot) error {
 	return nil
 }
+func (*retryablePurgeRepository) DeleteLegacy(context.Context, string) error { return nil }
 func TestLivePurgeRetainsTombstoneAcrossPartialSourceDeletion(t *testing.T) {
 	for _, tt := range []struct {

M internal/usecase/daemon/snapshot_import_test.go +6 -58


@@ -101,36 +101,20 @@ func TestLegacyPublicationMarshalObjectErrorsIdentifyPaneAndStage(t *testing.T)
101	101		}
102	102		}
103	103
104		-	func TestImportLegacyPublishesCompleteGenerationBeforeDeletingSource(t *testing.T) {
	104	+	func TestImportLegacySnapshotsIsRetiredRuntimeNoop(t *testing.T) {
105	105		snapshot := legacyAcceptanceSnapshot(t)
106	106		repository := &snapshotAcceptanceRepository{generations: make(map[string]ports.SnapshotGeneration)}
107	107		legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{legacyAcceptanceBlob(t, snapshot)}}
108	108		d := newTestDaemon(t, nil, stubClock{})
109	109		WithSnapshotRepository(repository, legacy)(d)
110		-	expected, verifyErr := legacyPublication(snapshot)
111		-	require.NoError(t, verifyErr)
112		-	_, verifyErr = snapcodec.Unmarshal(legacy.blobs[0].Data)
113		-	require.NoError(t, verifyErr)
114		-	require.NoError(t, verifyLegacyImportGeneration(acceptanceGeneration(t, snapshot, 1), expected))
115	110
116	111		d.importLegacySnapshots(context.Background())
117	112
118		-	require.Len(t, repository.publishes, 1)
119		-	publication := repository.publishes[0]
120		-	require.Equal(t, snapshot.Name, publication.Name)
121		-	require.EqualValues(t, 1, publication.Generation)
122		-	manifest, err := snapcodec.UnmarshalManifest(publication.Manifest)
123		-	require.NoError(t, err)
124		-	require.Equal(t, uint64(1), manifest.Generation)
125		-	require.Empty(t, manifest.Tabs)
126		-	for _, object := range publication.Objects {
127		-	kind, _, err := snapcodec.UnmarshalObject(object.Data)
128		-	require.NoError(t, err)
129		-	require.Contains(t, []snapcodec.ObjectKind{snapcodec.HistoryChunk, snapcodec.HistoryTail, snapcodec.Visible}, kind)
130		-	require.Equal(t, object.Digest, snapcodec.ManifestDigest(object.Data))
131		-	}
132		-	require.Equal(t, []string{snapshot.Name}, legacy.deleteCalls)
133		-	require.Empty(t, legacy.blobs)
	113	+	// Migration is adapter-owned. The daemon must never name-scan, publish, or
	114	+	// delete a legacy source during normal recovery.
	115	+	require.Empty(t, repository.publishes)
	116	+	require.Empty(t, legacy.deleteCalls)
	117	+	require.Len(t, legacy.blobs, 1)
134	118		}
135	119
136	120		func TestImportLegacyExistingIncrementalSkipsLegacy(t *testing.T) {

@@ -177,39 +161,3 @@ func TestImportLegacyRetainsSourceUntilExactReloadVerification(t *testing.T) {
177	161		})
178	162		}
179	163		}
180		-
181		-	func TestImportLegacyDeleteFailureRetriesWithoutRepublishing(t *testing.T) {
182		-	snapshot := legacyAcceptanceSnapshot(t)
183		-	repository := &snapshotAcceptanceRepository{generations: make(map[string]ports.SnapshotGeneration)}
184		-	legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{legacyAcceptanceBlob(t, snapshot)}, deleteErr: errors.New("sync failed")}
185		-	d := newTestDaemon(t, nil, stubClock{})
186		-	WithSnapshotRepository(repository, legacy)(d)
187		-
188		-	d.importLegacySnapshots(context.Background())
189		-	legacy.deleteErr = nil
190		-	d.importLegacySnapshots(context.Background())
191		-
192		-	require.Len(t, repository.publishes, 1, "a retry must only retry source deletion")
193		-	require.Equal(t, []string{snapshot.Name, snapshot.Name}, legacy.deleteCalls)
194		-	require.Empty(t, legacy.blobs)
195		-	}
196		-
197		-	func TestImportLegacyCancellationAndPerSessionContinuation(t *testing.T) {
198		-	bad := legacyAcceptanceSnapshot(t)
199		-	good := legacyAcceptanceSnapshot(t)
200		-	badBlob := legacyAcceptanceBlob(t, bad)
201		-	badBlob.Data = []byte("not-a-v3-snapshot")
202		-	repository := &snapshotAcceptanceRepository{generations: make(map[string]ports.SnapshotGeneration)}
203		-	legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{badBlob, legacyAcceptanceBlob(t, good)}}
204		-	d := newTestDaemon(t, nil, stubClock{})
205		-	WithSnapshotRepository(repository, legacy)(d)
206		-
207		-	d.importLegacySnapshots(context.Background())
208		-	require.Len(t, repository.publishes, 1, "a corrupt session must not block later sessions")
209		-	require.Equal(t, []string{good.Name}, legacy.deleteCalls)
210		-
211		-	cancelled, cancel := context.WithCancel(context.Background())
212		-	cancel()
213		-	d.importLegacySnapshots(cancelled)
214		-	require.Equal(t, 1, legacy.loadCalls, "a canceled import must not touch the legacy source")
215		-	}


@@ -101,36 +101,20 @@ func TestLegacyPublicationMarshalObjectErrorsIdentifyPaneAndStage(t *testing.T)
101		}	101		}
102		}	102		}
103			103
104	-	func TestImportLegacyPublishesCompleteGenerationBeforeDeletingSource(t *testing.T) {	104	+	func TestImportLegacySnapshotsIsRetiredRuntimeNoop(t *testing.T) {
105		snapshot := legacyAcceptanceSnapshot(t)	105		snapshot := legacyAcceptanceSnapshot(t)
106		repository := &snapshotAcceptanceRepository{generations: make(map[string]ports.SnapshotGeneration)}	106		repository := &snapshotAcceptanceRepository{generations: make(map[string]ports.SnapshotGeneration)}
107		legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{legacyAcceptanceBlob(t, snapshot)}}	107		legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{legacyAcceptanceBlob(t, snapshot)}}
108		d := newTestDaemon(t, nil, stubClock{})	108		d := newTestDaemon(t, nil, stubClock{})
109		WithSnapshotRepository(repository, legacy)(d)	109		WithSnapshotRepository(repository, legacy)(d)
110	-	expected, verifyErr := legacyPublication(snapshot)
111	-	require.NoError(t, verifyErr)
112	-	_, verifyErr = snapcodec.Unmarshal(legacy.blobs[0].Data)
113	-	require.NoError(t, verifyErr)
114	-	require.NoError(t, verifyLegacyImportGeneration(acceptanceGeneration(t, snapshot, 1), expected))
115			110
116		d.importLegacySnapshots(context.Background())	111		d.importLegacySnapshots(context.Background())
117			112
118	-	require.Len(t, repository.publishes, 1)	113	+	// Migration is adapter-owned. The daemon must never name-scan, publish, or
119	-	publication := repository.publishes[0]	114	+	// delete a legacy source during normal recovery.
120	-	require.Equal(t, snapshot.Name, publication.Name)	115	+	require.Empty(t, repository.publishes)
121	-	require.EqualValues(t, 1, publication.Generation)	116	+	require.Empty(t, legacy.deleteCalls)
122	-	manifest, err := snapcodec.UnmarshalManifest(publication.Manifest)	117	+	require.Len(t, legacy.blobs, 1)
123	-	require.NoError(t, err)
124	-	require.Equal(t, uint64(1), manifest.Generation)
125	-	require.Empty(t, manifest.Tabs)
126	-	for _, object := range publication.Objects {
127	-	kind, _, err := snapcodec.UnmarshalObject(object.Data)
128	-	require.NoError(t, err)
129	-	require.Contains(t, []snapcodec.ObjectKind{snapcodec.HistoryChunk, snapcodec.HistoryTail, snapcodec.Visible}, kind)
130	-	require.Equal(t, object.Digest, snapcodec.ManifestDigest(object.Data))
131	-	}
132	-	require.Equal(t, []string{snapshot.Name}, legacy.deleteCalls)
133	-	require.Empty(t, legacy.blobs)
134		}	118		}
135			119
136		func TestImportLegacyExistingIncrementalSkipsLegacy(t *testing.T) {	120		func TestImportLegacyExistingIncrementalSkipsLegacy(t *testing.T) {

@@ -177,39 +161,3 @@ func TestImportLegacyRetainsSourceUntilExactReloadVerification(t *testing.T) {
177		})	161		})
178		}	162		}
179		}	163		}
180	-
181	-	func TestImportLegacyDeleteFailureRetriesWithoutRepublishing(t *testing.T) {
182	-	snapshot := legacyAcceptanceSnapshot(t)
183	-	repository := &snapshotAcceptanceRepository{generations: make(map[string]ports.SnapshotGeneration)}
184	-	legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{legacyAcceptanceBlob(t, snapshot)}, deleteErr: errors.New("sync failed")}
185	-	d := newTestDaemon(t, nil, stubClock{})
186	-	WithSnapshotRepository(repository, legacy)(d)
187	-
188	-	d.importLegacySnapshots(context.Background())
189	-	legacy.deleteErr = nil
190	-	d.importLegacySnapshots(context.Background())
191	-
192	-	require.Len(t, repository.publishes, 1, "a retry must only retry source deletion")
193	-	require.Equal(t, []string{snapshot.Name, snapshot.Name}, legacy.deleteCalls)
194	-	require.Empty(t, legacy.blobs)
195	-	}
196	-
197	-	func TestImportLegacyCancellationAndPerSessionContinuation(t *testing.T) {
198	-	bad := legacyAcceptanceSnapshot(t)
199	-	good := legacyAcceptanceSnapshot(t)
200	-	badBlob := legacyAcceptanceBlob(t, bad)
201	-	badBlob.Data = []byte("not-a-v3-snapshot")
202	-	repository := &snapshotAcceptanceRepository{generations: make(map[string]ports.SnapshotGeneration)}
203	-	legacy := &snapshotAcceptanceLegacySource{blobs: []ports.LegacySnapshot{badBlob, legacyAcceptanceBlob(t, good)}}
204	-	d := newTestDaemon(t, nil, stubClock{})
205	-	WithSnapshotRepository(repository, legacy)(d)
206	-
207	-	d.importLegacySnapshots(context.Background())
208	-	require.Len(t, repository.publishes, 1, "a corrupt session must not block later sessions")
209	-	require.Equal(t, []string{good.Name}, legacy.deleteCalls)
210	-
211	-	cancelled, cancel := context.WithCancel(context.Background())
212	-	cancel()
213	-	d.importLegacySnapshots(cancelled)
214	-	require.Equal(t, 1, legacy.loadCalls, "a canceled import must not touch the legacy source")
215	-	}

M internal/usecase/daemon/snapshot_legacy_import.go +6 -2


@@ -3,12 +3,16 @@ package daemon
3	3		import (
4	4		"fmt"
5	5
	6	+	"github.com/bnema/vev/internal/domain"
6	7		"github.com/bnema/vev/internal/ports"
7	8		snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
8	9		)
9	10
10	11		func legacyPublication(snapshot snapcodec.Session) (ports.SnapshotPublication, error) {
11		-	manifest := snapcodec.Manifest{Generation: 1, Name: snapshot.Name, CreatedAt: snapshot.CreatedAt, Active: snapshot.Active, Tabs: make([]snapcodec.ManifestTab, 0, len(snapshot.Tabs))}
	12	+	// Legacy payloads have no durable identity; this test-only conversion uses
	13	+	// the fixture's stable synthetic identity.
	14	+	incarnation := domain.IncarnationID{1}
	15	+	manifest := snapcodec.Manifest{Generation: 1, IncarnationID: incarnation, Name: snapshot.Name, CreatedAt: snapshot.CreatedAt, Active: snapshot.Active, Tabs: make([]snapcodec.ManifestTab, 0, len(snapshot.Tabs))}
12	16		objects := make([]ports.SnapshotObject, 0)
13	17		for _, tab := range snapshot.Tabs {
14	18		outTab := snapcodec.ManifestTab{StableID: tab.StableID, Cols: tab.Cols, Rows: tab.Rows, NextPaneID: tab.NextPaneID, Focus: tab.Focus, Tree: tab.Tree, Panes: make([]snapcodec.ManifestPane, 0, len(tab.Panes))}

@@ -40,5 +44,5 @@ func legacyPublication(snapshot snapcodec.Session) (ports.SnapshotPublication, e
40	44		if err != nil {
41	45		return ports.SnapshotPublication{}, fmt.Errorf("snapshot: marshal legacy import manifest: %w", err)
42	46		}
43		-	return ports.SnapshotPublication{Name: snapshot.Name, Generation: 1, Manifest: encoded, Objects: objects}, nil
	47	+	return ports.SnapshotPublication{IncarnationID: incarnation, Name: snapshot.Name, Generation: 1, Manifest: encoded, Objects: objects}, nil
44	48		}

 import (
 	"fmt"
+	"github.com/bnema/vev/internal/domain"
 	"github.com/bnema/vev/internal/ports"
 	snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
 )
 func legacyPublication(snapshot snapcodec.Session) (ports.SnapshotPublication, error) {
-	manifest := snapcodec.Manifest{Generation: 1, Name: snapshot.Name, CreatedAt: snapshot.CreatedAt, Active: snapshot.Active, Tabs: make([]snapcodec.ManifestTab, 0, len(snapshot.Tabs))}
+	// Legacy payloads have no durable identity; this test-only conversion uses
+	// the fixture's stable synthetic identity.
+	incarnation := domain.IncarnationID{1}
+	manifest := snapcodec.Manifest{Generation: 1, IncarnationID: incarnation, Name: snapshot.Name, CreatedAt: snapshot.CreatedAt, Active: snapshot.Active, Tabs: make([]snapcodec.ManifestTab, 0, len(snapshot.Tabs))}
 	objects := make([]ports.SnapshotObject, 0)
 	for _, tab := range snapshot.Tabs {
 		outTab := snapcodec.ManifestTab{StableID: tab.StableID, Cols: tab.Cols, Rows: tab.Rows, NextPaneID: tab.NextPaneID, Focus: tab.Focus, Tree: tab.Tree, Panes: make([]snapcodec.ManifestPane, 0, len(tab.Panes))}
 	if err != nil {
 		return ports.SnapshotPublication{}, fmt.Errorf("snapshot: marshal legacy import manifest: %w", err)
 	}
-	return ports.SnapshotPublication{Name: snapshot.Name, Generation: 1, Manifest: encoded, Objects: objects}, nil
+	return ports.SnapshotPublication{IncarnationID: incarnation, Name: snapshot.Name, Generation: 1, Manifest: encoded, Objects: objects}, nil
 }

M internal/usecase/daemon/snapshot_restore_test.go +17 -16


@@ -18,6 +18,7 @@ import (
18	18		)
19	19
20	20		type snapshotAcceptanceRepository struct {
	21	+	noOpSnapshotRepository
21	22		mu sync.Mutex
22	23		names []string
23	24		generations map[string]ports.SnapshotGeneration

@@ -42,7 +43,7 @@ func (r *snapshotAcceptanceRepository) Publish(ctx context.Context, p ports.Snap
42	43		for _, object := range p.Objects {
43	44		objects[object.Digest] = append([]byte(nil), object.Data...)
44	45		}
45		-	r.generations[p.Name] = ports.SnapshotGeneration{Name: p.Name, Generation: p.Generation, Manifest: append([]byte(nil), p.Manifest...), Objects: objects}
	46	+	r.generations[p.Name] = ports.SnapshotGeneration{IncarnationID: p.IncarnationID, Name: p.Name, Generation: p.Generation, ParentCheckpoint: p.ParentCheckpoint, Manifest: append([]byte(nil), p.Manifest...), Objects: objects}
46	47		found := false
47	48		for _, name := range r.names {
48	49		found = found \|\| name == p.Name

@@ -85,10 +86,16 @@ func (r *snapshotAcceptanceRepository) Load(ctx context.Context, name string) (p
85	86		return cloneAcceptanceGeneration(generation), nil
86	87		}
87	88
88		-	func (*snapshotAcceptanceRepository) Delete(context.Context, string) error { return nil }
89		-	func (*snapshotAcceptanceRepository) Tombstone(context.Context, string) error { return nil }
90		-	func (*snapshotAcceptanceRepository) DeleteTombstone(context.Context, string) error { return nil }
91		-	func (*snapshotAcceptanceRepository) Maintain(context.Context) error { return nil }
	89	+	func (r *snapshotAcceptanceRepository) LoadCheckpoint(ctx context.Context, id domain.IncarnationID, name string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {
	90	+	generation, err := r.Load(ctx, name)
	91	+	if err != nil {
	92	+	return ports.SnapshotGeneration{}, err
	93	+	}
	94	+	if generation.IncarnationID != id \|\| generation.Generation != ref.Generation \|\| snapcodec.ManifestDigest(generation.Manifest) != ref.ManifestDigest {
	95	+	return ports.SnapshotGeneration{}, errors.New("checkpoint unavailable")
	96	+	}
	97	+	return generation, nil
	98	+	}
92	99
93	100		func TestSnapshotAcceptanceRepositoryLoadMutationDoesNotChangeStoredGeneration(t *testing.T) {
94	101		digest := ports.SnapshotDigest{1}

@@ -178,7 +185,7 @@ func acceptanceGeneration(t *testing.T, snapshot snapcodec.Session, generation u
178	185		for _, object := range publication.Objects {
179	186		objects[object.Digest] = append([]byte(nil), object.Data...)
180	187		}
181		-	return ports.SnapshotGeneration{Name: snapshot.Name, Generation: generation, Manifest: encoded, Objects: objects}
	188	+	return ports.SnapshotGeneration{IncarnationID: publication.IncarnationID, Name: snapshot.Name, Generation: generation, ParentCheckpoint: publication.ParentCheckpoint, Manifest: encoded, Objects: objects}
182	189		}
183	190
184	191		func TestValidateRestoreSessionSnapshot(t *testing.T) {

@@ -212,6 +219,10 @@ func TestRestoreIncrementalGenerationAcceptance(t *testing.T) {
212	219		d := newTestDaemon(t, newFactory(t, pty), stubClock{})
213	220		store, _ := newMockStore(t)
214	221		WithStore(store)(d)
	222	+	checkpoint := domain.CheckpointRef{Generation: generation.Generation, ManifestDigest: snapcodec.ManifestDigest(generation.Manifest)}
	223	+	record := domain.CatalogueRecord{Name: snapshot.Name, IncarnationID: generation.IncarnationID, Cwd: "/snapshot/cwd", CreatedAt: int64(snapshot.CreatedAt), RecoveryState: domain.RecoveryHealthy, Committed: &checkpoint}
	224	+	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
	225	+	WithCatalogue(catalogue, []domain.CatalogueRecord{record})(d)
215	226		WithSnapshotRepository(repository, nil)(d)
216	227		t.Cleanup(func() { release(); d.sessWg.Wait() })
217	228

@@ -249,16 +260,6 @@ func TestRestoreIncrementalGenerationAcceptance(t *testing.T) {
249	260		func TestRestoreIncrementalFallbackAndInvalidObjectMappings(t *testing.T) {
250	261		snapshot := restoreAcceptanceSession(t, "fallback")
251	262		valid := acceptanceGeneration(t, snapshot, 3)
252		-	pty, release := newBlockingPTY(t)
253		-	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
254		-	store, _ := newMockStore(t)
255		-	WithStore(store)(d)
256		-	repository := &snapshotAcceptanceRepository{names: []string{snapshot.Name}, generations: map[string]ports.SnapshotGeneration{snapshot.Name: valid}}
257		-	WithSnapshotRepository(repository, nil)(d)
258		-	t.Cleanup(func() { release(); d.sessWg.Wait() })
259		-	d.restoreIncrementalSnapshots(context.Background())
260		-	require.Len(t, d.notices.history(), 1)
261		-	require.Equal(t, domain.NoticeWarn, d.notices.history()[0].Severity)
262	263
263	264		for _, mutate := range []struct {
264	265		name string

 )
 type snapshotAcceptanceRepository struct {
+	noOpSnapshotRepository
 	mu          sync.Mutex
 	names       []string
 	generations map[string]ports.SnapshotGeneration
 	for _, object := range p.Objects {
 		objects[object.Digest] = append([]byte(nil), object.Data...)
 	}
-	r.generations[p.Name] = ports.SnapshotGeneration{Name: p.Name, Generation: p.Generation, Manifest: append([]byte(nil), p.Manifest...), Objects: objects}
+	r.generations[p.Name] = ports.SnapshotGeneration{IncarnationID: p.IncarnationID, Name: p.Name, Generation: p.Generation, ParentCheckpoint: p.ParentCheckpoint, Manifest: append([]byte(nil), p.Manifest...), Objects: objects}
 	found := false
 	for _, name := range r.names {
 		found = found || name == p.Name
 	return cloneAcceptanceGeneration(generation), nil
 }
-func (*snapshotAcceptanceRepository) Delete(context.Context, string) error          { return nil }
+func (r *snapshotAcceptanceRepository) LoadCheckpoint(ctx context.Context, id domain.IncarnationID, name string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {
-func (*snapshotAcceptanceRepository) Tombstone(context.Context, string) error       { return nil }
+	generation, err := r.Load(ctx, name)
-func (*snapshotAcceptanceRepository) DeleteTombstone(context.Context, string) error { return nil }
+	if err != nil {
-func (*snapshotAcceptanceRepository) Maintain(context.Context) error                { return nil }
+		return ports.SnapshotGeneration{}, err
+	}
+	if generation.IncarnationID != id || generation.Generation != ref.Generation || snapcodec.ManifestDigest(generation.Manifest) != ref.ManifestDigest {
+		return ports.SnapshotGeneration{}, errors.New("checkpoint unavailable")
+	}
+	return generation, nil
+}
 func TestSnapshotAcceptanceRepositoryLoadMutationDoesNotChangeStoredGeneration(t *testing.T) {
 	digest := ports.SnapshotDigest{1}
 	for _, object := range publication.Objects {
 		objects[object.Digest] = append([]byte(nil), object.Data...)
 	}
-	return ports.SnapshotGeneration{Name: snapshot.Name, Generation: generation, Manifest: encoded, Objects: objects}
+	return ports.SnapshotGeneration{IncarnationID: publication.IncarnationID, Name: snapshot.Name, Generation: generation, ParentCheckpoint: publication.ParentCheckpoint, Manifest: encoded, Objects: objects}
 }
 func TestValidateRestoreSessionSnapshot(t *testing.T) {
 	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
 	store, _ := newMockStore(t)
 	WithStore(store)(d)
+	checkpoint := domain.CheckpointRef{Generation: generation.Generation, ManifestDigest: snapcodec.ManifestDigest(generation.Manifest)}
+	record := domain.CatalogueRecord{Name: snapshot.Name, IncarnationID: generation.IncarnationID, Cwd: "/snapshot/cwd", CreatedAt: int64(snapshot.CreatedAt), RecoveryState: domain.RecoveryHealthy, Committed: &checkpoint}
+	catalogue := newDurableRecoveryCatalogue([]domain.CatalogueRecord{record})
+	WithCatalogue(catalogue, []domain.CatalogueRecord{record})(d)
 	WithSnapshotRepository(repository, nil)(d)
 	t.Cleanup(func() { release(); d.sessWg.Wait() })
 func TestRestoreIncrementalFallbackAndInvalidObjectMappings(t *testing.T) {
 	snapshot := restoreAcceptanceSession(t, "fallback")
 	valid := acceptanceGeneration(t, snapshot, 3)
-	pty, release := newBlockingPTY(t)
-	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
-	store, _ := newMockStore(t)
-	WithStore(store)(d)
-	repository := &snapshotAcceptanceRepository{names: []string{snapshot.Name}, generations: map[string]ports.SnapshotGeneration{snapshot.Name: valid}}
-	WithSnapshotRepository(repository, nil)(d)
-	t.Cleanup(func() { release(); d.sessWg.Wait() })
-	d.restoreIncrementalSnapshots(context.Background())
-	require.Len(t, d.notices.history(), 1)
-	require.Equal(t, domain.NoticeWarn, d.notices.history()[0].Severity)
 	for _, mutate := range []struct {
 		name string

M internal/usecase/daemon/snapshot_test.go +38 -0


@@ -53,6 +53,44 @@ func newSnapshotTestSession(t *testing.T, name string, ephemeral bool, cwd strin
53	53		return sess
54	54		}
55	55
	56	+	// noOpSnapshotRepository supplies the current durable repository contract to
	57	+	// focused test sinks that only need to observe publication or deletion calls.
	58	+	type noOpSnapshotRepository struct{}
	59	+
	60	+	func (noOpSnapshotRepository) LoadCheckpoint(context.Context, domain.IncarnationID, string, ports.CheckpointRef) (ports.SnapshotGeneration, error) {
	61	+	return ports.SnapshotGeneration{}, errors.New("unused")
	62	+	}
	63	+	func (noOpSnapshotRepository) RepairHEAD(context.Context, domain.IncarnationID, ports.CheckpointRef) error {
	64	+	return nil
	65	+	}
	66	+	func (noOpSnapshotRepository) WriteDeletionTombstone(context.Context, domain.DeletionTombstone) error {
	67	+	return nil
	68	+	}
	69	+	func (noOpSnapshotRepository) ListDeletionTombstones(context.Context, ports.DeletionTombstoneCursor, ports.MaintenanceBudget) (ports.DeletionTombstonePage, error) {
	70	+	return ports.DeletionTombstonePage{Done: true}, nil
	71	+	}
	72	+	func (noOpSnapshotRepository) QuarantineDeletionSources(context.Context, domain.DeletionTombstone, bool) error {
	73	+	return nil
	74	+	}
	75	+	func (noOpSnapshotRepository) DeleteDeletionTombstone(context.Context, domain.IncarnationID) error {
	76	+	return nil
	77	+	}
	78	+	func (noOpSnapshotRepository) SaveQuarantineDescriptor(context.Context, domain.QuarantineDescriptor) error {
	79	+	return nil
	80	+	}
	81	+	func (noOpSnapshotRepository) QuarantineIncarnation(context.Context, domain.IncarnationID) error {
	82	+	return nil
	83	+	}
	84	+	func (noOpSnapshotRepository) DeleteIncarnation(context.Context, domain.IncarnationID) error {
	85	+	return nil
	86	+	}
	87	+	func (noOpSnapshotRepository) MaintainSession(context.Context, ports.RetentionPlan, ports.MaintenanceBudget) (bool, error) {
	88	+	return true, nil
	89	+	}
	90	+	func (noOpSnapshotRepository) Reconcile(context.Context, []domain.CatalogueRecord, ports.ReconcileCursor, ports.MaintenanceBudget) (ports.ReconcileCursor, []ports.ReconcileFinding, error) {
	91	+	return ports.ReconcileCursor{}, nil, nil
	92	+	}
	93	+
56	94		func awaitSnapshotIdle(t testing.TB, sess *session) {
57	95		t.Helper()
58	96		timer := time.NewTimer(testWaitTimeout)


@@ -53,6 +53,44 @@ func newSnapshotTestSession(t *testing.T, name string, ephemeral bool, cwd strin
53	return sess	53		return sess
54	}	54		}
55		55
		56	+	// noOpSnapshotRepository supplies the current durable repository contract to
		57	+	// focused test sinks that only need to observe publication or deletion calls.
		58	+	type noOpSnapshotRepository struct{}
		59	+
		60	+	func (noOpSnapshotRepository) LoadCheckpoint(context.Context, domain.IncarnationID, string, ports.CheckpointRef) (ports.SnapshotGeneration, error) {
		61	+	return ports.SnapshotGeneration{}, errors.New("unused")
		62	+	}
		63	+	func (noOpSnapshotRepository) RepairHEAD(context.Context, domain.IncarnationID, ports.CheckpointRef) error {
		64	+	return nil
		65	+	}
		66	+	func (noOpSnapshotRepository) WriteDeletionTombstone(context.Context, domain.DeletionTombstone) error {
		67	+	return nil
		68	+	}
		69	+	func (noOpSnapshotRepository) ListDeletionTombstones(context.Context, ports.DeletionTombstoneCursor, ports.MaintenanceBudget) (ports.DeletionTombstonePage, error) {
		70	+	return ports.DeletionTombstonePage{Done: true}, nil
		71	+	}
		72	+	func (noOpSnapshotRepository) QuarantineDeletionSources(context.Context, domain.DeletionTombstone, bool) error {
		73	+	return nil
		74	+	}
		75	+	func (noOpSnapshotRepository) DeleteDeletionTombstone(context.Context, domain.IncarnationID) error {
		76	+	return nil
		77	+	}
		78	+	func (noOpSnapshotRepository) SaveQuarantineDescriptor(context.Context, domain.QuarantineDescriptor) error {
		79	+	return nil
		80	+	}
		81	+	func (noOpSnapshotRepository) QuarantineIncarnation(context.Context, domain.IncarnationID) error {
		82	+	return nil
		83	+	}
		84	+	func (noOpSnapshotRepository) DeleteIncarnation(context.Context, domain.IncarnationID) error {
		85	+	return nil
		86	+	}
		87	+	func (noOpSnapshotRepository) MaintainSession(context.Context, ports.RetentionPlan, ports.MaintenanceBudget) (bool, error) {
		88	+	return true, nil
		89	+	}
		90	+	func (noOpSnapshotRepository) Reconcile(context.Context, []domain.CatalogueRecord, ports.ReconcileCursor, ports.MaintenanceBudget) (ports.ReconcileCursor, []ports.ReconcileFinding, error) {
		91	+	return ports.ReconcileCursor{}, nil, nil
		92	+	}
		93	+
56	func awaitSnapshotIdle(t testing.TB, sess *session) {	94		func awaitSnapshotIdle(t testing.TB, sess *session) {
57	t.Helper()	95		t.Helper()
58	timer := time.NewTimer(testWaitTimeout)	96		timer := time.NewTimer(testWaitTimeout)