feat(daemon): restore sessions from catalogue checkpoints

commit: 52e29a32c76ad7840fe6809bd92c26b8c7bc6a63
author: brice · 2mo ago · 2026-07-25 23:24 UTC
parent: 305b83a
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +553 -16

M internal/ports/recovery.go +4 -1


@@ -100,7 +100,10 @@ type ReconcileDecision struct {
100	100		ReasonCode string
101	101		}
102	102
103		-	var ErrLegacySnapshotUncertain = errors.New("legacy snapshot state is uncertain")
	103	+	var (
	104	+	ErrLegacySnapshotUncertain = errors.New("legacy snapshot state is uncertain")
	105	+	ErrBudgetExhausted = errors.New("snapshot maintenance budget exhausted")
	106	+	)
104	107
105	108		type SnapshotMigrationRequest struct {
106	109		LegacyName string

 	ReasonCode string
 }
-var ErrLegacySnapshotUncertain = errors.New("legacy snapshot state is uncertain")
+var (
+	ErrLegacySnapshotUncertain = errors.New("legacy snapshot state is uncertain")
+	ErrBudgetExhausted         = errors.New("snapshot maintenance budget exhausted")
+)
 type SnapshotMigrationRequest struct {
 	LegacyName    string

M internal/usecase/daemon/daemon.go +33 -5


@@ -119,6 +119,7 @@ type Daemon struct {
119	119		shellOverride bool
120	120		persist *persist.Persister
121	121		persistEnabled bool
	122	+	catalogue ports.Catalogue
122	123		catalogueRecords []domain.CatalogueRecord
123	124		catalogueRecordsProvided bool
124	125		// snapshotRepository is the sole checkpoint contract. legacySnapshots is

@@ -210,6 +211,16 @@ type parkedAttachment struct {
210	211
211	212		// session is a single multiplexed session. It owns one or more full-screen
212	213
	214	+	type runtimeRecoveryState uint8
	215	+
	216	+	const (
	217	+	runtimeFresh runtimeRecoveryState = iota + 1
	218	+	runtimeRestoring
	219	+	runtimeHealthy
	220	+	runtimeDegraded
	221	+	runtimeDeleting
	222	+	)
	223	+
213	224		type stoppedSession struct {
214	225		name string
215	226		cwd string

@@ -218,6 +229,9 @@ type stoppedSession struct {
218	229		lastUsedSeq uint64
219	230		tabNames []string
220	231		purging bool
	232	+	record domain.CatalogueRecord
	233	+	state runtimeRecoveryState
	234	+	restoreDone chan struct{}
221	235		}
222	236
223	237		type Option func(*Daemon)

@@ -251,9 +265,12 @@ func WithStore(store ports.Store) Option {
251	265
252	266		// WithCatalogue installs the singular catalogue and the strictly opened
253	267		// records that define the daemon's expected-session registry at startup.
254		-	func WithCatalogue(catalogue *persist.Persister, records []domain.CatalogueRecord) Option {
	268	+	func WithCatalogue(catalogue ports.Catalogue, records []domain.CatalogueRecord) Option {
255	269		return func(d *Daemon) {
256		-	d.persist = catalogue
	270	+	d.catalogue = catalogue
	271	+	if persister, ok := catalogue.(*persist.Persister); ok {
	272	+	d.persist = persister
	273	+	}
257	274		d.persistEnabled = catalogue != nil
258	275		d.catalogueRecords = append([]domain.CatalogueRecord(nil), records...)
259	276		d.catalogueRecordsProvided = true

@@ -449,7 +466,12 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt
449	466		var maxCreatedAt int64
450	467		hasCreatedAt := false
451	468		for _, r := range records {
452		-	d.stopped[r.Name] = stoppedSession{name: r.Name, cwd: r.Cwd, createdAt: r.CreatedAt, incarnation: r.IncarnationID, lastUsedSeq: r.LastUsedSeq, tabNames: append([]string(nil), r.TabNames...)}
	469	+	stopped := stoppedSession{name: r.Name, cwd: r.Cwd, createdAt: r.CreatedAt, incarnation: r.IncarnationID, lastUsedSeq: r.LastUsedSeq, tabNames: append([]string(nil), r.TabNames...)}
	470	+	if d.catalogueRecordsProvided {
	471	+	stopped.record = r
	472	+	stopped.state, stopped.restoreDone = initialRuntimeRecoveryState(r)
	473	+	}
	474	+	d.stopped[r.Name] = stopped
453	475		if !hasCreatedAt \|\| r.CreatedAt > maxCreatedAt {
454	476		maxCreatedAt = r.CreatedAt
455	477		hasCreatedAt = true

@@ -555,8 +577,14 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
555	577		d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
556	578		d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)
557	579		d.waitNotifies()
558		-	if err := d.persist.Close(); err != nil {
559		-	d.log.Warn("closing session persister failed", "err", err)
	580	+	var closeErr error
	581	+	if d.catalogue != nil {
	582	+	closeErr = d.catalogue.Close()
	583	+	} else {
	584	+	closeErr = d.persist.Close()
	585	+	}
	586	+	if closeErr != nil {
	587	+	d.log.Warn("closing session persister failed", "err", closeErr)
560	588		}
561	589		return nil
562	590		}

 	shellOverride                  bool
 	persist                        *persist.Persister
 	persistEnabled                 bool
+	catalogue                      ports.Catalogue
 	catalogueRecords               []domain.CatalogueRecord
 	catalogueRecordsProvided       bool
 	// snapshotRepository is the sole checkpoint contract. legacySnapshots is
 // session is a single multiplexed session. It owns one or more full-screen
+type runtimeRecoveryState uint8
+const (
+	runtimeFresh runtimeRecoveryState = iota + 1
+	runtimeRestoring
+	runtimeHealthy
+	runtimeDegraded
+	runtimeDeleting
+)
 type stoppedSession struct {
 	name        string
 	cwd         string
 	lastUsedSeq uint64
 	tabNames    []string
 	purging     bool
+	record      domain.CatalogueRecord
+	state       runtimeRecoveryState
+	restoreDone chan struct{}
 }
 type Option func(*Daemon)
 // WithCatalogue installs the singular catalogue and the strictly opened
 // records that define the daemon's expected-session registry at startup.
-func WithCatalogue(catalogue *persist.Persister, records []domain.CatalogueRecord) Option {
+func WithCatalogue(catalogue ports.Catalogue, records []domain.CatalogueRecord) Option {
 	return func(d *Daemon) {
-		d.persist = catalogue
+		d.catalogue = catalogue
+		if persister, ok := catalogue.(*persist.Persister); ok {
+			d.persist = persister
+		}
 		d.persistEnabled = catalogue != nil
 		d.catalogueRecords = append([]domain.CatalogueRecord(nil), records...)
 		d.catalogueRecordsProvided = true
 	var maxCreatedAt int64
 	hasCreatedAt := false
 	for _, r := range records {
-		d.stopped[r.Name] = stoppedSession{name: r.Name, cwd: r.Cwd, createdAt: r.CreatedAt, incarnation: r.IncarnationID, lastUsedSeq: r.LastUsedSeq, tabNames: append([]string(nil), r.TabNames...)}
+		stopped := stoppedSession{name: r.Name, cwd: r.Cwd, createdAt: r.CreatedAt, incarnation: r.IncarnationID, lastUsedSeq: r.LastUsedSeq, tabNames: append([]string(nil), r.TabNames...)}
+		if d.catalogueRecordsProvided {
+			stopped.record = r
+			stopped.state, stopped.restoreDone = initialRuntimeRecoveryState(r)
+		}
+		d.stopped[r.Name] = stopped
 		if !hasCreatedAt || r.CreatedAt > maxCreatedAt {
 			maxCreatedAt = r.CreatedAt
 			hasCreatedAt = true
 	d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
 	d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)
 	d.waitNotifies()
-	if err := d.persist.Close(); err != nil {
+	var closeErr error
-		d.log.Warn("closing session persister failed", "err", err)
+	if d.catalogue != nil {
+		closeErr = d.catalogue.Close()
+	} else {
+		closeErr = d.persist.Close()
+	}
+	if closeErr != nil {
+		d.log.Warn("closing session persister failed", "err", closeErr)
 	}
 	return nil
 }

A internal/usecase/daemon/durable_recovery_test.go +253 -0

@@ -0,0 +1,253 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"fmt"
	7	+	"io"
	8	+	"log/slog"
	9	+	"sync"
	10	+	"testing"
	11	+	"time"
	12	+
	13	+	"github.com/stretchr/testify/require"
	14	+
	15	+	"github.com/bnema/vev/internal/domain"
	16	+	"github.com/bnema/vev/internal/ports"
	17	+	snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
	18	+	)
	19	+
	20	+	type durableRecoveryCatalogue struct {
	21	+	ports.Catalogue
	22	+	mu sync.Mutex
	23	+	records map[string]domain.CatalogueRecord
	24	+	}
	25	+
	26	+	func newDurableRecoveryCatalogue(records []domain.CatalogueRecord) *durableRecoveryCatalogue {
	27	+	catalogue := &durableRecoveryCatalogue{records: make(map[string]domain.CatalogueRecord, len(records))}
	28	+	for _, record := range records {
	29	+	catalogue.records[record.Name] = record
	30	+	}
	31	+	return catalogue
	32	+	}
	33	+
	34	+	func (c *durableRecoveryCatalogue) Records() []domain.CatalogueRecord {
	35	+	c.mu.Lock()
	36	+	defer c.mu.Unlock()
	37	+	records := make([]domain.CatalogueRecord, 0, len(c.records))
	38	+	for _, record := range c.records {
	39	+	records = append(records, record)
	40	+	}
	41	+	return records
	42	+	}
	43	+
	44	+	func (c *durableRecoveryCatalogue) Record(name string) (domain.CatalogueRecord, bool) {
	45	+	c.mu.Lock()
	46	+	defer c.mu.Unlock()
	47	+	record, ok := c.records[name]
	48	+	return record, ok
	49	+	}
	50	+
	51	+	func (c *durableRecoveryCatalogue) Replace(name string, record domain.CatalogueRecord) error {
	52	+	c.mu.Lock()
	53	+	defer c.mu.Unlock()
	54	+	if name != record.Name {
	55	+	return errors.New("catalogue key mismatch")
	56	+	}
	57	+	c.records[name] = record
	58	+	return nil
	59	+	}
	60	+
	61	+	func (c *durableRecoveryCatalogue) Close() error { return nil }
	62	+
	63	+	type durableRecoveryRepository struct {
	64	+	ports.SnapshotRepository
	65	+	mu sync.Mutex
	66	+	generations map[string]ports.SnapshotGeneration
	67	+	errors map[string]error
	68	+	loads map[string]int
	69	+	repairs map[string]domain.CheckpointRef
	70	+	}
	71	+
	72	+	func (r *durableRecoveryRepository) LoadCheckpoint(_ context.Context, id domain.IncarnationID, name string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {
	73	+	r.mu.Lock()
	74	+	defer r.mu.Unlock()
	75	+	r.loads[name]++
	76	+	if err := r.errors[name]; err != nil {
	77	+	return ports.SnapshotGeneration{}, err
	78	+	}
	79	+	generation, ok := r.generations[name]
	80	+	if !ok \|\| generation.Generation != ref.Generation {
	81	+	return ports.SnapshotGeneration{}, errors.New("checkpoint unavailable")
	82	+	}
	83	+	return generation, nil
	84	+	}
	85	+
	86	+	func (r *durableRecoveryRepository) RepairHEAD(_ context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {
	87	+	r.mu.Lock()
	88	+	defer r.mu.Unlock()
	89	+	r.repairs[id.String()] = ref
	90	+	return nil
	91	+	}
	92	+
	93	+	func durableRecoveryRecord(index int) domain.CatalogueRecord {
	94	+	id := domain.IncarnationID{0: byte(index>>8 + 1), 1: byte(index + 1)}
	95	+	ref := domain.CheckpointRef{Generation: uint64(index + 1), ManifestDigest: [32]byte{byte(index%255 + 1)}}
	96	+	return domain.CatalogueRecord{
	97	+	Name: fmt.Sprintf("session-%04d", index),
	98	+	IncarnationID: id,
	99	+	Cwd: "/tmp",
	100	+	CreatedAt: int64(index + 1),
	101	+	UpdatedAt: int64(index + 1),
	102	+	RecoveryState: domain.RecoveryHealthy,
	103	+	Committed: &ref,
	104	+	}
	105	+	}
	106	+
	107	+	func validGeneration(t testing.TB, record domain.CatalogueRecord) ports.SnapshotGeneration {
	108	+	t.Helper()
	109	+	manifest, err := snapcodec.MarshalManifest(snapcodec.Manifest{
	110	+	Generation: record.Committed.Generation,
	111	+	IncarnationID: record.IncarnationID,
	112	+	Name: record.Name,
	113	+	CreatedAt: uint64(record.CreatedAt),
	114	+	})
	115	+	require.NoError(t, err)
	116	+	return ports.SnapshotGeneration{
	117	+	IncarnationID: record.IncarnationID,
	118	+	Name: record.Name,
	119	+	Generation: record.Committed.Generation,
	120	+	Manifest: manifest,
	121	+	Objects: map[ports.SnapshotDigest][]byte{},
	122	+	}
	123	+	}
	124	+
	125	+	func newDurableRecoveryDaemon(t testing.TB, records []domain.CatalogueRecord, repository ports.SnapshotRepository) (Daemon, durableRecoveryCatalogue) {
	126	+	t.Helper()
	127	+	catalogue := newDurableRecoveryCatalogue(records)
	128	+	d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(catalogue, records), WithSnapshotRepository(repository, nil))
	129	+	d.serveCtx, d.serveCancel = context.WithCancel(context.Background())
	130	+	t.Cleanup(d.serveCancel)
	131	+	return d, catalogue
	132	+	}
	133	+
	134	+	func TestCatalogueRestoreIndependent(t *testing.T) {
	135	+	emptyDaemon, emptyCatalogue := newDurableRecoveryDaemon(t, nil, &durableRecoveryRepository{})
	136	+	emptyDone := make(chan struct{})
	137	+	go func() {
	138	+	emptyDaemon.restoreCatalogue(context.Background(), emptyCatalogue.Records())
	139	+	close(emptyDone)
	140	+	}()
	141	+	select {
	142	+	case <-emptyDone:
	143	+	case <-time.After(time.Second):
	144	+	t.Fatal("empty catalogue restoration blocked")
	145	+	}
	146	+
	147	+	healthy := durableRecoveryRecord(0)
	148	+	broken := durableRecoveryRecord(1)
	149	+	repository := &durableRecoveryRepository{
	150	+	generations: map[string]ports.SnapshotGeneration{healthy.Name: validGeneration(t, healthy)},
	151	+	errors: map[string]error{broken.Name: ports.ErrBudgetExhausted},
	152	+	loads: make(map[string]int),
	153	+	repairs: make(map[string]domain.CheckpointRef),
	154	+	}
	155	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{healthy, broken}, repository)
	156	+
	157	+	d.restoreCatalogue(context.Background(), catalogue.Records())
	158	+
	159	+	require.Equal(t, runtimeHealthy, d.stopped[healthy.Name].state)
	160	+	require.Equal(t, runtimeDegraded, d.stopped[broken.Name].state)
	161	+	for _, name := range []string{healthy.Name, broken.Name} {
	162	+	select {
	163	+	case <-d.stopped[name].restoreDone:
	164	+	default:
	165	+	t.Fatalf("restore completion for %q was not closed", name)
	166	+	}
	167	+	}
	168	+	degraded, ok := catalogue.Record(broken.Name)
	169	+	require.True(t, ok)
	170	+	require.Equal(t, domain.RecoveryDegraded, degraded.RecoveryState)
	171	+	require.Equal(t, 1, repository.loads[healthy.Name])
	172	+	require.Equal(t, 1, repository.loads[broken.Name])
	173	+	require.Empty(t, d.sessions, "a corrupt record must never open a replacement runtime")
	174	+
	175	+	fallback := durableRecoveryRecord(2)
	176	+	fallbackRef := domain.CheckpointRef{Generation: fallback.Committed.Generation - 1, ManifestDigest: [32]byte{77}}
	177	+	fallback.Fallbacks[0] = &fallbackRef
	178	+	fallbackSnapshotRecord := fallback
	179	+	fallbackSnapshotRecord.Committed = &fallbackRef
	180	+	fallbackRepository := &durableRecoveryRepository{
	181	+	generations: map[string]ports.SnapshotGeneration{fallback.Name: validGeneration(t, fallbackSnapshotRecord)},
	182	+	errors: make(map[string]error),
	183	+	loads: make(map[string]int),
	184	+	repairs: make(map[string]domain.CheckpointRef),
	185	+	}
	186	+	fallbackDaemon, fallbackCatalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{fallback}, fallbackRepository)
	187	+	fallbackDaemon.restoreCatalogue(context.Background(), fallbackCatalogue.Records())
	188	+	promoted, ok := fallbackCatalogue.Record(fallback.Name)
	189	+	require.True(t, ok)
	190	+	require.Equal(t, fallbackRef, *promoted.Committed)
	191	+	require.Equal(t, runtimeHealthy, fallbackDaemon.stopped[fallback.Name].state)
	192	+	require.Equal(t, 2, fallbackRepository.loads[fallback.Name])
	193	+	require.Equal(t, fallbackRef, fallbackRepository.repairs[fallback.IncarnationID.String()])
	194	+	}
	195	+
	196	+	func TestCatalogueRestoreAggregateOver4096(t *testing.T) {
	197	+	const count = 4200
	198	+	records := make([]domain.CatalogueRecord, count)
	199	+	repository := &durableRecoveryRepository{generations: make(map[string]ports.SnapshotGeneration, count), errors: make(map[string]error), loads: make(map[string]int), repairs: make(map[string]domain.CheckpointRef)}
	200	+	for i := range records {
	201	+	records[i] = durableRecoveryRecord(i)
	202	+	repository.generations[records[i].Name] = validGeneration(t, records[i])
	203	+	}
	204	+	d, catalogue := newDurableRecoveryDaemon(t, records, repository)
	205	+
	206	+	d.restoreCatalogue(context.Background(), catalogue.Records())
	207	+
	208	+	require.Len(t, d.stopped, count)
	209	+	for _, record := range records {
	210	+	require.Equal(t, runtimeHealthy, d.stopped[record.Name].state)
	211	+	require.Equal(t, 1, repository.loads[record.Name])
	212	+	}
	213	+	}
	214	+
	215	+	func TestCatalogueRestoreSingleSessionOver4096(t *testing.T) {
	216	+	record := durableRecoveryRecord(0)
	217	+	repository := &durableRecoveryRepository{generations: map[string]ports.SnapshotGeneration{record.Name: validGeneration(t, record)}, errors: make(map[string]error), loads: make(map[string]int), repairs: make(map[string]domain.CheckpointRef)}
	218	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
	219	+
	220	+	// The repository may contain any number of stale entries; restoration has no
	221	+	// discovery API and therefore performs exactly one direct catalogue lookup.
	222	+	d.restoreCatalogue(context.Background(), catalogue.Records())
	223	+
	224	+	require.Equal(t, runtimeHealthy, d.stopped[record.Name].state)
	225	+	require.Equal(t, 1, repository.loads[record.Name])
	226	+	}
	227	+
	228	+	func TestCatalogueRestoreIncarnationMismatch(t *testing.T) {
	229	+	record := durableRecoveryRecord(0)
	230	+	generation := validGeneration(t, record)
	231	+	generation.IncarnationID = domain.IncarnationID{99}
	232	+	repository := &durableRecoveryRepository{generations: map[string]ports.SnapshotGeneration{record.Name: generation}, errors: make(map[string]error), loads: make(map[string]int), repairs: make(map[string]domain.CheckpointRef)}
	233	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
	234	+
	235	+	d.restoreCatalogue(context.Background(), catalogue.Records())
	236	+
	237	+	require.Equal(t, runtimeDegraded, d.stopped[record.Name].state)
	238	+	got, _ := catalogue.Record(record.Name)
	239	+	require.Equal(t, domain.RecoveryDegraded, got.RecoveryState)
	240	+	}
	241	+
	242	+	func TestCatalogueRestoreDegradedVisible(t *testing.T) {
	243	+	record := durableRecoveryRecord(0)
	244	+	record.RecoveryState = domain.RecoveryDegraded
	245	+	record.DegradedReason = "checkpoint validation failed"
	246	+	repository := &durableRecoveryRepository{generations: make(map[string]ports.SnapshotGeneration), errors: make(map[string]error), loads: make(map[string]int), repairs: make(map[string]domain.CheckpointRef)}
	247	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
	248	+
	249	+	d.restoreCatalogue(context.Background(), catalogue.Records())
	250	+
	251	+	require.Equal(t, runtimeDegraded, d.stopped[record.Name].state)
	252	+	require.Zero(t, repository.loads[record.Name])
	253	+	}

@@ -0,0 +1,253 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"fmt"
	7	+	"io"
	8	+	"log/slog"
	9	+	"sync"
	10	+	"testing"
	11	+	"time"
	12	+
	13	+	"github.com/stretchr/testify/require"
	14	+
	15	+	"github.com/bnema/vev/internal/domain"
	16	+	"github.com/bnema/vev/internal/ports"
	17	+	snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
	18	+	)
	19	+
	20	+	type durableRecoveryCatalogue struct {
	21	+	ports.Catalogue
	22	+	mu sync.Mutex
	23	+	records map[string]domain.CatalogueRecord
	24	+	}
	25	+
	26	+	func newDurableRecoveryCatalogue(records []domain.CatalogueRecord) *durableRecoveryCatalogue {
	27	+	catalogue := &durableRecoveryCatalogue{records: make(map[string]domain.CatalogueRecord, len(records))}
	28	+	for _, record := range records {
	29	+	catalogue.records[record.Name] = record
	30	+	}
	31	+	return catalogue
	32	+	}
	33	+
	34	+	func (c *durableRecoveryCatalogue) Records() []domain.CatalogueRecord {
	35	+	c.mu.Lock()
	36	+	defer c.mu.Unlock()
	37	+	records := make([]domain.CatalogueRecord, 0, len(c.records))
	38	+	for _, record := range c.records {
	39	+	records = append(records, record)
	40	+	}
	41	+	return records
	42	+	}
	43	+
	44	+	func (c *durableRecoveryCatalogue) Record(name string) (domain.CatalogueRecord, bool) {
	45	+	c.mu.Lock()
	46	+	defer c.mu.Unlock()
	47	+	record, ok := c.records[name]
	48	+	return record, ok
	49	+	}
	50	+
	51	+	func (c *durableRecoveryCatalogue) Replace(name string, record domain.CatalogueRecord) error {
	52	+	c.mu.Lock()
	53	+	defer c.mu.Unlock()
	54	+	if name != record.Name {
	55	+	return errors.New("catalogue key mismatch")
	56	+	}
	57	+	c.records[name] = record
	58	+	return nil
	59	+	}
	60	+
	61	+	func (c *durableRecoveryCatalogue) Close() error { return nil }
	62	+
	63	+	type durableRecoveryRepository struct {
	64	+	ports.SnapshotRepository
	65	+	mu sync.Mutex
	66	+	generations map[string]ports.SnapshotGeneration
	67	+	errors map[string]error
	68	+	loads map[string]int
	69	+	repairs map[string]domain.CheckpointRef
	70	+	}
	71	+
	72	+	func (r *durableRecoveryRepository) LoadCheckpoint(_ context.Context, id domain.IncarnationID, name string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {
	73	+	r.mu.Lock()
	74	+	defer r.mu.Unlock()
	75	+	r.loads[name]++
	76	+	if err := r.errors[name]; err != nil {
	77	+	return ports.SnapshotGeneration{}, err
	78	+	}
	79	+	generation, ok := r.generations[name]
	80	+	if !ok \|\| generation.Generation != ref.Generation {
	81	+	return ports.SnapshotGeneration{}, errors.New("checkpoint unavailable")
	82	+	}
	83	+	return generation, nil
	84	+	}
	85	+
	86	+	func (r *durableRecoveryRepository) RepairHEAD(_ context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {
	87	+	r.mu.Lock()
	88	+	defer r.mu.Unlock()
	89	+	r.repairs[id.String()] = ref
	90	+	return nil
	91	+	}
	92	+
	93	+	func durableRecoveryRecord(index int) domain.CatalogueRecord {
	94	+	id := domain.IncarnationID{0: byte(index>>8 + 1), 1: byte(index + 1)}
	95	+	ref := domain.CheckpointRef{Generation: uint64(index + 1), ManifestDigest: [32]byte{byte(index%255 + 1)}}
	96	+	return domain.CatalogueRecord{
	97	+	Name: fmt.Sprintf("session-%04d", index),
	98	+	IncarnationID: id,
	99	+	Cwd: "/tmp",
	100	+	CreatedAt: int64(index + 1),
	101	+	UpdatedAt: int64(index + 1),
	102	+	RecoveryState: domain.RecoveryHealthy,
	103	+	Committed: &ref,
	104	+	}
	105	+	}
	106	+
	107	+	func validGeneration(t testing.TB, record domain.CatalogueRecord) ports.SnapshotGeneration {
	108	+	t.Helper()
	109	+	manifest, err := snapcodec.MarshalManifest(snapcodec.Manifest{
	110	+	Generation: record.Committed.Generation,
	111	+	IncarnationID: record.IncarnationID,
	112	+	Name: record.Name,
	113	+	CreatedAt: uint64(record.CreatedAt),
	114	+	})
	115	+	require.NoError(t, err)
	116	+	return ports.SnapshotGeneration{
	117	+	IncarnationID: record.IncarnationID,
	118	+	Name: record.Name,
	119	+	Generation: record.Committed.Generation,
	120	+	Manifest: manifest,
	121	+	Objects: map[ports.SnapshotDigest][]byte{},
	122	+	}
	123	+	}
	124	+
	125	+	func newDurableRecoveryDaemon(t testing.TB, records []domain.CatalogueRecord, repository ports.SnapshotRepository) (Daemon, durableRecoveryCatalogue) {
	126	+	t.Helper()
	127	+	catalogue := newDurableRecoveryCatalogue(records)
	128	+	d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(catalogue, records), WithSnapshotRepository(repository, nil))
	129	+	d.serveCtx, d.serveCancel = context.WithCancel(context.Background())
	130	+	t.Cleanup(d.serveCancel)
	131	+	return d, catalogue
	132	+	}
	133	+
	134	+	func TestCatalogueRestoreIndependent(t *testing.T) {
	135	+	emptyDaemon, emptyCatalogue := newDurableRecoveryDaemon(t, nil, &durableRecoveryRepository{})
	136	+	emptyDone := make(chan struct{})
	137	+	go func() {
	138	+	emptyDaemon.restoreCatalogue(context.Background(), emptyCatalogue.Records())
	139	+	close(emptyDone)
	140	+	}()
	141	+	select {
	142	+	case <-emptyDone:
	143	+	case <-time.After(time.Second):
	144	+	t.Fatal("empty catalogue restoration blocked")
	145	+	}
	146	+
	147	+	healthy := durableRecoveryRecord(0)
	148	+	broken := durableRecoveryRecord(1)
	149	+	repository := &durableRecoveryRepository{
	150	+	generations: map[string]ports.SnapshotGeneration{healthy.Name: validGeneration(t, healthy)},
	151	+	errors: map[string]error{broken.Name: ports.ErrBudgetExhausted},
	152	+	loads: make(map[string]int),
	153	+	repairs: make(map[string]domain.CheckpointRef),
	154	+	}
	155	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{healthy, broken}, repository)
	156	+
	157	+	d.restoreCatalogue(context.Background(), catalogue.Records())
	158	+
	159	+	require.Equal(t, runtimeHealthy, d.stopped[healthy.Name].state)
	160	+	require.Equal(t, runtimeDegraded, d.stopped[broken.Name].state)
	161	+	for _, name := range []string{healthy.Name, broken.Name} {
	162	+	select {
	163	+	case <-d.stopped[name].restoreDone:
	164	+	default:
	165	+	t.Fatalf("restore completion for %q was not closed", name)
	166	+	}
	167	+	}
	168	+	degraded, ok := catalogue.Record(broken.Name)
	169	+	require.True(t, ok)
	170	+	require.Equal(t, domain.RecoveryDegraded, degraded.RecoveryState)
	171	+	require.Equal(t, 1, repository.loads[healthy.Name])
	172	+	require.Equal(t, 1, repository.loads[broken.Name])
	173	+	require.Empty(t, d.sessions, "a corrupt record must never open a replacement runtime")
	174	+
	175	+	fallback := durableRecoveryRecord(2)
	176	+	fallbackRef := domain.CheckpointRef{Generation: fallback.Committed.Generation - 1, ManifestDigest: [32]byte{77}}
	177	+	fallback.Fallbacks[0] = &fallbackRef
	178	+	fallbackSnapshotRecord := fallback
	179	+	fallbackSnapshotRecord.Committed = &fallbackRef
	180	+	fallbackRepository := &durableRecoveryRepository{
	181	+	generations: map[string]ports.SnapshotGeneration{fallback.Name: validGeneration(t, fallbackSnapshotRecord)},
	182	+	errors: make(map[string]error),
	183	+	loads: make(map[string]int),
	184	+	repairs: make(map[string]domain.CheckpointRef),
	185	+	}
	186	+	fallbackDaemon, fallbackCatalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{fallback}, fallbackRepository)
	187	+	fallbackDaemon.restoreCatalogue(context.Background(), fallbackCatalogue.Records())
	188	+	promoted, ok := fallbackCatalogue.Record(fallback.Name)
	189	+	require.True(t, ok)
	190	+	require.Equal(t, fallbackRef, *promoted.Committed)
	191	+	require.Equal(t, runtimeHealthy, fallbackDaemon.stopped[fallback.Name].state)
	192	+	require.Equal(t, 2, fallbackRepository.loads[fallback.Name])
	193	+	require.Equal(t, fallbackRef, fallbackRepository.repairs[fallback.IncarnationID.String()])
	194	+	}
	195	+
	196	+	func TestCatalogueRestoreAggregateOver4096(t *testing.T) {
	197	+	const count = 4200
	198	+	records := make([]domain.CatalogueRecord, count)
	199	+	repository := &durableRecoveryRepository{generations: make(map[string]ports.SnapshotGeneration, count), errors: make(map[string]error), loads: make(map[string]int), repairs: make(map[string]domain.CheckpointRef)}
	200	+	for i := range records {
	201	+	records[i] = durableRecoveryRecord(i)
	202	+	repository.generations[records[i].Name] = validGeneration(t, records[i])
	203	+	}
	204	+	d, catalogue := newDurableRecoveryDaemon(t, records, repository)
	205	+
	206	+	d.restoreCatalogue(context.Background(), catalogue.Records())
	207	+
	208	+	require.Len(t, d.stopped, count)
	209	+	for _, record := range records {
	210	+	require.Equal(t, runtimeHealthy, d.stopped[record.Name].state)
	211	+	require.Equal(t, 1, repository.loads[record.Name])
	212	+	}
	213	+	}
	214	+
	215	+	func TestCatalogueRestoreSingleSessionOver4096(t *testing.T) {
	216	+	record := durableRecoveryRecord(0)
	217	+	repository := &durableRecoveryRepository{generations: map[string]ports.SnapshotGeneration{record.Name: validGeneration(t, record)}, errors: make(map[string]error), loads: make(map[string]int), repairs: make(map[string]domain.CheckpointRef)}
	218	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
	219	+
	220	+	// The repository may contain any number of stale entries; restoration has no
	221	+	// discovery API and therefore performs exactly one direct catalogue lookup.
	222	+	d.restoreCatalogue(context.Background(), catalogue.Records())
	223	+
	224	+	require.Equal(t, runtimeHealthy, d.stopped[record.Name].state)
	225	+	require.Equal(t, 1, repository.loads[record.Name])
	226	+	}
	227	+
	228	+	func TestCatalogueRestoreIncarnationMismatch(t *testing.T) {
	229	+	record := durableRecoveryRecord(0)
	230	+	generation := validGeneration(t, record)
	231	+	generation.IncarnationID = domain.IncarnationID{99}
	232	+	repository := &durableRecoveryRepository{generations: map[string]ports.SnapshotGeneration{record.Name: generation}, errors: make(map[string]error), loads: make(map[string]int), repairs: make(map[string]domain.CheckpointRef)}
	233	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
	234	+
	235	+	d.restoreCatalogue(context.Background(), catalogue.Records())
	236	+
	237	+	require.Equal(t, runtimeDegraded, d.stopped[record.Name].state)
	238	+	got, _ := catalogue.Record(record.Name)
	239	+	require.Equal(t, domain.RecoveryDegraded, got.RecoveryState)
	240	+	}
	241	+
	242	+	func TestCatalogueRestoreDegradedVisible(t *testing.T) {
	243	+	record := durableRecoveryRecord(0)
	244	+	record.RecoveryState = domain.RecoveryDegraded
	245	+	record.DegradedReason = "checkpoint validation failed"
	246	+	repository := &durableRecoveryRepository{generations: make(map[string]ports.SnapshotGeneration), errors: make(map[string]error), loads: make(map[string]int), repairs: make(map[string]domain.CheckpointRef)}
	247	+	d, catalogue := newDurableRecoveryDaemon(t, []domain.CatalogueRecord{record}, repository)
	248	+
	249	+	d.restoreCatalogue(context.Background(), catalogue.Records())
	250	+
	251	+	require.Equal(t, runtimeDegraded, d.stopped[record.Name].state)
	252	+	require.Zero(t, repository.loads[record.Name])
	253	+	}

M internal/usecase/daemon/session_test.go +22 -4


@@ -1766,15 +1766,33 @@ func TestNamedSessionLifecycleExhaustionDoesNotMutateSessionState(t *testing.T)
1766	1766		}
1767	1767
1768	1768		func TestCatalogueRecordsConstructExpectedSessionRegistry(t *testing.T) {
	1769	+	alphaRef := &domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
	1770	+	workRef := &domain.CheckpointRef{Generation: 2, ManifestDigest: [32]byte{2}}
1769	1771		records := []domain.CatalogueRecord{
1770		-	{Name: "alpha", IncarnationID: domain.IncarnationID{1}, Cwd: "/tmp/alpha", CreatedAt: 7, LastUsedSeq: 11, TabNames: []string{"shell"}, RecoveryState: domain.RecoveryHealthy},
1771		-	{Name: "work", IncarnationID: domain.IncarnationID{2}, Cwd: "/tmp/work", CreatedAt: 9, LastUsedSeq: 13, TabNames: []string{"editor", "logs"}, RecoveryState: domain.RecoveryDegraded, DegradedReason: "uncertain legacy checkpoint"},
	1772	+	{Name: "alpha", IncarnationID: domain.IncarnationID{1}, Cwd: "/tmp/alpha", CreatedAt: 7, LastUsedSeq: 11, TabNames: []string{"shell"}, RecoveryState: domain.RecoveryHealthy, Committed: alphaRef},
	1773	+	{Name: "work", IncarnationID: domain.IncarnationID{2}, Cwd: "/tmp/work", CreatedAt: 9, LastUsedSeq: 13, TabNames: []string{"editor", "logs"}, RecoveryState: domain.RecoveryDegraded, Committed: workRef, DegradedReason: "uncertain legacy checkpoint"},
1772	1774		}
1773	1775
1774	1776		d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(persist.New(nil), records))
1775	1777
1776		-	require.Equal(t, stoppedSession{name: "alpha", incarnation: domain.IncarnationID{1}, cwd: "/tmp/alpha", createdAt: 7, lastUsedSeq: 11, tabNames: []string{"shell"}}, d.stopped["alpha"])
1777		-	require.Equal(t, stoppedSession{name: "work", incarnation: domain.IncarnationID{2}, cwd: "/tmp/work", createdAt: 9, lastUsedSeq: 13, tabNames: []string{"editor", "logs"}}, d.stopped["work"])
	1778	+	alpha := d.stopped["alpha"]
	1779	+	require.Equal(t, records[0], alpha.record)
	1780	+	require.Equal(t, runtimeRestoring, alpha.state)
	1781	+	require.NotNil(t, alpha.restoreDone)
	1782	+	select {
	1783	+	case <-alpha.restoreDone:
	1784	+	t.Fatal("healthy catalogue record must begin in restoring state")
	1785	+	default:
	1786	+	}
	1787	+	work := d.stopped["work"]
	1788	+	require.Equal(t, records[1], work.record)
	1789	+	require.Equal(t, runtimeDegraded, work.state)
	1790	+	require.NotNil(t, work.restoreDone)
	1791	+	select {
	1792	+	case <-work.restoreDone:
	1793	+	default:
	1794	+	t.Fatal("degraded catalogue record must not wait for restoration")
	1795	+	}
1778	1796		require.Equal(t, int64(9), d.lastAllocatedCreatedAt)
1779	1797		require.Equal(t, uint64(13), d.mruSeq.Load())
1780	1798		}


@@ -1766,15 +1766,33 @@ func TestNamedSessionLifecycleExhaustionDoesNotMutateSessionState(t *testing.T)
1766		}	1766		}
1767			1767
1768		func TestCatalogueRecordsConstructExpectedSessionRegistry(t *testing.T) {	1768		func TestCatalogueRecordsConstructExpectedSessionRegistry(t *testing.T) {
			1769	+	alphaRef := &domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
			1770	+	workRef := &domain.CheckpointRef{Generation: 2, ManifestDigest: [32]byte{2}}
1769		records := []domain.CatalogueRecord{	1771		records := []domain.CatalogueRecord{
1770	-	{Name: "alpha", IncarnationID: domain.IncarnationID{1}, Cwd: "/tmp/alpha", CreatedAt: 7, LastUsedSeq: 11, TabNames: []string{"shell"}, RecoveryState: domain.RecoveryHealthy},	1772	+	{Name: "alpha", IncarnationID: domain.IncarnationID{1}, Cwd: "/tmp/alpha", CreatedAt: 7, LastUsedSeq: 11, TabNames: []string{"shell"}, RecoveryState: domain.RecoveryHealthy, Committed: alphaRef},
1771	-	{Name: "work", IncarnationID: domain.IncarnationID{2}, Cwd: "/tmp/work", CreatedAt: 9, LastUsedSeq: 13, TabNames: []string{"editor", "logs"}, RecoveryState: domain.RecoveryDegraded, DegradedReason: "uncertain legacy checkpoint"},	1773	+	{Name: "work", IncarnationID: domain.IncarnationID{2}, Cwd: "/tmp/work", CreatedAt: 9, LastUsedSeq: 13, TabNames: []string{"editor", "logs"}, RecoveryState: domain.RecoveryDegraded, Committed: workRef, DegradedReason: "uncertain legacy checkpoint"},
1772		}	1774		}
1773			1775
1774		d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(persist.New(nil), records))	1776		d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(persist.New(nil), records))
1775			1777
1776	-	require.Equal(t, stoppedSession{name: "alpha", incarnation: domain.IncarnationID{1}, cwd: "/tmp/alpha", createdAt: 7, lastUsedSeq: 11, tabNames: []string{"shell"}}, d.stopped["alpha"])	1778	+	alpha := d.stopped["alpha"]
1777	-	require.Equal(t, stoppedSession{name: "work", incarnation: domain.IncarnationID{2}, cwd: "/tmp/work", createdAt: 9, lastUsedSeq: 13, tabNames: []string{"editor", "logs"}}, d.stopped["work"])	1779	+	require.Equal(t, records[0], alpha.record)
			1780	+	require.Equal(t, runtimeRestoring, alpha.state)
			1781	+	require.NotNil(t, alpha.restoreDone)
			1782	+	select {
			1783	+	case <-alpha.restoreDone:
			1784	+	t.Fatal("healthy catalogue record must begin in restoring state")
			1785	+	default:
			1786	+	}
			1787	+	work := d.stopped["work"]
			1788	+	require.Equal(t, records[1], work.record)
			1789	+	require.Equal(t, runtimeDegraded, work.state)
			1790	+	require.NotNil(t, work.restoreDone)
			1791	+	select {
			1792	+	case <-work.restoreDone:
			1793	+	default:
			1794	+	t.Fatal("degraded catalogue record must not wait for restoration")
			1795	+	}
1778		require.Equal(t, int64(9), d.lastAllocatedCreatedAt)	1796		require.Equal(t, int64(9), d.lastAllocatedCreatedAt)
1779		require.Equal(t, uint64(13), d.mruSeq.Load())	1797		require.Equal(t, uint64(13), d.mruSeq.Load())
1780		}	1798		}

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


@@ -267,8 +267,12 @@ func (d Daemon) persistAndRegisterRestoredSession(ctx context.Context, sess se
267	267		}
268	268		}
269	269		}
270		-	if err := d.persist.Save(sess.persistRecordLocked(sess.createdAt)); err != nil {
271		-	return false, err
	270	+	// The catalogue checkpoint already authorizes this runtime. Rewriting it
	271	+	// during restoration would discard committed/fallback recovery metadata.
	272	+	if d.catalogue == nil {
	273	+	if err := d.persist.Save(sess.persistRecordLocked(sess.createdAt)); err != nil {
	274	+	return false, err
	275	+	}
272	276		}
273	277		d.mu.Lock()
274	278		defer d.mu.Unlock()

 			}
 		}
 	}
-	if err := d.persist.Save(sess.persistRecordLocked(sess.createdAt)); err != nil {
+	// The catalogue checkpoint already authorizes this runtime. Rewriting it
-		return false, err
+	// during restoration would discard committed/fallback recovery metadata.
+	if d.catalogue == nil {
+		if err := d.persist.Save(sess.persistRecordLocked(sess.createdAt)); err != nil {
+			return false, err
+		}
 	}
 	d.mu.Lock()
 	defer d.mu.Unlock()

M internal/usecase/daemon/snapshot_restore_incremental.go +235 -4


@@ -3,16 +3,247 @@ package daemon
3	3		import (
4	4		"context"
5	5		"crypto/sha256"
	6	+	"errors"
6	7		"fmt"
	8	+	"sync"
7	9
	10	+	"github.com/bnema/vev/internal/domain"
8	11		"github.com/bnema/vev/internal/ports"
9	12		snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
10	13		)
11	14
12		-	// restoreIncrementalSnapshots intentionally performs no repository discovery.
13		-	// Catalogue-indexed restoration is introduced as a separate recovery phase;
14		-	// until then startup must not fall back to name scans or generation traversal.
15		-	func (d *Daemon) restoreIncrementalSnapshots(context.Context) {}
	15	+	const catalogueRestoreConcurrency = 8
	16	+
	17	+	func initialRuntimeRecoveryState(record domain.CatalogueRecord) (runtimeRecoveryState, chan struct{}) {
	18	+	done := make(chan struct{})
	19	+	var state runtimeRecoveryState
	20	+	switch record.RecoveryState {
	21	+	case domain.RecoveryFresh:
	22	+	state = runtimeFresh
	23	+	close(done)
	24	+	case domain.RecoveryHealthy:
	25	+	state = runtimeRestoring
	26	+	case domain.RecoveryDegraded:
	27	+	state = runtimeDegraded
	28	+	close(done)
	29	+	case domain.RecoveryDeleting:
	30	+	state = runtimeDeleting
	31	+	close(done)
	32	+	default:
	33	+	state = runtimeDegraded
	34	+	close(done)
	35	+	}
	36	+	return state, done
	37	+	}
	38	+
	39	+	// restoreIncrementalSnapshots restores only records named by the authoritative
	40	+	// catalogue. Snapshot repository discovery is deliberately not part of this
	41	+	// path, so stale or oversized namespaces cannot affect unrelated sessions.
	42	+	func (d *Daemon) restoreIncrementalSnapshots(ctx context.Context) {
	43	+	if d.catalogue == nil {
	44	+	return
	45	+	}
	46	+	d.restoreCatalogue(ctx, d.catalogue.Records())
	47	+	}
	48	+
	49	+	// restoreCatalogue gives every expected record an independent bounded job.
	50	+	// A failed record is degraded without cancelling any sibling restoration.
	51	+	func (d *Daemon) restoreCatalogue(ctx context.Context, records []domain.CatalogueRecord) {
	52	+	jobs := make(chan domain.CatalogueRecord)
	53	+	workers := min(catalogueRestoreConcurrency, len(records))
	54	+	var wg sync.WaitGroup
	55	+	for range workers {
	56	+	wg.Go(func() {
	57	+	for record := range jobs {
	58	+	d.ensureCatalogueRegistryEntry(record)
	59	+	done := d.recordRestoreDone(record.Name)
	60	+	err := d.restoreRecord(ctx, record)
	61	+	if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
	62	+	d.markRecordDegraded(record, err)
	63	+	}
	64	+	closeRuntimeRestoreDone(done)
	65	+	}
	66	+	})
	67	+	}
	68	+	for _, record := range records {
	69	+	jobs <- record
	70	+	}
	71	+	close(jobs)
	72	+	wg.Wait()
	73	+	}
	74	+
	75	+	func (d *Daemon) ensureCatalogueRegistryEntry(record domain.CatalogueRecord) {
	76	+	d.mu.Lock()
	77	+	defer d.mu.Unlock()
	78	+	if entry, ok := d.stopped[record.Name]; ok && entry.restoreDone != nil {
	79	+	return
	80	+	}
	81	+	state, done := initialRuntimeRecoveryState(record)
	82	+	d.stopped[record.Name] = stoppedSession{
	83	+	name: record.Name,
	84	+	cwd: record.Cwd,
	85	+	createdAt: record.CreatedAt,
	86	+	incarnation: record.IncarnationID,
	87	+	lastUsedSeq: record.LastUsedSeq,
	88	+	tabNames: append([]string(nil), record.TabNames...),
	89	+	record: record,
	90	+	state: state,
	91	+	restoreDone: done,
	92	+	}
	93	+	}
	94	+
	95	+	func (d *Daemon) recordRestoreDone(name string) chan struct{} {
	96	+	d.mu.Lock()
	97	+	defer d.mu.Unlock()
	98	+	return d.stopped[name].restoreDone
	99	+	}
	100	+
	101	+	func closeRuntimeRestoreDone(done chan struct{}) {
	102	+	if done == nil {
	103	+	return
	104	+	}
	105	+	select {
	106	+	case <-done:
	107	+	default:
	108	+	close(done)
	109	+	}
	110	+	}
	111	+
	112	+	func (d *Daemon) setStoppedRecovery(record domain.CatalogueRecord, state runtimeRecoveryState) {
	113	+	d.mu.Lock()
	114	+	defer d.mu.Unlock()
	115	+	entry, ok := d.stopped[record.Name]
	116	+	if !ok {
	117	+	return
	118	+	}
	119	+	entry.record = record
	120	+	entry.state = state
	121	+	entry.cwd = record.Cwd
	122	+	entry.createdAt = record.CreatedAt
	123	+	entry.incarnation = record.IncarnationID
	124	+	entry.lastUsedSeq = record.LastUsedSeq
	125	+	entry.tabNames = append([]string(nil), record.TabNames...)
	126	+	d.stopped[record.Name] = entry
	127	+	}
	128	+
	129	+	func (d *Daemon) markRecordDegraded(record domain.CatalogueRecord, restoreErr error) {
	130	+	if d.catalogue != nil {
	131	+	if current, ok := d.catalogue.Record(record.Name); ok && current.IncarnationID == record.IncarnationID {
	132	+	record = current
	133	+	}
	134	+	}
	135	+	record.RecoveryState = domain.RecoveryDegraded
	136	+	record.DegradedReason = "checkpoint validation failed"
	137	+	if d.catalogue != nil {
	138	+	if err := d.catalogue.Replace(record.Name, record); err != nil {
	139	+	d.log.Error("marking session degraded failed", "session", record.Name, "err", err)
	140	+	}
	141	+	}
	142	+	d.setStoppedRecovery(record, runtimeDegraded)
	143	+	d.log.Warn("session checkpoint restore failed", "session", record.Name, "err", restoreErr)
	144	+	}
	145	+
	146	+	func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecord) error {
	147	+	switch record.RecoveryState {
	148	+	case domain.RecoveryFresh:
	149	+	d.setStoppedRecovery(record, runtimeFresh)
	150	+	return nil
	151	+	case domain.RecoveryDegraded:
	152	+	d.setStoppedRecovery(record, runtimeDegraded)
	153	+	return nil
	154	+	case domain.RecoveryDeleting:
	155	+	d.setStoppedRecovery(record, runtimeDeleting)
	156	+	return nil
	157	+	case domain.RecoveryHealthy:
	158	+	// Continue below.
	159	+	default:
	160	+	return errors.New("snapshot: invalid catalogue recovery state")
	161	+	}
	162	+	if d.snapshotRepository == nil {
	163	+	return errors.New("snapshot: repository unavailable")
	164	+	}
	165	+
	166	+	candidates := checkpointCandidates(record)
	167	+	if len(candidates) == 0 {
	168	+	return errors.New("snapshot: healthy record has no checkpoint")
	169	+	}
	170	+	valid := make([]domain.CheckpointRef, 0, len(candidates))
	171	+	var selected domain.CheckpointRef
	172	+	var selectedGeneration ports.SnapshotGeneration
	173	+	var selectedSnapshot snapcodec.Session
	174	+	selectedIndex := -1
	175	+	for i, candidate := range candidates {
	176	+	generation, err := d.snapshotRepository.LoadCheckpoint(ctx, record.IncarnationID, record.Name, candidate)
	177	+	if err != nil {
	178	+	continue
	179	+	}
	180	+	snapshot, err := sessionFromGeneration(generation)
	181	+	if err != nil \|\| generation.IncarnationID != record.IncarnationID \|\| generation.Name != record.Name \|\| generation.Generation != candidate.Generation {
	182	+	continue
	183	+	}
	184	+	valid = append(valid, candidate)
	185	+	if selectedIndex < 0 {
	186	+	selected, selectedGeneration, selectedSnapshot, selectedIndex = candidate, generation, snapshot, i
	187	+	if i == 0 {
	188	+	break
	189	+	}
	190	+	}
	191	+	}
	192	+	if selectedIndex < 0 {
	193	+	return errors.New("snapshot: no catalogue checkpoint validated")
	194	+	}
	195	+
	196	+	if selectedIndex > 0 {
	197	+	record = promoteFallback(record, selected, valid)
	198	+	if d.catalogue == nil {
	199	+	return errors.New("snapshot: catalogue unavailable for fallback promotion")
	200	+	}
	201	+	if err := d.catalogue.Replace(record.Name, record); err != nil {
	202	+	return fmt.Errorf("snapshot: promote fallback: %w", err)
	203	+	}
	204	+	if err := d.snapshotRepository.RepairHEAD(ctx, record.IncarnationID, selected); err != nil {
	205	+	return fmt.Errorf("snapshot: repair head: %w", err)
	206	+	}
	207	+	d.log.Warn("restored session from fallback checkpoint", "session", record.Name, "generation", selected.Generation)
	208	+	}
	209	+
	210	+	if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation); err != nil {
	211	+	return err
	212	+	}
	213	+	d.setStoppedRecovery(record, runtimeHealthy)
	214	+	return nil
	215	+	}
	216	+
	217	+	func checkpointCandidates(record domain.CatalogueRecord) []domain.CheckpointRef {
	218	+	candidates := make([]domain.CheckpointRef, 0, 3)
	219	+	if record.Committed != nil {
	220	+	candidates = append(candidates, *record.Committed)
	221	+	}
	222	+	for _, fallback := range record.Fallbacks {
	223	+	if fallback != nil {
	224	+	candidates = append(candidates, *fallback)
	225	+	}
	226	+	}
	227	+	return candidates
	228	+	}
	229	+
	230	+	func promoteFallback(record domain.CatalogueRecord, selected domain.CheckpointRef, valid []domain.CheckpointRef) domain.CatalogueRecord {
	231	+	next := record
	232	+	next.RecoveryState = domain.RecoveryHealthy
	233	+	next.Committed = &selected
	234	+	next.Fallbacks = [2]*domain.CheckpointRef{}
	235	+	next.DegradedReason = ""
	236	+	slot := 0
	237	+	for _, candidate := range valid {
	238	+	if candidate == selected \|\| slot == len(next.Fallbacks) {
	239	+	continue
	240	+	}
	241	+	copy := candidate
	242	+	next.Fallbacks[slot] = &copy
	243	+	slot++
	244	+	}
	245	+	return next
	246	+	}
16	247
17	248		func sessionFromGeneration(generation ports.SnapshotGeneration) (snapcodec.Session, error) {
18	249		manifest, err := snapcodec.UnmarshalManifest(generation.Manifest)


@@ -3,16 +3,247 @@ package daemon
3		import (	3		import (
4		"context"	4		"context"
5		"crypto/sha256"	5		"crypto/sha256"
			6	+	"errors"
6		"fmt"	7		"fmt"
			8	+	"sync"
7			9
			10	+	"github.com/bnema/vev/internal/domain"
8		"github.com/bnema/vev/internal/ports"	11		"github.com/bnema/vev/internal/ports"
9		snapcodec "github.com/bnema/vev/internal/usecase/snapshot"	12		snapcodec "github.com/bnema/vev/internal/usecase/snapshot"
10		)	13		)
11			14
12	-	// restoreIncrementalSnapshots intentionally performs no repository discovery.	15	+	const catalogueRestoreConcurrency = 8
13	-	// Catalogue-indexed restoration is introduced as a separate recovery phase;	16	+
14	-	// until then startup must not fall back to name scans or generation traversal.	17	+	func initialRuntimeRecoveryState(record domain.CatalogueRecord) (runtimeRecoveryState, chan struct{}) {
15	-	func (d *Daemon) restoreIncrementalSnapshots(context.Context) {}	18	+	done := make(chan struct{})
			19	+	var state runtimeRecoveryState
			20	+	switch record.RecoveryState {
			21	+	case domain.RecoveryFresh:
			22	+	state = runtimeFresh
			23	+	close(done)
			24	+	case domain.RecoveryHealthy:
			25	+	state = runtimeRestoring
			26	+	case domain.RecoveryDegraded:
			27	+	state = runtimeDegraded
			28	+	close(done)
			29	+	case domain.RecoveryDeleting:
			30	+	state = runtimeDeleting
			31	+	close(done)
			32	+	default:
			33	+	state = runtimeDegraded
			34	+	close(done)
			35	+	}
			36	+	return state, done
			37	+	}
			38	+
			39	+	// restoreIncrementalSnapshots restores only records named by the authoritative
			40	+	// catalogue. Snapshot repository discovery is deliberately not part of this
			41	+	// path, so stale or oversized namespaces cannot affect unrelated sessions.
			42	+	func (d *Daemon) restoreIncrementalSnapshots(ctx context.Context) {
			43	+	if d.catalogue == nil {
			44	+	return
			45	+	}
			46	+	d.restoreCatalogue(ctx, d.catalogue.Records())
			47	+	}
			48	+
			49	+	// restoreCatalogue gives every expected record an independent bounded job.
			50	+	// A failed record is degraded without cancelling any sibling restoration.
			51	+	func (d *Daemon) restoreCatalogue(ctx context.Context, records []domain.CatalogueRecord) {
			52	+	jobs := make(chan domain.CatalogueRecord)
			53	+	workers := min(catalogueRestoreConcurrency, len(records))
			54	+	var wg sync.WaitGroup
			55	+	for range workers {
			56	+	wg.Go(func() {
			57	+	for record := range jobs {
			58	+	d.ensureCatalogueRegistryEntry(record)
			59	+	done := d.recordRestoreDone(record.Name)
			60	+	err := d.restoreRecord(ctx, record)
			61	+	if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
			62	+	d.markRecordDegraded(record, err)
			63	+	}
			64	+	closeRuntimeRestoreDone(done)
			65	+	}
			66	+	})
			67	+	}
			68	+	for _, record := range records {
			69	+	jobs <- record
			70	+	}
			71	+	close(jobs)
			72	+	wg.Wait()
			73	+	}
			74	+
			75	+	func (d *Daemon) ensureCatalogueRegistryEntry(record domain.CatalogueRecord) {
			76	+	d.mu.Lock()
			77	+	defer d.mu.Unlock()
			78	+	if entry, ok := d.stopped[record.Name]; ok && entry.restoreDone != nil {
			79	+	return
			80	+	}
			81	+	state, done := initialRuntimeRecoveryState(record)
			82	+	d.stopped[record.Name] = stoppedSession{
			83	+	name: record.Name,
			84	+	cwd: record.Cwd,
			85	+	createdAt: record.CreatedAt,
			86	+	incarnation: record.IncarnationID,
			87	+	lastUsedSeq: record.LastUsedSeq,
			88	+	tabNames: append([]string(nil), record.TabNames...),
			89	+	record: record,
			90	+	state: state,
			91	+	restoreDone: done,
			92	+	}
			93	+	}
			94	+
			95	+	func (d *Daemon) recordRestoreDone(name string) chan struct{} {
			96	+	d.mu.Lock()
			97	+	defer d.mu.Unlock()
			98	+	return d.stopped[name].restoreDone
			99	+	}
			100	+
			101	+	func closeRuntimeRestoreDone(done chan struct{}) {
			102	+	if done == nil {
			103	+	return
			104	+	}
			105	+	select {
			106	+	case <-done:
			107	+	default:
			108	+	close(done)
			109	+	}
			110	+	}
			111	+
			112	+	func (d *Daemon) setStoppedRecovery(record domain.CatalogueRecord, state runtimeRecoveryState) {
			113	+	d.mu.Lock()
			114	+	defer d.mu.Unlock()
			115	+	entry, ok := d.stopped[record.Name]
			116	+	if !ok {
			117	+	return
			118	+	}
			119	+	entry.record = record
			120	+	entry.state = state
			121	+	entry.cwd = record.Cwd
			122	+	entry.createdAt = record.CreatedAt
			123	+	entry.incarnation = record.IncarnationID
			124	+	entry.lastUsedSeq = record.LastUsedSeq
			125	+	entry.tabNames = append([]string(nil), record.TabNames...)
			126	+	d.stopped[record.Name] = entry
			127	+	}
			128	+
			129	+	func (d *Daemon) markRecordDegraded(record domain.CatalogueRecord, restoreErr error) {
			130	+	if d.catalogue != nil {
			131	+	if current, ok := d.catalogue.Record(record.Name); ok && current.IncarnationID == record.IncarnationID {
			132	+	record = current
			133	+	}
			134	+	}
			135	+	record.RecoveryState = domain.RecoveryDegraded
			136	+	record.DegradedReason = "checkpoint validation failed"
			137	+	if d.catalogue != nil {
			138	+	if err := d.catalogue.Replace(record.Name, record); err != nil {
			139	+	d.log.Error("marking session degraded failed", "session", record.Name, "err", err)
			140	+	}
			141	+	}
			142	+	d.setStoppedRecovery(record, runtimeDegraded)
			143	+	d.log.Warn("session checkpoint restore failed", "session", record.Name, "err", restoreErr)
			144	+	}
			145	+
			146	+	func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecord) error {
			147	+	switch record.RecoveryState {
			148	+	case domain.RecoveryFresh:
			149	+	d.setStoppedRecovery(record, runtimeFresh)
			150	+	return nil
			151	+	case domain.RecoveryDegraded:
			152	+	d.setStoppedRecovery(record, runtimeDegraded)
			153	+	return nil
			154	+	case domain.RecoveryDeleting:
			155	+	d.setStoppedRecovery(record, runtimeDeleting)
			156	+	return nil
			157	+	case domain.RecoveryHealthy:
			158	+	// Continue below.
			159	+	default:
			160	+	return errors.New("snapshot: invalid catalogue recovery state")
			161	+	}
			162	+	if d.snapshotRepository == nil {
			163	+	return errors.New("snapshot: repository unavailable")
			164	+	}
			165	+
			166	+	candidates := checkpointCandidates(record)
			167	+	if len(candidates) == 0 {
			168	+	return errors.New("snapshot: healthy record has no checkpoint")
			169	+	}
			170	+	valid := make([]domain.CheckpointRef, 0, len(candidates))
			171	+	var selected domain.CheckpointRef
			172	+	var selectedGeneration ports.SnapshotGeneration
			173	+	var selectedSnapshot snapcodec.Session
			174	+	selectedIndex := -1
			175	+	for i, candidate := range candidates {
			176	+	generation, err := d.snapshotRepository.LoadCheckpoint(ctx, record.IncarnationID, record.Name, candidate)
			177	+	if err != nil {
			178	+	continue
			179	+	}
			180	+	snapshot, err := sessionFromGeneration(generation)
			181	+	if err != nil \|\| generation.IncarnationID != record.IncarnationID \|\| generation.Name != record.Name \|\| generation.Generation != candidate.Generation {
			182	+	continue
			183	+	}
			184	+	valid = append(valid, candidate)
			185	+	if selectedIndex < 0 {
			186	+	selected, selectedGeneration, selectedSnapshot, selectedIndex = candidate, generation, snapshot, i
			187	+	if i == 0 {
			188	+	break
			189	+	}
			190	+	}
			191	+	}
			192	+	if selectedIndex < 0 {
			193	+	return errors.New("snapshot: no catalogue checkpoint validated")
			194	+	}
			195	+
			196	+	if selectedIndex > 0 {
			197	+	record = promoteFallback(record, selected, valid)
			198	+	if d.catalogue == nil {
			199	+	return errors.New("snapshot: catalogue unavailable for fallback promotion")
			200	+	}
			201	+	if err := d.catalogue.Replace(record.Name, record); err != nil {
			202	+	return fmt.Errorf("snapshot: promote fallback: %w", err)
			203	+	}
			204	+	if err := d.snapshotRepository.RepairHEAD(ctx, record.IncarnationID, selected); err != nil {
			205	+	return fmt.Errorf("snapshot: repair head: %w", err)
			206	+	}
			207	+	d.log.Warn("restored session from fallback checkpoint", "session", record.Name, "generation", selected.Generation)
			208	+	}
			209	+
			210	+	if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation); err != nil {
			211	+	return err
			212	+	}
			213	+	d.setStoppedRecovery(record, runtimeHealthy)
			214	+	return nil
			215	+	}
			216	+
			217	+	func checkpointCandidates(record domain.CatalogueRecord) []domain.CheckpointRef {
			218	+	candidates := make([]domain.CheckpointRef, 0, 3)
			219	+	if record.Committed != nil {
			220	+	candidates = append(candidates, *record.Committed)
			221	+	}
			222	+	for _, fallback := range record.Fallbacks {
			223	+	if fallback != nil {
			224	+	candidates = append(candidates, *fallback)
			225	+	}
			226	+	}
			227	+	return candidates
			228	+	}
			229	+
			230	+	func promoteFallback(record domain.CatalogueRecord, selected domain.CheckpointRef, valid []domain.CheckpointRef) domain.CatalogueRecord {
			231	+	next := record
			232	+	next.RecoveryState = domain.RecoveryHealthy
			233	+	next.Committed = &selected
			234	+	next.Fallbacks = [2]*domain.CheckpointRef{}
			235	+	next.DegradedReason = ""
			236	+	slot := 0
			237	+	for _, candidate := range valid {
			238	+	if candidate == selected \|\| slot == len(next.Fallbacks) {
			239	+	continue
			240	+	}
			241	+	copy := candidate
			242	+	next.Fallbacks[slot] = &copy
			243	+	slot++
			244	+	}
			245	+	return next
			246	+	}
16			247
17		func sessionFromGeneration(generation ports.SnapshotGeneration) (snapcodec.Session, error) {	248		func sessionFromGeneration(generation ports.SnapshotGeneration) (snapcodec.Session, error) {
18		manifest, err := snapcodec.UnmarshalManifest(generation.Manifest)	249		manifest, err := snapcodec.UnmarshalManifest(generation.Manifest)