test(recovery): prove migration record boundaries

commit: ea0c8496bf71304e629339007429807c5bb14c04
author: brice · 2mo ago · 2026-07-25 21:43 UTC
parent: 0b9089a
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

5 files · +158 -40

M internal/app/run.go +16 -5


@@ -464,6 +464,15 @@ func runDaemon() error {
464	464		return runWithLifecycleOwner(ctx, ipc.SocketDir(), platform.StateDir(), runDaemonOwned)
465	465		}
466	466
	467	+	func constructDaemonBeforeSocketPublication(
	468	+	construct func() *daemon.Daemon,
	469	+	listen func() (ports.Listener, error),
	470	+	) (*daemon.Daemon, ports.Listener, error) {
	471	+	d := construct()
	472	+	ln, err := listen()
	473	+	return d, ln, err
	474	+	}
	475	+
467	476		func runDaemonOwned(ctx context.Context) (retErr error) {
468	477		log, logCloser, err := configureLogging(logging.Daemon, true)
469	478		if err != nil {

@@ -522,11 +531,14 @@ func runDaemonOwned(ctx context.Context) (retErr error) {
522	531		return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())
523	532		}
524	533		log.Info("session persistence enabled", "path", storePath)
525		-	daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue))
	534	+	daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))
526	535
527		-	// Socket publication follows lifecycle acquisition, migration, and strict
528		-	// recovery. Any earlier failure leaves no apparently healthy daemon.
529		-	ln, err := ipc.Listen(ipc.SocketDir(), ipc.WithRuntimeObserver(observer))
	536	+	// Construct the catalogue-backed expected-session registry before socket
	537	+	// publication. Phase 3 snapshot restoration remains asynchronous in Serve.
	538	+	d, ln, err := constructDaemonBeforeSocketPublication(
	539	+	func() *daemon.Daemon { return daemon.New(pty.NewFactory(), clk, log, daemonOpts...) },
	540	+	func() (ports.Listener, error) { return ipc.Listen(ipc.SocketDir(), ipc.WithRuntimeObserver(observer)) },
	541	+	)
530	542		if err != nil {
531	543		closeErr := opened.Catalogue.Close()
532	544		log.Error("daemon listen failed", "socket_dir", ipc.SocketDir(), "err", err)

@@ -535,7 +547,6 @@ func runDaemonOwned(ctx context.Context) (retErr error) {
535	547		defer func() { _ = ln.Close() }()
536	548		log.Info("daemon starting", "socket", ln.Addr())
537	549
538		-	d := daemon.New(pty.NewFactory(), clk, log, daemonOpts...)
539	550		watchCtx, stopWatch := context.WithCancel(ctx)
540	551		defer stopWatch()
541	552		go func() {

 	return runWithLifecycleOwner(ctx, ipc.SocketDir(), platform.StateDir(), runDaemonOwned)
 }
+func constructDaemonBeforeSocketPublication(
+	construct func() *daemon.Daemon,
+	listen func() (ports.Listener, error),
+) (*daemon.Daemon, ports.Listener, error) {
+	d := construct()
+	ln, err := listen()
+	return d, ln, err
+}
 func runDaemonOwned(ctx context.Context) (retErr error) {
 	log, logCloser, err := configureLogging(logging.Daemon, true)
 	if err != nil {
 		return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())
 	}
 	log.Info("session persistence enabled", "path", storePath)
-	daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue))
+	daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))
-	// Socket publication follows lifecycle acquisition, migration, and strict
+	// Construct the catalogue-backed expected-session registry before socket
-	// recovery. Any earlier failure leaves no apparently healthy daemon.
+	// publication. Phase 3 snapshot restoration remains asynchronous in Serve.
-	ln, err := ipc.Listen(ipc.SocketDir(), ipc.WithRuntimeObserver(observer))
+	d, ln, err := constructDaemonBeforeSocketPublication(
+		func() *daemon.Daemon { return daemon.New(pty.NewFactory(), clk, log, daemonOpts...) },
+		func() (ports.Listener, error) { return ipc.Listen(ipc.SocketDir(), ipc.WithRuntimeObserver(observer)) },
+	)
 	if err != nil {
 		closeErr := opened.Catalogue.Close()
 		log.Error("daemon listen failed", "socket_dir", ipc.SocketDir(), "err", err)
 	defer func() { _ = ln.Close() }()
 	log.Info("daemon starting", "socket", ln.Addr())
-	d := daemon.New(pty.NewFactory(), clk, log, daemonOpts...)
 	watchCtx, stopWatch := context.WithCancel(ctx)
 	defer stopWatch()
 	go func() {

M internal/app/run_test.go +18 -0


@@ -21,6 +21,7 @@ import (
21	21		portsmocks "github.com/bnema/vev/internal/ports/mocks"
22	22		"github.com/bnema/vev/internal/usecase/client"
23	23		"github.com/bnema/vev/internal/usecase/confirm"
	24	+	"github.com/bnema/vev/internal/usecase/daemon"
24	25		"github.com/bnema/vev/pkg/kv"
25	26		"github.com/stretchr/testify/mock"
26	27		"github.com/stretchr/testify/require"

@@ -32,6 +33,23 @@ type fakeLifecycleOwnership struct {
32	33
33	34		func (o fakeLifecycleOwnership) Release() error { return o.release() }
34	35
	36	+	func TestCatalogueRegistryConstructionPrecedesSocketPublication(t *testing.T) {
	37	+	var events []string
	38	+	_, _, err := constructDaemonBeforeSocketPublication(
	39	+	func() *daemon.Daemon {
	40	+	events = append(events, "catalogue-registry")
	41	+	return nil
	42	+	},
	43	+	func() (ports.Listener, error) {
	44	+	require.Equal(t, []string{"catalogue-registry"}, events)
	45	+	events = append(events, "socket-publication")
	46	+	return nil, nil
	47	+	},
	48	+	)
	49	+	require.NoError(t, err)
	50	+	require.Equal(t, []string{"catalogue-registry", "socket-publication"}, events)
	51	+	}
	52	+
35	53		func TestLifecycleOwnershipPrecedesDaemonStartup(t *testing.T) {
36	54		t.Run("ownership prefix is exact", func(t *testing.T) {
37	55		var events []string


@@ -21,6 +21,7 @@ import (
21	portsmocks "github.com/bnema/vev/internal/ports/mocks"	21		portsmocks "github.com/bnema/vev/internal/ports/mocks"
22	"github.com/bnema/vev/internal/usecase/client"	22		"github.com/bnema/vev/internal/usecase/client"
23	"github.com/bnema/vev/internal/usecase/confirm"	23		"github.com/bnema/vev/internal/usecase/confirm"
		24	+	"github.com/bnema/vev/internal/usecase/daemon"
24	"github.com/bnema/vev/pkg/kv"	25		"github.com/bnema/vev/pkg/kv"
25	"github.com/stretchr/testify/mock"	26		"github.com/stretchr/testify/mock"
26	"github.com/stretchr/testify/require"	27		"github.com/stretchr/testify/require"

@@ -32,6 +33,23 @@ type fakeLifecycleOwnership struct {
32		33
33	func (o fakeLifecycleOwnership) Release() error { return o.release() }	34		func (o fakeLifecycleOwnership) Release() error { return o.release() }
34		35
		36	+	func TestCatalogueRegistryConstructionPrecedesSocketPublication(t *testing.T) {
		37	+	var events []string
		38	+	_, _, err := constructDaemonBeforeSocketPublication(
		39	+	func() *daemon.Daemon {
		40	+	events = append(events, "catalogue-registry")
		41	+	return nil
		42	+	},
		43	+	func() (ports.Listener, error) {
		44	+	require.Equal(t, []string{"catalogue-registry"}, events)
		45	+	events = append(events, "socket-publication")
		46	+	return nil, nil
		47	+	},
		48	+	)
		49	+	require.NoError(t, err)
		50	+	require.Equal(t, []string{"catalogue-registry", "socket-publication"}, events)
		51	+	}
		52	+
35	func TestLifecycleOwnershipPrecedesDaemonStartup(t *testing.T) {	53		func TestLifecycleOwnershipPrecedesDaemonStartup(t *testing.T) {
36	t.Run("ownership prefix is exact", func(t *testing.T) {	54		t.Run("ownership prefix is exact", func(t *testing.T) {
37	var events []string	55		var events []string

M internal/persist/migration_test.go +80 -15


@@ -8,6 +8,7 @@ import (
8	8		"io"
9	9		"os"
10	10		"path/filepath"
	11	+	"strconv"
11	12		"testing"
12	13
13	14		"github.com/bnema/vev/internal/domain"

@@ -22,10 +23,16 @@ type migrationStub struct {
22	23		heads map[string]domain.CheckpointRef
23	24		migrateErr map[string]error
24	25		assignments map[string]domain.IncarnationID
	26	+	sources map[string]string
25	27		}
26	28
27	29		func (s *migrationStub) HasLegacyState(context.Context) (bool, error) { return s.legacy, s.probeErr }
28	30		func (s *migrationStub) ReadLegacyHEAD(_ context.Context, name string) (domain.CheckpointRef, error) {
	31	+	if source := s.sources[name]; source != "" {
	32	+	if _, err := os.Stat(source); err != nil {
	33	+	return domain.CheckpointRef{}, err
	34	+	}
	35	+	}
29	36		ref, ok := s.heads[name]
30	37		if !ok {
31	38		return domain.CheckpointRef{}, errors.New("uncertain legacy checkpoint")

@@ -33,6 +40,11 @@ func (s *migrationStub) ReadLegacyHEAD(_ context.Context, name string) (domain.C
33	40		return ref, nil
34	41		}
35	42		func (s *migrationStub) MigrateV1Checkpoint(_ context.Context, req ports.SnapshotMigrationRequest) (domain.CheckpointRef, error) {
	43	+	if source := s.sources[req.LegacyName]; source != "" {
	44	+	if _, err := os.Stat(source); err != nil {
	45	+	return domain.CheckpointRef{}, err
	46	+	}
	47	+	}
36	48		if err := s.migrateErr[req.LegacyName]; err != nil {
37	49		return domain.CheckpointRef{}, err
38	50		}

@@ -108,42 +120,95 @@ func TestMigrationOver4096(t *testing.T) {
108	120		}
109	121
110	122		func TestMigrationResumeMatrix(t *testing.T) {
111		-	for _, boundary := range []string{"backup-sync", "intent-sync", "identity-sync", "head-validation", "catalogue-sync", "receipt-sync", "complete-sync"} {
112		-	t.Run(boundary, func(t *testing.T) {
	123	+	type crashPoint struct {
	124	+	boundary string
	125	+	occurrence int
	126	+	record string
	127	+	}
	128	+	points := []crashPoint{
	129	+	{boundary: "backup-sync", occurrence: 1},
	130	+	{boundary: "intent-sync", occurrence: 1},
	131	+	{boundary: "identity-sync", occurrence: 1, record: "alpha"},
	132	+	{boundary: "identity-sync", occurrence: 2, record: "uncertain"},
	133	+	{boundary: "identity-sync", occurrence: 3, record: "work"},
	134	+	{boundary: "head-validation", occurrence: 1, record: "alpha"},
	135	+	{boundary: "head-validation", occurrence: 2, record: "uncertain"},
	136	+	{boundary: "head-validation", occurrence: 3, record: "work"},
	137	+	{boundary: "catalogue-sync", occurrence: 1},
	138	+	{boundary: "receipt-sync", occurrence: 1},
	139	+	{boundary: "complete-sync", occurrence: 1},
	140	+	}
	141	+	for _, point := range points {
	142	+	name := point.boundary + "-" + strconv.Itoa(point.occurrence)
	143	+	if point.record != "" {
	144	+	name += "-" + point.record
	145	+	}
	146	+	t.Run(name, func(t *testing.T) {
113	147		dir := t.TempDir()
114		-	writeLegacyCatalogue(t, dir,
	148	+	legacyCatalogue := writeLegacyCatalogue(t, dir,
115	149		legacyCatalogueRecordV0{Name: "alpha", Cwd: "/tmp"},
	150	+	legacyCatalogueRecordV0{Name: "uncertain", Cwd: "/tmp"},
116	151		legacyCatalogueRecordV0{Name: "work", Cwd: "/tmp"},
117	152		)
118	153		ref := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
119		-	stub := &migrationStub{heads: map[string]domain.CheckpointRef{"alpha": ref, "work": ref}, migrateErr: map[string]error{}}
120		-	crashed := false
	154	+	heads := map[string]domain.CheckpointRef{"alpha": ref, "uncertain": ref, "work": ref}
	155	+	migrationErrors := map[string]error{"uncertain": errors.Join(ports.ErrLegacySnapshotUncertain, errors.New("corrupt legacy manifest"))}
	156	+	legacySnapshotDir := filepath.Join(dir, "legacy-snapshots")
	157	+	require.NoError(t, os.Mkdir(legacySnapshotDir, 0o700))
	158	+	legacySnapshotSources := make(map[string]string, 3)
	159	+	for _, recordName := range []string{"alpha", "uncertain", "work"} {
	160	+	path := filepath.Join(legacySnapshotDir, recordName+".snap")
	161	+	require.NoError(t, os.WriteFile(path, []byte("legacy source"), 0o600))
	162	+	legacySnapshotSources[recordName] = path
	163	+	}
	164	+	stub := &migrationStub{heads: heads, migrateErr: migrationErrors, sources: legacySnapshotSources}
	165	+	seen := 0
121	166		_, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{}, SnapshotMigration: stub, Fault: func(got string) error {
122		-	if got == boundary && !crashed {
123		-	crashed = true
	167	+	if got != point.boundary {
	168	+	return nil
	169	+	}
	170	+	seen++
	171	+	if seen == point.occurrence {
124	172		return errors.New("simulated crash")
125	173		}
126	174		return nil
127	175		}})
128	176		require.Error(t, err)
129		-	require.True(t, crashed)
	177	+	require.Equal(t, point.occurrence, seen)
130	178
131	179		before := map[string]domain.IncarnationID{}
132		-	intentPath := filepath.Join(dir, "migration", migrationIntentName)
	180	+	intentPath := filepath.Join(dir, migrationDirName, migrationIntentName)
133	181		if journal, journalErr := readMigrationRecord(intentPath); journalErr == nil {
134		-	for name, id := range journal.Assignments {
135		-	before[name] = id
	182	+	for recordName, id := range journal.Assignments {
	183	+	before[recordName] = id
136	184		}
137	185		}
138		-	restarted := &migrationStub{heads: map[string]domain.CheckpointRef{"alpha": ref, "work": ref}, migrateErr: map[string]error{}}
	186	+	restarted := &migrationStub{heads: heads, migrateErr: migrationErrors, sources: legacySnapshotSources}
139	187		result, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{next: 64}, SnapshotMigration: restarted})
140	188		require.NoError(t, err)
141		-	require.NoError(t, result.Catalogue.Close())
142	189		completed, err := readMigrationRecord(intentPath)
143	190		require.NoError(t, err)
144		-	for name, id := range before {
145		-	require.Equal(t, id, completed.Assignments[name], "assigned identities must survive restart")
	191	+	require.Len(t, completed.Assignments, 3)
	192	+	for recordName, id := range before {
	193	+	require.Equal(t, id, completed.Assignments[recordName], "assigned identities must survive restart")
	194	+	}
	195	+	for _, recordName := range []string{"alpha", "uncertain", "work"} {
	196	+	require.Equal(t, completed.Assignments[recordName], recordByName(t, result.Records, recordName).IncarnationID)
	197	+	require.FileExists(t, legacySnapshotSources[recordName], "legacy snapshot source must remain retained")
	198	+	}
	199	+	uncertain := recordByName(t, result.Records, "uncertain")
	200	+	require.Equal(t, domain.RecoveryDegraded, uncertain.RecoveryState)
	201	+	require.Contains(t, uncertain.DegradedReason, "corrupt legacy manifest")
	202	+	require.NoError(t, result.Catalogue.Close())
	203	+
	204	+	reopened, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: io.LimitReader(&countingReader{}, 0), SnapshotMigration: restarted})
	205	+	require.NoError(t, err)
	206	+	for _, recordName := range []string{"alpha", "uncertain", "work"} {
	207	+	require.Equal(t, completed.Assignments[recordName], recordByName(t, reopened.Records, recordName).IncarnationID)
146	208		}
	209	+	require.Equal(t, domain.RecoveryDegraded, recordByName(t, reopened.Records, "uncertain").RecoveryState)
	210	+	require.NoError(t, reopened.Catalogue.Close())
	211	+	require.FileExists(t, legacyCatalogue)
147	212		require.FileExists(t, filepath.Join(dir, migrationDirName, migrationBackupName))
148	213		require.FileExists(t, filepath.Join(dir, migrationDirName, migrationCompleteName))
149	214		})


@@ -8,6 +8,7 @@ import (
8		"io"	8		"io"
9		"os"	9		"os"
10		"path/filepath"	10		"path/filepath"
			11	+	"strconv"
11		"testing"	12		"testing"
12			13
13		"github.com/bnema/vev/internal/domain"	14		"github.com/bnema/vev/internal/domain"

@@ -22,10 +23,16 @@ type migrationStub struct {
22		heads map[string]domain.CheckpointRef	23		heads map[string]domain.CheckpointRef
23		migrateErr map[string]error	24		migrateErr map[string]error
24		assignments map[string]domain.IncarnationID	25		assignments map[string]domain.IncarnationID
			26	+	sources map[string]string
25		}	27		}
26			28
27		func (s *migrationStub) HasLegacyState(context.Context) (bool, error) { return s.legacy, s.probeErr }	29		func (s *migrationStub) HasLegacyState(context.Context) (bool, error) { return s.legacy, s.probeErr }
28		func (s *migrationStub) ReadLegacyHEAD(_ context.Context, name string) (domain.CheckpointRef, error) {	30		func (s *migrationStub) ReadLegacyHEAD(_ context.Context, name string) (domain.CheckpointRef, error) {
			31	+	if source := s.sources[name]; source != "" {
			32	+	if _, err := os.Stat(source); err != nil {
			33	+	return domain.CheckpointRef{}, err
			34	+	}
			35	+	}
29		ref, ok := s.heads[name]	36		ref, ok := s.heads[name]
30		if !ok {	37		if !ok {
31		return domain.CheckpointRef{}, errors.New("uncertain legacy checkpoint")	38		return domain.CheckpointRef{}, errors.New("uncertain legacy checkpoint")

@@ -33,6 +40,11 @@ func (s *migrationStub) ReadLegacyHEAD(_ context.Context, name string) (domain.C
33		return ref, nil	40		return ref, nil
34		}	41		}
35		func (s *migrationStub) MigrateV1Checkpoint(_ context.Context, req ports.SnapshotMigrationRequest) (domain.CheckpointRef, error) {	42		func (s *migrationStub) MigrateV1Checkpoint(_ context.Context, req ports.SnapshotMigrationRequest) (domain.CheckpointRef, error) {
			43	+	if source := s.sources[req.LegacyName]; source != "" {
			44	+	if _, err := os.Stat(source); err != nil {
			45	+	return domain.CheckpointRef{}, err
			46	+	}
			47	+	}
36		if err := s.migrateErr[req.LegacyName]; err != nil {	48		if err := s.migrateErr[req.LegacyName]; err != nil {
37		return domain.CheckpointRef{}, err	49		return domain.CheckpointRef{}, err
38		}	50		}

@@ -108,42 +120,95 @@ func TestMigrationOver4096(t *testing.T) {
108		}	120		}
109			121
110		func TestMigrationResumeMatrix(t *testing.T) {	122		func TestMigrationResumeMatrix(t *testing.T) {
111	-	for _, boundary := range []string{"backup-sync", "intent-sync", "identity-sync", "head-validation", "catalogue-sync", "receipt-sync", "complete-sync"} {	123	+	type crashPoint struct {
112	-	t.Run(boundary, func(t *testing.T) {	124	+	boundary string
			125	+	occurrence int
			126	+	record string
			127	+	}
			128	+	points := []crashPoint{
			129	+	{boundary: "backup-sync", occurrence: 1},
			130	+	{boundary: "intent-sync", occurrence: 1},
			131	+	{boundary: "identity-sync", occurrence: 1, record: "alpha"},
			132	+	{boundary: "identity-sync", occurrence: 2, record: "uncertain"},
			133	+	{boundary: "identity-sync", occurrence: 3, record: "work"},
			134	+	{boundary: "head-validation", occurrence: 1, record: "alpha"},
			135	+	{boundary: "head-validation", occurrence: 2, record: "uncertain"},
			136	+	{boundary: "head-validation", occurrence: 3, record: "work"},
			137	+	{boundary: "catalogue-sync", occurrence: 1},
			138	+	{boundary: "receipt-sync", occurrence: 1},
			139	+	{boundary: "complete-sync", occurrence: 1},
			140	+	}
			141	+	for _, point := range points {
			142	+	name := point.boundary + "-" + strconv.Itoa(point.occurrence)
			143	+	if point.record != "" {
			144	+	name += "-" + point.record
			145	+	}
			146	+	t.Run(name, func(t *testing.T) {
113		dir := t.TempDir()	147		dir := t.TempDir()
114	-	writeLegacyCatalogue(t, dir,	148	+	legacyCatalogue := writeLegacyCatalogue(t, dir,
115		legacyCatalogueRecordV0{Name: "alpha", Cwd: "/tmp"},	149		legacyCatalogueRecordV0{Name: "alpha", Cwd: "/tmp"},
			150	+	legacyCatalogueRecordV0{Name: "uncertain", Cwd: "/tmp"},
116		legacyCatalogueRecordV0{Name: "work", Cwd: "/tmp"},	151		legacyCatalogueRecordV0{Name: "work", Cwd: "/tmp"},
117		)	152		)
118		ref := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}	153		ref := domain.CheckpointRef{Generation: 1, ManifestDigest: [32]byte{1}}
119	-	stub := &migrationStub{heads: map[string]domain.CheckpointRef{"alpha": ref, "work": ref}, migrateErr: map[string]error{}}	154	+	heads := map[string]domain.CheckpointRef{"alpha": ref, "uncertain": ref, "work": ref}
120	-	crashed := false	155	+	migrationErrors := map[string]error{"uncertain": errors.Join(ports.ErrLegacySnapshotUncertain, errors.New("corrupt legacy manifest"))}
			156	+	legacySnapshotDir := filepath.Join(dir, "legacy-snapshots")
			157	+	require.NoError(t, os.Mkdir(legacySnapshotDir, 0o700))
			158	+	legacySnapshotSources := make(map[string]string, 3)
			159	+	for _, recordName := range []string{"alpha", "uncertain", "work"} {
			160	+	path := filepath.Join(legacySnapshotDir, recordName+".snap")
			161	+	require.NoError(t, os.WriteFile(path, []byte("legacy source"), 0o600))
			162	+	legacySnapshotSources[recordName] = path
			163	+	}
			164	+	stub := &migrationStub{heads: heads, migrateErr: migrationErrors, sources: legacySnapshotSources}
			165	+	seen := 0
121		_, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{}, SnapshotMigration: stub, Fault: func(got string) error {	166		_, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{}, SnapshotMigration: stub, Fault: func(got string) error {
122	-	if got == boundary && !crashed {	167	+	if got != point.boundary {
123	-	crashed = true	168	+	return nil
			169	+	}
			170	+	seen++
			171	+	if seen == point.occurrence {
124		return errors.New("simulated crash")	172		return errors.New("simulated crash")
125		}	173		}
126		return nil	174		return nil
127		}})	175		}})
128		require.Error(t, err)	176		require.Error(t, err)
129	-	require.True(t, crashed)	177	+	require.Equal(t, point.occurrence, seen)
130			178
131		before := map[string]domain.IncarnationID{}	179		before := map[string]domain.IncarnationID{}
132	-	intentPath := filepath.Join(dir, "migration", migrationIntentName)	180	+	intentPath := filepath.Join(dir, migrationDirName, migrationIntentName)
133		if journal, journalErr := readMigrationRecord(intentPath); journalErr == nil {	181		if journal, journalErr := readMigrationRecord(intentPath); journalErr == nil {
134	-	for name, id := range journal.Assignments {	182	+	for recordName, id := range journal.Assignments {
135	-	before[name] = id	183	+	before[recordName] = id
136		}	184		}
137		}	185		}
138	-	restarted := &migrationStub{heads: map[string]domain.CheckpointRef{"alpha": ref, "work": ref}, migrateErr: map[string]error{}}	186	+	restarted := &migrationStub{heads: heads, migrateErr: migrationErrors, sources: legacySnapshotSources}
139		result, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{next: 64}, SnapshotMigration: restarted})	187		result, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: &countingReader{next: 64}, SnapshotMigration: restarted})
140		require.NoError(t, err)	188		require.NoError(t, err)
141	-	require.NoError(t, result.Catalogue.Close())
142		completed, err := readMigrationRecord(intentPath)	189		completed, err := readMigrationRecord(intentPath)
143		require.NoError(t, err)	190		require.NoError(t, err)
144	-	for name, id := range before {	191	+	require.Len(t, completed.Assignments, 3)
145	-	require.Equal(t, id, completed.Assignments[name], "assigned identities must survive restart")	192	+	for recordName, id := range before {
			193	+	require.Equal(t, id, completed.Assignments[recordName], "assigned identities must survive restart")
			194	+	}
			195	+	for _, recordName := range []string{"alpha", "uncertain", "work"} {
			196	+	require.Equal(t, completed.Assignments[recordName], recordByName(t, result.Records, recordName).IncarnationID)
			197	+	require.FileExists(t, legacySnapshotSources[recordName], "legacy snapshot source must remain retained")
			198	+	}
			199	+	uncertain := recordByName(t, result.Records, "uncertain")
			200	+	require.Equal(t, domain.RecoveryDegraded, uncertain.RecoveryState)
			201	+	require.Contains(t, uncertain.DegradedReason, "corrupt legacy manifest")
			202	+	require.NoError(t, result.Catalogue.Close())
			203	+
			204	+	reopened, err := OpenOrMigrate(context.Background(), OpenDeps{StateDir: dir, Random: io.LimitReader(&countingReader{}, 0), SnapshotMigration: restarted})
			205	+	require.NoError(t, err)
			206	+	for _, recordName := range []string{"alpha", "uncertain", "work"} {
			207	+	require.Equal(t, completed.Assignments[recordName], recordByName(t, reopened.Records, recordName).IncarnationID)
146		}	208		}
			209	+	require.Equal(t, domain.RecoveryDegraded, recordByName(t, reopened.Records, "uncertain").RecoveryState)
			210	+	require.NoError(t, reopened.Catalogue.Close())
			211	+	require.FileExists(t, legacyCatalogue)
147		require.FileExists(t, filepath.Join(dir, migrationDirName, migrationBackupName))	212		require.FileExists(t, filepath.Join(dir, migrationDirName, migrationBackupName))
148		require.FileExists(t, filepath.Join(dir, migrationDirName, migrationCompleteName))	213		require.FileExists(t, filepath.Join(dir, migrationDirName, migrationCompleteName))
149		})	214		})

M internal/usecase/daemon/daemon.go +30 -20


@@ -119,6 +119,8 @@ type Daemon struct {
119	119		shellOverride bool
120	120		persist *persist.Persister
121	121		persistEnabled bool
	122	+	catalogueRecords []domain.CatalogueRecord
	123	+	catalogueRecordsProvided bool
122	124		// snapshotRepository is the sole checkpoint contract. legacySnapshots is
123	125		// read-only migration input and is never used for new writes.
124	126		snapshotRepository ports.SnapshotRepository

@@ -247,11 +249,14 @@ func WithStore(store ports.Store) Option {
247	249		}
248	250		}
249	251
250		-	// WithCatalogue installs the singular catalogue opened by startup.
251		-	func WithCatalogue(catalogue *persist.Persister) Option {
	252	+	// WithCatalogue installs the singular catalogue and the strictly opened
	253	+	// records that define the daemon's expected-session registry at startup.
	254	+	func WithCatalogue(catalogue *persist.Persister, records []domain.CatalogueRecord) Option {
252	255		return func(d *Daemon) {
253	256		d.persist = catalogue
254	257		d.persistEnabled = catalogue != nil
	258	+	d.catalogueRecords = append([]domain.CatalogueRecord(nil), records...)
	259	+	d.catalogueRecordsProvided = true
255	260		}
256	261		}
257	262

@@ -431,27 +436,32 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt
431	436		allow := buildRestoreProcessAllowlist(domain.DefaultSnapshotRestoreProcesses())
432	437		d.restoreProcessAllowlist.Store(&allow)
433	438		}
434		-	if records, err := d.persist.LoadAll(); err != nil {
435		-	d.log.Warn("loading persisted sessions failed", "err", err)
436		-	} else {
437		-	var maxSeq uint64
438		-	var maxCreatedAt int64
439		-	hasCreatedAt := false
440		-	for _, r := range records {
441		-	d.stopped[r.Name] = stoppedSession{name: r.Name, cwd: r.Cwd, createdAt: r.CreatedAt, incarnation: r.IncarnationID, lastUsedSeq: r.LastUsedSeq, tabNames: r.TabNames}
442		-	if !hasCreatedAt \|\| r.CreatedAt > maxCreatedAt {
443		-	maxCreatedAt = r.CreatedAt
444		-	hasCreatedAt = true
445		-	}
446		-	if r.LastUsedSeq > maxSeq {
447		-	maxSeq = r.LastUsedSeq
448		-	}
	439	+	records := d.catalogueRecords
	440	+	if !d.catalogueRecordsProvided {
	441	+	var err error
	442	+	records, err = d.persist.LoadAll()
	443	+	if err != nil {
	444	+	d.log.Warn("loading persisted sessions failed", "err", err)
	445	+	return d
449	446		}
450		-	if hasCreatedAt {
451		-	d.lastAllocatedCreatedAt = maxCreatedAt
	447	+	}
	448	+	var maxSeq uint64
	449	+	var maxCreatedAt int64
	450	+	hasCreatedAt := false
	451	+	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...)}
	453	+	if !hasCreatedAt \|\| r.CreatedAt > maxCreatedAt {
	454	+	maxCreatedAt = r.CreatedAt
	455	+	hasCreatedAt = true
	456	+	}
	457	+	if r.LastUsedSeq > maxSeq {
	458	+	maxSeq = r.LastUsedSeq
452	459		}
453		-	d.mruSeq.Store(maxSeq)
454	460		}
	461	+	if hasCreatedAt {
	462	+	d.lastAllocatedCreatedAt = maxCreatedAt
	463	+	}
	464	+	d.mruSeq.Store(maxSeq)
455	465		return d
456	466		}
457	467

 	shellOverride                  bool
 	persist                        *persist.Persister
 	persistEnabled                 bool
+	catalogueRecords               []domain.CatalogueRecord
+	catalogueRecordsProvided       bool
 	// snapshotRepository is the sole checkpoint contract. legacySnapshots is
 	// read-only migration input and is never used for new writes.
 	snapshotRepository ports.SnapshotRepository
 	}
 }
-// WithCatalogue installs the singular catalogue opened by startup.
+// WithCatalogue installs the singular catalogue and the strictly opened
-func WithCatalogue(catalogue *persist.Persister) Option {
+// records that define the daemon's expected-session registry at startup.
+func WithCatalogue(catalogue *persist.Persister, records []domain.CatalogueRecord) Option {
 	return func(d *Daemon) {
 		d.persist = catalogue
 		d.persistEnabled = catalogue != nil
+		d.catalogueRecords = append([]domain.CatalogueRecord(nil), records...)
+		d.catalogueRecordsProvided = true
 	}
 }
 		allow := buildRestoreProcessAllowlist(domain.DefaultSnapshotRestoreProcesses())
 		d.restoreProcessAllowlist.Store(&allow)
 	}
-	if records, err := d.persist.LoadAll(); err != nil {
+	records := d.catalogueRecords
-		d.log.Warn("loading persisted sessions failed", "err", err)
+	if !d.catalogueRecordsProvided {
-	} else {
+		var err error
-		var maxSeq uint64
+		records, err = d.persist.LoadAll()
-		var maxCreatedAt int64
+		if err != nil {
-		hasCreatedAt := false
+			d.log.Warn("loading persisted sessions failed", "err", err)
-		for _, r := range records {
+			return d
-			d.stopped[r.Name] = stoppedSession{name: r.Name, cwd: r.Cwd, createdAt: r.CreatedAt, incarnation: r.IncarnationID, lastUsedSeq: r.LastUsedSeq, tabNames: r.TabNames}
-			if !hasCreatedAt || r.CreatedAt > maxCreatedAt {
-				maxCreatedAt = r.CreatedAt
-				hasCreatedAt = true
-			}
-			if r.LastUsedSeq > maxSeq {
-				maxSeq = r.LastUsedSeq
-			}
 		}
-		if hasCreatedAt {
+	}
-			d.lastAllocatedCreatedAt = maxCreatedAt
+	var maxSeq uint64
+	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...)}
+		if !hasCreatedAt || r.CreatedAt > maxCreatedAt {
+			maxCreatedAt = r.CreatedAt
+			hasCreatedAt = true
+		}
+		if r.LastUsedSeq > maxSeq {
+			maxSeq = r.LastUsedSeq
 		}
-		d.mruSeq.Store(maxSeq)
 	}
+	if hasCreatedAt {
+		d.lastAllocatedCreatedAt = maxCreatedAt
+	}
+	d.mruSeq.Store(maxSeq)
 	return d
 }

M internal/usecase/daemon/session_test.go +14 -0


@@ -1765,6 +1765,20 @@ func TestNamedSessionLifecycleExhaustionDoesNotMutateSessionState(t *testing.T)
1765	1765		require.Equal(t, stoppedSession{name: "retained", cwd: "/tmp", createdAt: 9}, d.stopped["retained"])
1766	1766		}
1767	1767
	1768	+	func TestCatalogueRecordsConstructExpectedSessionRegistry(t *testing.T) {
	1769	+	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	+	}
	1773	+
	1774	+	d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(persist.New(nil), records))
	1775	+
	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	+	require.Equal(t, int64(9), d.lastAllocatedCreatedAt)
	1779	+	require.Equal(t, uint64(13), d.mruSeq.Load())
	1780	+	}
	1781	+
1768	1782		func TestNamedSessionLifecycleTimestampStartsAfterPersistedHighWaterMark(t *testing.T) {
1769	1783		store, _ := newMockStore(t)
1770	1784		seed := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithStore(store))


@@ -1765,6 +1765,20 @@ func TestNamedSessionLifecycleExhaustionDoesNotMutateSessionState(t *testing.T)
1765	require.Equal(t, stoppedSession{name: "retained", cwd: "/tmp", createdAt: 9}, d.stopped["retained"])	1765		require.Equal(t, stoppedSession{name: "retained", cwd: "/tmp", createdAt: 9}, d.stopped["retained"])
1766	}	1766		}
1767		1767
		1768	+	func TestCatalogueRecordsConstructExpectedSessionRegistry(t *testing.T) {
		1769	+	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	+	}
		1773	+
		1774	+	d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithCatalogue(persist.New(nil), records))
		1775	+
		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	+	require.Equal(t, int64(9), d.lastAllocatedCreatedAt)
		1779	+	require.Equal(t, uint64(13), d.mruSeq.Load())
		1780	+	}
		1781	+
1768	func TestNamedSessionLifecycleTimestampStartsAfterPersistedHighWaterMark(t *testing.T) {	1782		func TestNamedSessionLifecycleTimestampStartsAfterPersistedHighWaterMark(t *testing.T) {
1769	store, _ := newMockStore(t)	1783		store, _ := newMockStore(t)
1770	seed := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithStore(store))	1784		seed := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)), WithStore(store))