docs(recovery): document and observe durable state

commit: 2d93e51ec1c9c5dbd0e0652f02f81fb50d59747b
author: brice · 2mo ago · 2026-07-26 02:02 UTC
parent: df0f462
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

11 files · +441 -13

M docs/configuration.md +8 -0


@@ -74,6 +74,14 @@ code.detach = DET
74	74
75	75		Invalid values log a warning and resolve that setting to its default on both initial load and reload.
76	76
	77	+	## Logs and durable state
	78	+
	79	+	Set `VEV_LOG=debug`, `VEV_LOG=warn`, or `VEV_LOG=error` to change verbosity; the default is `info`. JSON-line logs such as `vev-daemon.log` live in `$XDG_STATE_HOME/vev`, or `~/.local/state/vev` when unset. The same state directory contains the strict session catalogue, migration journal, notices, and `snapshots/`. The lifecycle lock and socket live in `$XDG_RUNTIME_DIR/vev` (with platform runtime fallbacks).
	80	+
	81	+	Recovery events include `lifecycle_owner_wait`, `lifecycle_owner_acquired`, `lifecycle_owner_released`, `catalogue_validated`, `catalogue_compaction_recovery_complete`, `session_restore_complete`, `fallback_checkpoint_promoted`, `snapshot_head_repair_complete`, `session_degraded`, `snapshot_maintenance_progress`, `interrupted_transaction_recovery_complete`, and `daemon_startup_complete`.
	82	+
	83	+	Catalogue failure is fail-closed: vev does not publish an empty replacement daemon. Preserve the state directory, inspect `catalogue_validation_failed`, correct storage or ownership problems, and retry without editing catalogue files. See [Durable session recovery](durable-session-recovery.md) for explicit recovery commands, migration, diagnostics, and the committed checkpoint plus up to two direct fallbacks retention policy.
	84	+
77	85		## Theme
78	86
79	87		With `theme = auto`, vev follows the terminal's reported foreground, background, light/dark scheme, and ANSI palette. `theme.palette = on` is the default. `theme.accent = auto` derives one accent from the terminal's chromatic ANSI colors; repeated terminal colors are preferred, otherwise vev uses an eligible blue slot when available. `theme.accent = 0` through `theme.accent = 15` selects exactly that ANSI slot. Arbitrary RGB values and `off` are not accepted for `theme.accent`.


@@ -74,6 +74,14 @@ code.detach = DET
74		74
75	Invalid values log a warning and resolve that setting to its default on both initial load and reload.	75		Invalid values log a warning and resolve that setting to its default on both initial load and reload.
76		76
		77	+	## Logs and durable state
		78	+
		79	+	Set `VEV_LOG=debug`, `VEV_LOG=warn`, or `VEV_LOG=error` to change verbosity; the default is `info`. JSON-line logs such as `vev-daemon.log` live in `$XDG_STATE_HOME/vev`, or `~/.local/state/vev` when unset. The same state directory contains the strict session catalogue, migration journal, notices, and `snapshots/`. The lifecycle lock and socket live in `$XDG_RUNTIME_DIR/vev` (with platform runtime fallbacks).
		80	+
		81	+	Recovery events include `lifecycle_owner_wait`, `lifecycle_owner_acquired`, `lifecycle_owner_released`, `catalogue_validated`, `catalogue_compaction_recovery_complete`, `session_restore_complete`, `fallback_checkpoint_promoted`, `snapshot_head_repair_complete`, `session_degraded`, `snapshot_maintenance_progress`, `interrupted_transaction_recovery_complete`, and `daemon_startup_complete`.
		82	+
		83	+	Catalogue failure is fail-closed: vev does not publish an empty replacement daemon. Preserve the state directory, inspect `catalogue_validation_failed`, correct storage or ownership problems, and retry without editing catalogue files. See [Durable session recovery](durable-session-recovery.md) for explicit recovery commands, migration, diagnostics, and the committed checkpoint plus up to two direct fallbacks retention policy.
		84	+
77	## Theme	85		## Theme
78		86
79	With `theme = auto`, vev follows the terminal's reported foreground, background, light/dark scheme, and ANSI palette. `theme.palette = on` is the default. `theme.accent = auto` derives one accent from the terminal's chromatic ANSI colors; repeated terminal colors are preferred, otherwise vev uses an eligible blue slot when available. `theme.accent = 0` through `theme.accent = 15` selects exactly that ANSI slot. Arbitrary RGB values and `off` are not accepted for `theme.accent`.	87		With `theme = auto`, vev follows the terminal's reported foreground, background, light/dark scheme, and ANSI palette. `theme.palette = on` is the default. `theme.accent = auto` derives one accent from the terminal's chromatic ANSI colors; repeated terminal colors are preferred, otherwise vev uses an eligible blue slot when available. `theme.accent = 0` through `theme.accent = 15` selects exactly that ANSI slot. Arbitrary RGB values and `off` are not accepted for `theme.accent`.

A docs/durable-session-recovery.md +32 -0

@@ -0,0 +1,32 @@
	1	+	# Durable session recovery
	2	+
	3	+	vev opens named-session state only while holding `$XDG_RUNTIME_DIR/vev/lifecycle.lock` (or the platform runtime fallback). Catalogue corruption prevents socket publication; vev never substitutes an empty healthy daemon.
	4	+
	5	+	## Session states
	6	+
	7	+	`stopped` is safe fresh metadata, `restoring` is validation in progress, and `degraded` preserves uncertain state without starting a replacement shell. A degraded session remains visible in `vev ls`, but attach is refused until an explicit recovery action succeeds.
	8	+
	9	+	## Recovery commands
	10	+
	11	+	- `vev cmd -s NAME session-recovery retry`
	12	+	- `vev cmd -s NAME session-recovery restore GENERATION`
	13	+	- `vev cmd -s NAME session-recovery export PATH`
	14	+	- `vev cmd -s NAME session-recovery discard`
	15	+
	16	+	Retry validates persisted state without replacing it. Restore validates and promotes the selected catalogue fallback. Export writes recoverable data without mutation. Discard creates a new incarnation and retains the old record and snapshots under `quarantine/` until an explicit later purge.
	17	+
	18	+	## Migration and retention
	19	+
	20	+	The v0.x migration is resumable, retains legacy sources, and validates each referenced HEAD. Healthy sessions retain the committed checkpoint and up to two direct fallbacks; degraded, deleting, migrating, and unresolved transactions remain pinned.
	21	+
	22	+	## Fail-closed startup
	23	+
	24	+	If the catalogue cannot be opened or validated, the daemon does not publish its socket. Do not delete or edit state files. Check `vev-daemon.log` for `catalogue_validation_failed`, preserve the complete state directory, correct filesystem ownership or capacity problems, and retry. For corruption, make a backup before attempting recovery or exporting data.
	25	+
	26	+	A `lifecycle_owner_wait` event normally means another daemon is initializing or shutting down. Wait for ownership transfer; only terminate the prior process when its identity is known. The operating system releases `lifecycle.lock` when that process exits.
	27	+
	28	+	## Diagnostics
	29	+
	30	+	Durable state and recovery journals are under `$XDG_STATE_HOME/vev` (default `~/.local/state/vev`). Snapshot quarantine is under its `snapshots/quarantine/` tree. Runtime ownership is under `$XDG_RUNTIME_DIR/vev`; logs are JSON lines in the state directory.
	31	+
	32	+	Use session names, incarnation IDs, generation numbers, reason codes, and cursors from recovery events when diagnosing a failure. Recovery logs never include terminal or snapshot object contents.

@@ -0,0 +1,32 @@
	1	+	# Durable session recovery
	2	+
	3	+	vev opens named-session state only while holding `$XDG_RUNTIME_DIR/vev/lifecycle.lock` (or the platform runtime fallback). Catalogue corruption prevents socket publication; vev never substitutes an empty healthy daemon.
	4	+
	5	+	## Session states
	6	+
	7	+	`stopped` is safe fresh metadata, `restoring` is validation in progress, and `degraded` preserves uncertain state without starting a replacement shell. A degraded session remains visible in `vev ls`, but attach is refused until an explicit recovery action succeeds.
	8	+
	9	+	## Recovery commands
	10	+
	11	+	- `vev cmd -s NAME session-recovery retry`
	12	+	- `vev cmd -s NAME session-recovery restore GENERATION`
	13	+	- `vev cmd -s NAME session-recovery export PATH`
	14	+	- `vev cmd -s NAME session-recovery discard`
	15	+
	16	+	Retry validates persisted state without replacing it. Restore validates and promotes the selected catalogue fallback. Export writes recoverable data without mutation. Discard creates a new incarnation and retains the old record and snapshots under `quarantine/` until an explicit later purge.
	17	+
	18	+	## Migration and retention
	19	+
	20	+	The v0.x migration is resumable, retains legacy sources, and validates each referenced HEAD. Healthy sessions retain the committed checkpoint and up to two direct fallbacks; degraded, deleting, migrating, and unresolved transactions remain pinned.
	21	+
	22	+	## Fail-closed startup
	23	+
	24	+	If the catalogue cannot be opened or validated, the daemon does not publish its socket. Do not delete or edit state files. Check `vev-daemon.log` for `catalogue_validation_failed`, preserve the complete state directory, correct filesystem ownership or capacity problems, and retry. For corruption, make a backup before attempting recovery or exporting data.
	25	+
	26	+	A `lifecycle_owner_wait` event normally means another daemon is initializing or shutting down. Wait for ownership transfer; only terminate the prior process when its identity is known. The operating system releases `lifecycle.lock` when that process exits.
	27	+
	28	+	## Diagnostics
	29	+
	30	+	Durable state and recovery journals are under `$XDG_STATE_HOME/vev` (default `~/.local/state/vev`). Snapshot quarantine is under its `snapshots/quarantine/` tree. Runtime ownership is under `$XDG_RUNTIME_DIR/vev`; logs are JSON lines in the state directory.
	31	+
	32	+	Use session names, incarnation IDs, generation numbers, reason codes, and cursors from recovery events when diagnosing a failure. Recovery logs never include terminal or snapshot object contents.

M internal/adapters/snapshot/maintenance_gc.go +9 -1


@@ -370,7 +370,7 @@ func keepSet(plan ports.RetentionPlan) map[uint64]ports.SnapshotDigest {
370	370
371	371		// MaintainSession incrementally validates and retains only catalogue-indexed
372	372		// checkpoints. Every incarnation owns its budget and continuation state.
373		-	func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPlan, budget ports.MaintenanceBudget) (bool, error) {
	373	+	func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPlan, budget ports.MaintenanceBudget) (done bool, err error) {
374	374		if err := ctx.Err(); err != nil {
375	375		return false, err
376	376		}

@@ -378,6 +378,14 @@ func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPl
378	378		if err != nil {
379	379		return false, err
380	380		}
	381	+	defer func() {
	382	+	r.log.Info("snapshot_maintenance_progress",
	383	+	"incarnation", plan.IncarnationID.String(),
	384	+	"retained", len(plan.Keep),
	385	+	"cursor", key,
	386	+	"budget_exhausted", !done && err == nil,
	387	+	)
	388	+	}()
381	389		if plan.PinAll {
382	390		r.maintenanceMu.Lock()
383	391		r.clearRetentionMaintenance(key)

 // MaintainSession incrementally validates and retains only catalogue-indexed
 // checkpoints. Every incarnation owns its budget and continuation state.
-func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPlan, budget ports.MaintenanceBudget) (bool, error) {
+func (r *Repository) MaintainSession(ctx context.Context, plan ports.RetentionPlan, budget ports.MaintenanceBudget) (done bool, err error) {
 	if err := ctx.Err(); err != nil {
 		return false, err
 	}
 	if err != nil {
 		return false, err
 	}
+	defer func() {
+		r.log.Info("snapshot_maintenance_progress",
+			"incarnation", plan.IncarnationID.String(),
+			"retained", len(plan.Keep),
+			"cursor", key,
+			"budget_exhausted", !done && err == nil,
+		)
+	}()
 	if plan.PinAll {
 		r.maintenanceMu.Lock()
 		r.clearRetentionMaintenance(key)

A internal/adapters/snapshot/recovery_observability_test.go +38 -0

@@ -0,0 +1,38 @@
	1	+	package snapshot
	2	+
	3	+	import (
	4	+	"bufio"
	5	+	"bytes"
	6	+	"context"
	7	+	"encoding/json"
	8	+	"log/slog"
	9	+	"testing"
	10	+
	11	+	"github.com/bnema/vev/internal/domain"
	12	+	"github.com/bnema/vev/internal/ports"
	13	+	"github.com/stretchr/testify/require"
	14	+	)
	15	+
	16	+	func TestGCBudgetDiagnostic(t *testing.T) {
	17	+	var buffer bytes.Buffer
	18	+	log := slog.New(slog.NewJSONHandler(&buffer, nil))
	19	+	repository := NewRepository(t.TempDir(), log)
	20	+	id := domain.IncarnationID{1}
	21	+	done, err := repository.MaintainSession(context.Background(), ports.RetentionPlan{
	22	+	IncarnationID: id,
	23	+	Keep: []ports.CheckpointRef{{Generation: 3}, {Generation: 2}, {Generation: 1}},
	24	+	}, ports.MaintenanceBudget{})
	25	+	require.NoError(t, err)
	26	+	require.False(t, done)
	27	+
	28	+	scanner := bufio.NewScanner(bytes.NewReader(buffer.Bytes()))
	29	+	require.True(t, scanner.Scan())
	30	+	var entry map[string]any
	31	+	require.NoError(t, json.Unmarshal(scanner.Bytes(), &entry))
	32	+	require.NoError(t, scanner.Err())
	33	+	require.Equal(t, "snapshot_maintenance_progress", entry["msg"])
	34	+	require.Equal(t, id.String(), entry["incarnation"])
	35	+	require.EqualValues(t, 3, entry["retained"])
	36	+	require.Contains(t, entry, "cursor")
	37	+	require.Equal(t, true, entry["budget_exhausted"])
	38	+	}

@@ -0,0 +1,38 @@
	1	+	package snapshot
	2	+
	3	+	import (
	4	+	"bufio"
	5	+	"bytes"
	6	+	"context"
	7	+	"encoding/json"
	8	+	"log/slog"
	9	+	"testing"
	10	+
	11	+	"github.com/bnema/vev/internal/domain"
	12	+	"github.com/bnema/vev/internal/ports"
	13	+	"github.com/stretchr/testify/require"
	14	+	)
	15	+
	16	+	func TestGCBudgetDiagnostic(t *testing.T) {
	17	+	var buffer bytes.Buffer
	18	+	log := slog.New(slog.NewJSONHandler(&buffer, nil))
	19	+	repository := NewRepository(t.TempDir(), log)
	20	+	id := domain.IncarnationID{1}
	21	+	done, err := repository.MaintainSession(context.Background(), ports.RetentionPlan{
	22	+	IncarnationID: id,
	23	+	Keep: []ports.CheckpointRef{{Generation: 3}, {Generation: 2}, {Generation: 1}},
	24	+	}, ports.MaintenanceBudget{})
	25	+	require.NoError(t, err)
	26	+	require.False(t, done)
	27	+
	28	+	scanner := bufio.NewScanner(bytes.NewReader(buffer.Bytes()))
	29	+	require.True(t, scanner.Scan())
	30	+	var entry map[string]any
	31	+	require.NoError(t, json.Unmarshal(scanner.Bytes(), &entry))
	32	+	require.NoError(t, scanner.Err())
	33	+	require.Equal(t, "snapshot_maintenance_progress", entry["msg"])
	34	+	require.Equal(t, id.String(), entry["incarnation"])
	35	+	require.EqualValues(t, 3, entry["retained"])
	36	+	require.Contains(t, entry, "cursor")
	37	+	require.Equal(t, true, entry["budget_exhausted"])
	38	+	}

M internal/adapters/snapshot/repository.go +8 -1

@@ -1,6 +1,7 @@
1	1		package snapshot
2	2
3	3		import (
	4	+	"log/slog"
4	5		"sync"
5	6
6	7		"github.com/bnema/vev/internal/ports"

@@ -24,6 +25,7 @@ const (
24	25		type Repository struct {
25	26		dir string
26	27		hooks repositoryHooks
	28	+	log *slog.Logger
27	29
28	30		// sessionStateMu owns both maps. A caller retains a reference before it
29	31		// waits on a session mutex, so an idle entry can never be removed while a

@@ -95,9 +97,14 @@ var _ ports.LegacySnapshotSource = (*Repository)(nil)
95	97
96	98		// NewRepository creates a repository rooted at dir. It does not create files
97	99		// until the first publication, so merely constructing it is side-effect free.
98		-	func NewRepository(dir string) *Repository {
	100	+	func NewRepository(dir string, logs ...slog.Logger) Repository {
	101	+	log := slog.Default()
	102	+	if len(logs) > 0 && logs[0] != nil {
	103	+	log = logs[0]
	104	+	}
99	105		return &Repository{
100	106		dir: dir,
	107	+	log: log,
101	108		locks: make(map[string]*sessionMutex),
102	109		storageEpochs: make(map[string]uint64),
103	110		maintenanceCursors: make(map[string]*maintenanceCursor),

 package snapshot
 import (
+	"log/slog"
 	"sync"
 	"github.com/bnema/vev/internal/ports"
 type Repository struct {
 	dir   string
 	hooks repositoryHooks
+	log   *slog.Logger
 	// sessionStateMu owns both maps. A caller retains a reference before it
 	// waits on a session mutex, so an idle entry can never be removed while a
 // NewRepository creates a repository rooted at dir. It does not create files
 // until the first publication, so merely constructing it is side-effect free.
-func NewRepository(dir string) *Repository {
+func NewRepository(dir string, logs ...*slog.Logger) *Repository {
+	log := slog.Default()
+	if len(logs) > 0 && logs[0] != nil {
+		log = logs[0]
+	}
 	return &Repository{
 		dir:                 dir,
+		log:                 log,
 		locks:               make(map[string]*sessionMutex),
 		storageEpochs:       make(map[string]uint64),
 		maintenanceCursors:  make(map[string]*maintenanceCursor),

M internal/app/run.go +129 -7


@@ -426,6 +426,7 @@ func joinLifecycleReleaseError(retErr *error, owner lifecycleOwnership) {
426	426		type lifecycleStartupDeps struct {
427	427		ensurePrivate func(string) error
428	428		acquire func(context.Context, string, time.Duration) (lifecycleOwnership, error)
	429	+	log *slog.Logger
429	430		}
430	431
431	432		const lifecycleAcquireRetry = 25 * time.Millisecond

@@ -440,18 +441,32 @@ func runWithLifecycleOwner(ctx context.Context, runtimeDir, stateDir string, sta
440	441		}
441	442
442	443		func runWithLifecycleOwnerDeps(ctx context.Context, runtimeDir, stateDir string, start ownedDaemonStart, deps lifecycleStartupDeps) (retErr error) {
	444	+	log := deps.log
	445	+	if log == nil {
	446	+	log = slog.Default()
	447	+	}
443	448		if err := deps.ensurePrivate(runtimeDir); err != nil {
444	449		return fmt.Errorf("vev: secure runtime directory: %w", err)
445	450		}
446	451		if err := deps.ensurePrivate(stateDir); err != nil {
447	452		return fmt.Errorf("vev: secure state directory: %w", err)
448	453		}
	454	+	lockPath := lifecycle.Path(runtimeDir)
	455	+	log.Info("lifecycle_owner_wait", "path", lockPath)
449	456		owner, err := deps.acquire(ctx, runtimeDir, lifecycleAcquireRetry)
450	457		if err != nil {
	458	+	log.Warn("lifecycle_owner_wait_failed", "path", lockPath, "reason_code", "acquire-failed")
451	459		return fmt.Errorf("vev: acquire lifecycle ownership: %w", err)
452	460		}
	461	+	log.Info("lifecycle_owner_acquired", "path", lockPath)
453	462		defer func() {
454		-	retErr = errors.Join(retErr, owner.Release())
	463	+	releaseErr := owner.Release()
	464	+	if releaseErr != nil {
	465	+	log.Error("lifecycle_owner_release_failed", "path", lockPath, "reason_code", "release-failed")
	466	+	} else {
	467	+	log.Info("lifecycle_owner_released", "path", lockPath)
	468	+	}
	469	+	retErr = errors.Join(retErr, releaseErr)
455	470		}()
456	471		if start == nil {
457	472		return errors.New("vev: nil owned daemon startup")

@@ -462,10 +477,98 @@ func runWithLifecycleOwnerDeps(ctx context.Context, runtimeDir, stateDir string,
462	477		// runDaemon runs the daemon in the foreground (the hidden --daemon path,
463	478		// entered by an auto-spawned child) while holding exclusive lifecycle
464	479		// ownership through complete teardown.
465		-	func runDaemon() error {
	480	+	func runDaemon() (retErr error) {
466	481		ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
467	482		defer stop()
468		-	return runWithLifecycleOwner(ctx, ipc.SocketDir(), platform.StateDir(), runDaemonOwned)
	483	+	log, logCloser, err := configureLogging(logging.Daemon, true)
	484	+	if err != nil {
	485	+	return err
	486	+	}
	487	+	defer func() { retErr = errors.Join(retErr, logCloser.Close()) }()
	488	+	return runWithLifecycleOwnerDeps(ctx, ipc.SocketDir(), platform.StateDir(), func(ctx context.Context) error {
	489	+	return runDaemonOwnedWithLogger(ctx, log)
	490	+	}, lifecycleStartupDeps{
	491	+	ensurePrivate: safedir.EnsurePrivate,
	492	+	acquire: func(ctx context.Context, runtimeDir string, retry time.Duration) (lifecycleOwnership, error) {
	493	+	return lifecycle.Acquire(ctx, runtimeDir, retry)
	494	+	},
	495	+	log: log,
	496	+	})
	497	+	}
	498	+
	499	+	func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, recoveryMode string) {
	500	+	log.Info("catalogue_validated", "records", len(records), "recovery", recoveryMode)
	501	+	log.Info("catalogue_compaction_recovery_complete", "recovery", recoveryMode)
	502	+	}
	503	+
	504	+	func logTransactionRecovery(log *slog.Logger, examined int) {
	505	+	log.Info("interrupted_transaction_recovery_complete", "sessions_examined", examined)
	506	+	}
	507	+
	508	+	type interruptedRecoveryIdentity struct {
	509	+	name string
	510	+	id domain.IncarnationID
	511	+	}
	512	+
	513	+	func pendingInterruptedRecoveries(ctx context.Context, records []domain.CatalogueRecord, journal ports.RecoveryJournal, repository ports.SnapshotRepository) ([]interruptedRecoveryIdentity, error) {
	514	+	pending := make([]interruptedRecoveryIdentity, 0)
	515	+	for _, record := range records {
	516	+	if record.RecoveryState == domain.RecoveryDeleting {
	517	+	pending = append(pending, interruptedRecoveryIdentity{name: record.Name, id: record.IncarnationID})
	518	+	}
	519	+	}
	520	+	intents, err := journal.ListDiscards(ctx)
	521	+	if err != nil {
	522	+	return nil, err
	523	+	}
	524	+	for _, intent := range intents {
	525	+	pending = append(pending, interruptedRecoveryIdentity{name: intent.SessionName, id: intent.OldIncarnation})
	526	+	}
	527	+	cursor := ports.DeletionTombstoneCursor{}
	528	+	for {
	529	+	page, err := repository.ListDeletionTombstones(ctx, cursor, ports.MaintenanceBudget{Entries: 64, Bytes: 64 << 10})
	530	+	if err != nil {
	531	+	return nil, err
	532	+	}
	533	+	for _, tombstone := range page.Tombstones {
	534	+	pending = append(pending, interruptedRecoveryIdentity{name: tombstone.Name, id: tombstone.IncarnationID})
	535	+	}
	536	+	if page.Done {
	537	+	return pending, nil
	538	+	}
	539	+	if page.Next.After == cursor.After {
	540	+	return nil, errors.New("vev: recovery tombstone listing did not advance")
	541	+	}
	542	+	cursor = page.Next
	543	+	}
	544	+	}
	545	+
	546	+	func persistTransactionRecoveryNotices(store ports.NoticeStore, recovered []interruptedRecoveryIdentity, now time.Time) {
	547	+	for _, item := range recovered {
	548	+	_ = store.Append(domain.Notification{
	549	+	Code: domain.NoticeSnapshotRestore,
	550	+	Severity: domain.NoticeWarn,
	551	+	Message: "recovered an interrupted durable session transaction",
	552	+	Details: "session=" + item.name + " incarnation=" + item.id.String(),
	553	+	Time: now,
	554	+	SessionID: domain.SessionID(item.id.String()),
	555	+	})
	556	+	}
	557	+	}
	558	+
	559	+	func logStartupRecoveryCounts(log *slog.Logger, records []domain.CatalogueRecord, restoring int) {
	560	+	healthy, fresh, degraded := 0, 0, 0
	561	+	for _, record := range records {
	562	+	switch record.RecoveryState {
	563	+	case domain.RecoveryHealthy:
	564	+	healthy++
	565	+	case domain.RecoveryFresh:
	566	+	fresh++
	567	+	case domain.RecoveryDegraded:
	568	+	degraded++
	569	+	}
	570	+	}
	571	+	log.Info("daemon_startup_complete", "healthy", healthy, "fresh", fresh, "restoring", restoring, "degraded", degraded)
469	572		}
470	573
471	574		func constructDaemonBeforeSocketPublication(

@@ -482,8 +585,11 @@ func runDaemonOwned(ctx context.Context) (retErr error) {
482	585		if err != nil {
483	586		return err
484	587		}
485		-	defer func() { _ = logCloser.Close() }()
	588	+	defer func() { retErr = errors.Join(retErr, logCloser.Close()) }()
	589	+	return runDaemonOwnedWithLogger(ctx, log)
	590	+	}
486	591
	592	+	func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr error) {
487	593		clk := clock.New()
488	594		observer, observerCloser, err := newPerformanceTrace(clk)
489	595		if err != nil {

@@ -518,9 +624,10 @@ func runDaemonOwned(ctx context.Context) (retErr error) {
518	624		daemonOpts = append(daemonOpts, daemon.WithConfig(cfg))
519	625		daemonOpts = append(daemonOpts, daemon.WithBarScriptCommandRunner(shellcmd.New()))
520	626		daemonOpts = append(daemonOpts, daemon.WithProcessInspector(platform.NewProcessInspector()), daemon.WithDirOrHome(platform.DirOrHome))
521		-	snapshotRepository := snapshotadapter.NewRepository(snapshotDir())
	627	+	snapshotRepository := snapshotadapter.NewRepository(snapshotDir(), log)
522	628		daemonOpts = append(daemonOpts, daemon.WithSnapshotRepository(snapshotRepository, snapshotRepository))
523		-	daemonOpts = append(daemonOpts, daemon.WithNoticeStore(noticefile.New(platform.StateDir())))
	629	+	noticeStore := noticefile.New(platform.StateDir())
	630	+	daemonOpts = append(daemonOpts, daemon.WithNoticeStore(noticeStore))
524	631		storePath := persist.StorePath(platform.StateDir())
525	632		opened, err := persist.OpenOrMigrate(ctx, persist.OpenDeps{
526	633		StateDir: platform.StateDir(),

@@ -528,12 +635,27 @@ func runDaemonOwned(ctx context.Context) (retErr error) {
528	635		SnapshotMigration: snapshotRepository,
529	636		})
530	637		if err != nil {
	638	+	log.Error("catalogue_validation_failed", "path", storePath, "reason_code", "open-failed")
531	639		return fmt.Errorf("vev: open or migrate durable session state %s: %w", storePath, err)
532	640		}
533		-	coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, recoveryfs.New(platform.StateDir()), rand.Reader)
	641	+	recoveryMode := "current"
	642	+	if opened.NewInstall {
	643	+	recoveryMode = "new-install"
	644	+	} else if opened.Migrated {
	645	+	recoveryMode = "migrated"
	646	+	}
	647	+	logCatalogueRecovery(log, opened.Records, recoveryMode)
	648	+	journal := recoveryfs.New(platform.StateDir())
	649	+	pendingRecoveries, err := pendingInterruptedRecoveries(ctx, opened.Records, journal, snapshotRepository)
	650	+	if err != nil {
	651	+	return errors.Join(fmt.Errorf("vev: inspect durable session transactions: %w", err), opened.Catalogue.Close())
	652	+	}
	653	+	coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, journal, rand.Reader)
534	654		if err := coordinator.Recover(ctx); err != nil {
535	655		return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())
536	656		}
	657	+	persistTransactionRecoveryNotices(noticeStore, pendingRecoveries, clk.Now())
	658	+	logTransactionRecovery(log, len(pendingRecoveries))
537	659		daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))
538	660		log.Info("session persistence enabled", "path", storePath)
539	661		daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))


@@ -426,6 +426,7 @@ func joinLifecycleReleaseError(retErr *error, owner lifecycleOwnership) {
426		type lifecycleStartupDeps struct {	426		type lifecycleStartupDeps struct {
427		ensurePrivate func(string) error	427		ensurePrivate func(string) error
428		acquire func(context.Context, string, time.Duration) (lifecycleOwnership, error)	428		acquire func(context.Context, string, time.Duration) (lifecycleOwnership, error)
			429	+	log *slog.Logger
429		}	430		}
430			431
431		const lifecycleAcquireRetry = 25 * time.Millisecond	432		const lifecycleAcquireRetry = 25 * time.Millisecond

@@ -440,18 +441,32 @@ func runWithLifecycleOwner(ctx context.Context, runtimeDir, stateDir string, sta
440		}	441		}
441			442
442		func runWithLifecycleOwnerDeps(ctx context.Context, runtimeDir, stateDir string, start ownedDaemonStart, deps lifecycleStartupDeps) (retErr error) {	443		func runWithLifecycleOwnerDeps(ctx context.Context, runtimeDir, stateDir string, start ownedDaemonStart, deps lifecycleStartupDeps) (retErr error) {
			444	+	log := deps.log
			445	+	if log == nil {
			446	+	log = slog.Default()
			447	+	}
443		if err := deps.ensurePrivate(runtimeDir); err != nil {	448		if err := deps.ensurePrivate(runtimeDir); err != nil {
444		return fmt.Errorf("vev: secure runtime directory: %w", err)	449		return fmt.Errorf("vev: secure runtime directory: %w", err)
445		}	450		}
446		if err := deps.ensurePrivate(stateDir); err != nil {	451		if err := deps.ensurePrivate(stateDir); err != nil {
447		return fmt.Errorf("vev: secure state directory: %w", err)	452		return fmt.Errorf("vev: secure state directory: %w", err)
448		}	453		}
			454	+	lockPath := lifecycle.Path(runtimeDir)
			455	+	log.Info("lifecycle_owner_wait", "path", lockPath)
449		owner, err := deps.acquire(ctx, runtimeDir, lifecycleAcquireRetry)	456		owner, err := deps.acquire(ctx, runtimeDir, lifecycleAcquireRetry)
450		if err != nil {	457		if err != nil {
			458	+	log.Warn("lifecycle_owner_wait_failed", "path", lockPath, "reason_code", "acquire-failed")
451		return fmt.Errorf("vev: acquire lifecycle ownership: %w", err)	459		return fmt.Errorf("vev: acquire lifecycle ownership: %w", err)
452		}	460		}
			461	+	log.Info("lifecycle_owner_acquired", "path", lockPath)
453		defer func() {	462		defer func() {
454	-	retErr = errors.Join(retErr, owner.Release())	463	+	releaseErr := owner.Release()
			464	+	if releaseErr != nil {
			465	+	log.Error("lifecycle_owner_release_failed", "path", lockPath, "reason_code", "release-failed")
			466	+	} else {
			467	+	log.Info("lifecycle_owner_released", "path", lockPath)
			468	+	}
			469	+	retErr = errors.Join(retErr, releaseErr)
455		}()	470		}()
456		if start == nil {	471		if start == nil {
457		return errors.New("vev: nil owned daemon startup")	472		return errors.New("vev: nil owned daemon startup")

@@ -462,10 +477,98 @@ func runWithLifecycleOwnerDeps(ctx context.Context, runtimeDir, stateDir string,
462		// runDaemon runs the daemon in the foreground (the hidden --daemon path,	477		// runDaemon runs the daemon in the foreground (the hidden --daemon path,
463		// entered by an auto-spawned child) while holding exclusive lifecycle	478		// entered by an auto-spawned child) while holding exclusive lifecycle
464		// ownership through complete teardown.	479		// ownership through complete teardown.
465	-	func runDaemon() error {	480	+	func runDaemon() (retErr error) {
466		ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)	481		ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
467		defer stop()	482		defer stop()
468	-	return runWithLifecycleOwner(ctx, ipc.SocketDir(), platform.StateDir(), runDaemonOwned)	483	+	log, logCloser, err := configureLogging(logging.Daemon, true)
			484	+	if err != nil {
			485	+	return err
			486	+	}
			487	+	defer func() { retErr = errors.Join(retErr, logCloser.Close()) }()
			488	+	return runWithLifecycleOwnerDeps(ctx, ipc.SocketDir(), platform.StateDir(), func(ctx context.Context) error {
			489	+	return runDaemonOwnedWithLogger(ctx, log)
			490	+	}, lifecycleStartupDeps{
			491	+	ensurePrivate: safedir.EnsurePrivate,
			492	+	acquire: func(ctx context.Context, runtimeDir string, retry time.Duration) (lifecycleOwnership, error) {
			493	+	return lifecycle.Acquire(ctx, runtimeDir, retry)
			494	+	},
			495	+	log: log,
			496	+	})
			497	+	}
			498	+
			499	+	func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, recoveryMode string) {
			500	+	log.Info("catalogue_validated", "records", len(records), "recovery", recoveryMode)
			501	+	log.Info("catalogue_compaction_recovery_complete", "recovery", recoveryMode)
			502	+	}
			503	+
			504	+	func logTransactionRecovery(log *slog.Logger, examined int) {
			505	+	log.Info("interrupted_transaction_recovery_complete", "sessions_examined", examined)
			506	+	}
			507	+
			508	+	type interruptedRecoveryIdentity struct {
			509	+	name string
			510	+	id domain.IncarnationID
			511	+	}
			512	+
			513	+	func pendingInterruptedRecoveries(ctx context.Context, records []domain.CatalogueRecord, journal ports.RecoveryJournal, repository ports.SnapshotRepository) ([]interruptedRecoveryIdentity, error) {
			514	+	pending := make([]interruptedRecoveryIdentity, 0)
			515	+	for _, record := range records {
			516	+	if record.RecoveryState == domain.RecoveryDeleting {
			517	+	pending = append(pending, interruptedRecoveryIdentity{name: record.Name, id: record.IncarnationID})
			518	+	}
			519	+	}
			520	+	intents, err := journal.ListDiscards(ctx)
			521	+	if err != nil {
			522	+	return nil, err
			523	+	}
			524	+	for _, intent := range intents {
			525	+	pending = append(pending, interruptedRecoveryIdentity{name: intent.SessionName, id: intent.OldIncarnation})
			526	+	}
			527	+	cursor := ports.DeletionTombstoneCursor{}
			528	+	for {
			529	+	page, err := repository.ListDeletionTombstones(ctx, cursor, ports.MaintenanceBudget{Entries: 64, Bytes: 64 << 10})
			530	+	if err != nil {
			531	+	return nil, err
			532	+	}
			533	+	for _, tombstone := range page.Tombstones {
			534	+	pending = append(pending, interruptedRecoveryIdentity{name: tombstone.Name, id: tombstone.IncarnationID})
			535	+	}
			536	+	if page.Done {
			537	+	return pending, nil
			538	+	}
			539	+	if page.Next.After == cursor.After {
			540	+	return nil, errors.New("vev: recovery tombstone listing did not advance")
			541	+	}
			542	+	cursor = page.Next
			543	+	}
			544	+	}
			545	+
			546	+	func persistTransactionRecoveryNotices(store ports.NoticeStore, recovered []interruptedRecoveryIdentity, now time.Time) {
			547	+	for _, item := range recovered {
			548	+	_ = store.Append(domain.Notification{
			549	+	Code: domain.NoticeSnapshotRestore,
			550	+	Severity: domain.NoticeWarn,
			551	+	Message: "recovered an interrupted durable session transaction",
			552	+	Details: "session=" + item.name + " incarnation=" + item.id.String(),
			553	+	Time: now,
			554	+	SessionID: domain.SessionID(item.id.String()),
			555	+	})
			556	+	}
			557	+	}
			558	+
			559	+	func logStartupRecoveryCounts(log *slog.Logger, records []domain.CatalogueRecord, restoring int) {
			560	+	healthy, fresh, degraded := 0, 0, 0
			561	+	for _, record := range records {
			562	+	switch record.RecoveryState {
			563	+	case domain.RecoveryHealthy:
			564	+	healthy++
			565	+	case domain.RecoveryFresh:
			566	+	fresh++
			567	+	case domain.RecoveryDegraded:
			568	+	degraded++
			569	+	}
			570	+	}
			571	+	log.Info("daemon_startup_complete", "healthy", healthy, "fresh", fresh, "restoring", restoring, "degraded", degraded)
469		}	572		}
470			573
471		func constructDaemonBeforeSocketPublication(	574		func constructDaemonBeforeSocketPublication(

@@ -482,8 +585,11 @@ func runDaemonOwned(ctx context.Context) (retErr error) {
482		if err != nil {	585		if err != nil {
483		return err	586		return err
484		}	587		}
485	-	defer func() { _ = logCloser.Close() }()	588	+	defer func() { retErr = errors.Join(retErr, logCloser.Close()) }()
			589	+	return runDaemonOwnedWithLogger(ctx, log)
			590	+	}
486			591
			592	+	func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr error) {
487		clk := clock.New()	593		clk := clock.New()
488		observer, observerCloser, err := newPerformanceTrace(clk)	594		observer, observerCloser, err := newPerformanceTrace(clk)
489		if err != nil {	595		if err != nil {

@@ -518,9 +624,10 @@ func runDaemonOwned(ctx context.Context) (retErr error) {
518		daemonOpts = append(daemonOpts, daemon.WithConfig(cfg))	624		daemonOpts = append(daemonOpts, daemon.WithConfig(cfg))
519		daemonOpts = append(daemonOpts, daemon.WithBarScriptCommandRunner(shellcmd.New()))	625		daemonOpts = append(daemonOpts, daemon.WithBarScriptCommandRunner(shellcmd.New()))
520		daemonOpts = append(daemonOpts, daemon.WithProcessInspector(platform.NewProcessInspector()), daemon.WithDirOrHome(platform.DirOrHome))	626		daemonOpts = append(daemonOpts, daemon.WithProcessInspector(platform.NewProcessInspector()), daemon.WithDirOrHome(platform.DirOrHome))
521	-	snapshotRepository := snapshotadapter.NewRepository(snapshotDir())	627	+	snapshotRepository := snapshotadapter.NewRepository(snapshotDir(), log)
522		daemonOpts = append(daemonOpts, daemon.WithSnapshotRepository(snapshotRepository, snapshotRepository))	628		daemonOpts = append(daemonOpts, daemon.WithSnapshotRepository(snapshotRepository, snapshotRepository))
523	-	daemonOpts = append(daemonOpts, daemon.WithNoticeStore(noticefile.New(platform.StateDir())))	629	+	noticeStore := noticefile.New(platform.StateDir())
			630	+	daemonOpts = append(daemonOpts, daemon.WithNoticeStore(noticeStore))
524		storePath := persist.StorePath(platform.StateDir())	631		storePath := persist.StorePath(platform.StateDir())
525		opened, err := persist.OpenOrMigrate(ctx, persist.OpenDeps{	632		opened, err := persist.OpenOrMigrate(ctx, persist.OpenDeps{
526		StateDir: platform.StateDir(),	633		StateDir: platform.StateDir(),

@@ -528,12 +635,27 @@ func runDaemonOwned(ctx context.Context) (retErr error) {
528		SnapshotMigration: snapshotRepository,	635		SnapshotMigration: snapshotRepository,
529		})	636		})
530		if err != nil {	637		if err != nil {
			638	+	log.Error("catalogue_validation_failed", "path", storePath, "reason_code", "open-failed")
531		return fmt.Errorf("vev: open or migrate durable session state %s: %w", storePath, err)	639		return fmt.Errorf("vev: open or migrate durable session state %s: %w", storePath, err)
532		}	640		}
533	-	coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, recoveryfs.New(platform.StateDir()), rand.Reader)	641	+	recoveryMode := "current"
			642	+	if opened.NewInstall {
			643	+	recoveryMode = "new-install"
			644	+	} else if opened.Migrated {
			645	+	recoveryMode = "migrated"
			646	+	}
			647	+	logCatalogueRecovery(log, opened.Records, recoveryMode)
			648	+	journal := recoveryfs.New(platform.StateDir())
			649	+	pendingRecoveries, err := pendingInterruptedRecoveries(ctx, opened.Records, journal, snapshotRepository)
			650	+	if err != nil {
			651	+	return errors.Join(fmt.Errorf("vev: inspect durable session transactions: %w", err), opened.Catalogue.Close())
			652	+	}
			653	+	coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, journal, rand.Reader)
534		if err := coordinator.Recover(ctx); err != nil {	654		if err := coordinator.Recover(ctx); err != nil {
535		return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())	655		return errors.Join(fmt.Errorf("vev: recover durable session transactions: %w", err), opened.Catalogue.Close())
536		}	656		}
			657	+	persistTransactionRecoveryNotices(noticeStore, pendingRecoveries, clk.Now())
			658	+	logTransactionRecovery(log, len(pendingRecoveries))
537		daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))	659		daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))
538		log.Info("session persistence enabled", "path", storePath)	660		log.Info("session persistence enabled", "path", storePath)
539		daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))	661		daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))

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

@@ -1,10 +1,14 @@
1	1		package app
2	2
3	3		import (
	4	+	"bufio"
4	5		"bytes"
5	6		"context"
	7	+	"encoding/json"
6	8		"errors"
	9	+	"fmt"
7	10		"io"
	11	+	"log/slog"
8	12		"net"
9	13		"os"
10	14		"os/exec"

@@ -27,6 +31,87 @@ import (
27	31		"github.com/stretchr/testify/require"
28	32		)
29	33
	34	+	func decodeJSONLogs(t *testing.T, data []byte) []map[string]any {
	35	+	t.Helper()
	36	+	var entries []map[string]any
	37	+	scanner := bufio.NewScanner(bytes.NewReader(data))
	38	+	for scanner.Scan() {
	39	+	var entry map[string]any
	40	+	require.NoError(t, json.Unmarshal(scanner.Bytes(), &entry))
	41	+	entries = append(entries, entry)
	42	+	}
	43	+	require.NoError(t, scanner.Err())
	44	+	return entries
	45	+	}
	46	+
	47	+	func eventNames(entries []map[string]any) []string {
	48	+	names := make([]string, 0, len(entries))
	49	+	for _, entry := range entries {
	50	+	if name, ok := entry["msg"].(string); ok {
	51	+	names = append(names, name)
	52	+	}
	53	+	}
	54	+	return names
	55	+	}
	56	+
	57	+	func requireEvent(t *testing.T, entries []map[string]any, name string) map[string]any {
	58	+	t.Helper()
	59	+	for _, entry := range entries {
	60	+	if entry["msg"] == name {
	61	+	return entry
	62	+	}
	63	+	}
	64	+	t.Fatalf("event %q not found in %v", name, eventNames(entries))
	65	+	return nil
	66	+	}
	67	+
	68	+	func TestRecoveryObservability(t *testing.T) {
	69	+	var logBuffer bytes.Buffer
	70	+	log := slog.New(slog.NewJSONHandler(&logBuffer, nil))
	71	+	owner := fakeLifecycleOwnership{release: func() error { return nil }}
	72	+	deps := lifecycleStartupDeps{
	73	+	ensurePrivate: func(string) error { return nil },
	74	+	acquire: func(context.Context, string, time.Duration) (lifecycleOwnership, error) {
	75	+	return owner, nil
	76	+	},
	77	+	log: log,
	78	+	}
	79	+	records := []domain.CatalogueRecord{
	80	+	{RecoveryState: domain.RecoveryHealthy},
	81	+	{RecoveryState: domain.RecoveryHealthy},
	82	+	{RecoveryState: domain.RecoveryFresh},
	83	+	{RecoveryState: domain.RecoveryDegraded},
	84	+	}
	85	+	noticeStore := portsmocks.NewMockNoticeStore(t)
	86	+	var recoveryNotice domain.Notification
	87	+	noticeStore.EXPECT().Append(mock.Anything).Run(func(n domain.Notification) { recoveryNotice = n }).Return(nil).Once()
	88	+	require.NoError(t, runWithLifecycleOwnerDeps(context.Background(), "/runtime/vev", "/state/vev", func(context.Context) error {
	89	+	logCatalogueRecovery(log, records, "current")
	90	+	logTransactionRecovery(log, 2)
	91	+	persistTransactionRecoveryNotices(noticeStore, []interruptedRecoveryIdentity{{name: "work", id: domain.IncarnationID{9}}}, time.Unix(1, 0))
	92	+	logStartupRecoveryCounts(log, records, 0)
	93	+	return nil
	94	+	}, deps))
	95	+
	96	+	entries := decodeJSONLogs(t, logBuffer.Bytes())
	97	+	for _, name := range []string{"lifecycle_owner_wait", "lifecycle_owner_acquired", "catalogue_validated", "catalogue_compaction_recovery_complete", "interrupted_transaction_recovery_complete", "daemon_startup_complete", "lifecycle_owner_released"} {
	98	+	require.Contains(t, eventNames(entries), name)
	99	+	}
	100	+	startup := requireEvent(t, entries, "daemon_startup_complete")
	101	+	require.EqualValues(t, 2, startup["healthy"])
	102	+	require.EqualValues(t, 1, startup["fresh"])
	103	+	require.EqualValues(t, 0, startup["restoring"])
	104	+	require.EqualValues(t, 1, startup["degraded"])
	105	+
	106	+	logBuffer.Reset()
	107	+	log.Error("catalogue_validation_failed", "path", "/state/vev/sessions.kv", "reason_code", "corrupt")
	108	+	require.NotContains(t, eventNames(decodeJSONLogs(t, logBuffer.Bytes())), "daemon_startup_complete")
	109	+	require.NotContains(t, fmt.Sprint(requireEvent(t, entries, "interrupted_transaction_recovery_complete")), "terminal contents")
	110	+	require.Equal(t, domain.NoticeSnapshotRestore, recoveryNotice.Code)
	111	+	require.Equal(t, domain.SessionID(domain.IncarnationID{9}.String()), recoveryNotice.SessionID)
	112	+	require.NotContains(t, fmt.Sprint(recoveryNotice), "terminal contents")
	113	+	}
	114	+
30	115		type fakeLifecycleOwnership struct {
31	116		release func() error
32	117		}

@@ -1,10 +1,14 @@
1	package app	1		package app
2		2
3	import (	3		import (
		4	+	"bufio"
4	"bytes"	5		"bytes"
5	"context"	6		"context"
		7	+	"encoding/json"
6	"errors"	8		"errors"
		9	+	"fmt"
7	"io"	10		"io"
		11	+	"log/slog"
8	"net"	12		"net"
9	"os"	13		"os"
10	"os/exec"	14		"os/exec"

@@ -27,6 +31,87 @@ import (
27	"github.com/stretchr/testify/require"	31		"github.com/stretchr/testify/require"
28	)	32		)
29		33
		34	+	func decodeJSONLogs(t *testing.T, data []byte) []map[string]any {
		35	+	t.Helper()
		36	+	var entries []map[string]any
		37	+	scanner := bufio.NewScanner(bytes.NewReader(data))
		38	+	for scanner.Scan() {
		39	+	var entry map[string]any
		40	+	require.NoError(t, json.Unmarshal(scanner.Bytes(), &entry))
		41	+	entries = append(entries, entry)
		42	+	}
		43	+	require.NoError(t, scanner.Err())
		44	+	return entries
		45	+	}
		46	+
		47	+	func eventNames(entries []map[string]any) []string {
		48	+	names := make([]string, 0, len(entries))
		49	+	for _, entry := range entries {
		50	+	if name, ok := entry["msg"].(string); ok {
		51	+	names = append(names, name)
		52	+	}
		53	+	}
		54	+	return names
		55	+	}
		56	+
		57	+	func requireEvent(t *testing.T, entries []map[string]any, name string) map[string]any {
		58	+	t.Helper()
		59	+	for _, entry := range entries {
		60	+	if entry["msg"] == name {
		61	+	return entry
		62	+	}
		63	+	}
		64	+	t.Fatalf("event %q not found in %v", name, eventNames(entries))
		65	+	return nil
		66	+	}
		67	+
		68	+	func TestRecoveryObservability(t *testing.T) {
		69	+	var logBuffer bytes.Buffer
		70	+	log := slog.New(slog.NewJSONHandler(&logBuffer, nil))
		71	+	owner := fakeLifecycleOwnership{release: func() error { return nil }}
		72	+	deps := lifecycleStartupDeps{
		73	+	ensurePrivate: func(string) error { return nil },
		74	+	acquire: func(context.Context, string, time.Duration) (lifecycleOwnership, error) {
		75	+	return owner, nil
		76	+	},
		77	+	log: log,
		78	+	}
		79	+	records := []domain.CatalogueRecord{
		80	+	{RecoveryState: domain.RecoveryHealthy},
		81	+	{RecoveryState: domain.RecoveryHealthy},
		82	+	{RecoveryState: domain.RecoveryFresh},
		83	+	{RecoveryState: domain.RecoveryDegraded},
		84	+	}
		85	+	noticeStore := portsmocks.NewMockNoticeStore(t)
		86	+	var recoveryNotice domain.Notification
		87	+	noticeStore.EXPECT().Append(mock.Anything).Run(func(n domain.Notification) { recoveryNotice = n }).Return(nil).Once()
		88	+	require.NoError(t, runWithLifecycleOwnerDeps(context.Background(), "/runtime/vev", "/state/vev", func(context.Context) error {
		89	+	logCatalogueRecovery(log, records, "current")
		90	+	logTransactionRecovery(log, 2)
		91	+	persistTransactionRecoveryNotices(noticeStore, []interruptedRecoveryIdentity{{name: "work", id: domain.IncarnationID{9}}}, time.Unix(1, 0))
		92	+	logStartupRecoveryCounts(log, records, 0)
		93	+	return nil
		94	+	}, deps))
		95	+
		96	+	entries := decodeJSONLogs(t, logBuffer.Bytes())
		97	+	for _, name := range []string{"lifecycle_owner_wait", "lifecycle_owner_acquired", "catalogue_validated", "catalogue_compaction_recovery_complete", "interrupted_transaction_recovery_complete", "daemon_startup_complete", "lifecycle_owner_released"} {
		98	+	require.Contains(t, eventNames(entries), name)
		99	+	}
		100	+	startup := requireEvent(t, entries, "daemon_startup_complete")
		101	+	require.EqualValues(t, 2, startup["healthy"])
		102	+	require.EqualValues(t, 1, startup["fresh"])
		103	+	require.EqualValues(t, 0, startup["restoring"])
		104	+	require.EqualValues(t, 1, startup["degraded"])
		105	+
		106	+	logBuffer.Reset()
		107	+	log.Error("catalogue_validation_failed", "path", "/state/vev/sessions.kv", "reason_code", "corrupt")
		108	+	require.NotContains(t, eventNames(decodeJSONLogs(t, logBuffer.Bytes())), "daemon_startup_complete")
		109	+	require.NotContains(t, fmt.Sprint(requireEvent(t, entries, "interrupted_transaction_recovery_complete")), "terminal contents")
		110	+	require.Equal(t, domain.NoticeSnapshotRestore, recoveryNotice.Code)
		111	+	require.Equal(t, domain.SessionID(domain.IncarnationID{9}.String()), recoveryNotice.SessionID)
		112	+	require.NotContains(t, fmt.Sprint(recoveryNotice), "terminal contents")
		113	+	}
		114	+
30	type fakeLifecycleOwnership struct {	115		type fakeLifecycleOwnership struct {
31	release func() error	116		release func() error
32	}	117		}

A internal/usecase/daemon/recovery_observability.go +51 -0

@@ -0,0 +1,51 @@
	1	+	package daemon
	2	+
	3	+	import "github.com/bnema/vev/internal/domain"
	4	+
	5	+	func (d *Daemon) logSessionRestoreComplete(record domain.CatalogueRecord, generation uint64, fallback bool) {
	6	+	d.log.Info("session_restore_complete",
	7	+	"session", record.Name,
	8	+	"incarnation", record.IncarnationID.String(),
	9	+	"generation", generation,
	10	+	"fallback", fallback,
	11	+	)
	12	+	}
	13	+
	14	+	func (d *Daemon) logSessionDegraded(record domain.CatalogueRecord, reasonCode string) {
	15	+	d.log.Warn("session_degraded",
	16	+	"session", record.Name,
	17	+	"incarnation", record.IncarnationID.String(),
	18	+	"reason_code", reasonCode,
	19	+	)
	20	+	}
	21	+
	22	+	func (d *Daemon) logStartupRecoveryCounts(restoring int) {
	23	+	var records []domain.CatalogueRecord
	24	+	if d.catalogue != nil {
	25	+	records = d.catalogue.Records()
	26	+	} else {
	27	+	d.mu.Lock()
	28	+	records = make([]domain.CatalogueRecord, 0, len(d.stopped))
	29	+	for _, entry := range d.stopped {
	30	+	records = append(records, entry.record)
	31	+	}
	32	+	d.mu.Unlock()
	33	+	}
	34	+	healthy, fresh, degraded := 0, 0, 0
	35	+	for _, record := range records {
	36	+	switch record.RecoveryState {
	37	+	case domain.RecoveryHealthy:
	38	+	healthy++
	39	+	case domain.RecoveryFresh:
	40	+	fresh++
	41	+	case domain.RecoveryDegraded:
	42	+	degraded++
	43	+	}
	44	+	}
	45	+	d.log.Info("daemon_startup_complete",
	46	+	"healthy", healthy,
	47	+	"fresh", fresh,
	48	+	"restoring", restoring,
	49	+	"degraded", degraded,
	50	+	)
	51	+	}

@@ -0,0 +1,51 @@
	1	+	package daemon
	2	+
	3	+	import "github.com/bnema/vev/internal/domain"
	4	+
	5	+	func (d *Daemon) logSessionRestoreComplete(record domain.CatalogueRecord, generation uint64, fallback bool) {
	6	+	d.log.Info("session_restore_complete",
	7	+	"session", record.Name,
	8	+	"incarnation", record.IncarnationID.String(),
	9	+	"generation", generation,
	10	+	"fallback", fallback,
	11	+	)
	12	+	}
	13	+
	14	+	func (d *Daemon) logSessionDegraded(record domain.CatalogueRecord, reasonCode string) {
	15	+	d.log.Warn("session_degraded",
	16	+	"session", record.Name,
	17	+	"incarnation", record.IncarnationID.String(),
	18	+	"reason_code", reasonCode,
	19	+	)
	20	+	}
	21	+
	22	+	func (d *Daemon) logStartupRecoveryCounts(restoring int) {
	23	+	var records []domain.CatalogueRecord
	24	+	if d.catalogue != nil {
	25	+	records = d.catalogue.Records()
	26	+	} else {
	27	+	d.mu.Lock()
	28	+	records = make([]domain.CatalogueRecord, 0, len(d.stopped))
	29	+	for _, entry := range d.stopped {
	30	+	records = append(records, entry.record)
	31	+	}
	32	+	d.mu.Unlock()
	33	+	}
	34	+	healthy, fresh, degraded := 0, 0, 0
	35	+	for _, record := range records {
	36	+	switch record.RecoveryState {
	37	+	case domain.RecoveryHealthy:
	38	+	healthy++
	39	+	case domain.RecoveryFresh:
	40	+	fresh++
	41	+	case domain.RecoveryDegraded:
	42	+	degraded++
	43	+	}
	44	+	}
	45	+	d.log.Info("daemon_startup_complete",
	46	+	"healthy", healthy,
	47	+	"fresh", fresh,
	48	+	"restoring", restoring,
	49	+	"degraded", degraded,
	50	+	)
	51	+	}

A internal/usecase/daemon/recovery_observability_test.go +65 -0

@@ -0,0 +1,65 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"bufio"
	5	+	"bytes"
	6	+	"encoding/json"
	7	+	"log/slog"
	8	+	"testing"
	9	+
	10	+	"github.com/bnema/vev/internal/domain"
	11	+	"github.com/stretchr/testify/require"
	12	+	)
	13	+
	14	+	func TestStartupRecoveryCounts(t *testing.T) {
	15	+	var buffer bytes.Buffer
	16	+	log := slog.New(slog.NewJSONHandler(&buffer, nil))
	17	+	records := []domain.CatalogueRecord{
	18	+	{Name: "one", IncarnationID: domain.IncarnationID{1}, RecoveryState: domain.RecoveryHealthy},
	19	+	{Name: "two", IncarnationID: domain.IncarnationID{2}, RecoveryState: domain.RecoveryHealthy},
	20	+	{Name: "new", IncarnationID: domain.IncarnationID{3}, RecoveryState: domain.RecoveryFresh},
	21	+	{Name: "broken", IncarnationID: domain.IncarnationID{4}, RecoveryState: domain.RecoveryDegraded, DegradedReason: "terminal contents must not appear"},
	22	+	}
	23	+	d := New(nil, stubClock{}, log, WithCatalogue(newDurableRecoveryCatalogue(records), records))
	24	+	d.logSessionRestoreComplete(records[0], 7, false)
	25	+	d.logSessionDegraded(records[3], "checkpoint-invalid")
	26	+	d.logStartupRecoveryCounts(0)
	27	+
	28	+	entries := daemonJSONLogs(t, buffer.Bytes())
	29	+	restore := daemonRequireEvent(t, entries, "session_restore_complete")
	30	+	require.Equal(t, "one", restore["session"])
	31	+	require.Equal(t, records[0].IncarnationID.String(), restore["incarnation"])
	32	+	require.EqualValues(t, 7, restore["generation"])
	33	+	degraded := daemonRequireEvent(t, entries, "session_degraded")
	34	+	require.Equal(t, "checkpoint-invalid", degraded["reason_code"])
	35	+	require.NotContains(t, degraded, "reason")
	36	+	startup := daemonRequireEvent(t, entries, "daemon_startup_complete")
	37	+	require.EqualValues(t, 2, startup["healthy"])
	38	+	require.EqualValues(t, 1, startup["fresh"])
	39	+	require.EqualValues(t, 0, startup["restoring"])
	40	+	require.EqualValues(t, 1, startup["degraded"])
	41	+	}
	42	+
	43	+	func daemonJSONLogs(t *testing.T, data []byte) []map[string]any {
	44	+	t.Helper()
	45	+	var entries []map[string]any
	46	+	scanner := bufio.NewScanner(bytes.NewReader(data))
	47	+	for scanner.Scan() {
	48	+	var entry map[string]any
	49	+	require.NoError(t, json.Unmarshal(scanner.Bytes(), &entry))
	50	+	entries = append(entries, entry)
	51	+	}
	52	+	require.NoError(t, scanner.Err())
	53	+	return entries
	54	+	}
	55	+
	56	+	func daemonRequireEvent(t *testing.T, entries []map[string]any, name string) map[string]any {
	57	+	t.Helper()
	58	+	for _, entry := range entries {
	59	+	if entry["msg"] == name {
	60	+	return entry
	61	+	}
	62	+	}
	63	+	t.Fatalf("event %q not found", name)
	64	+	return nil
	65	+	}

@@ -0,0 +1,65 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"bufio"
	5	+	"bytes"
	6	+	"encoding/json"
	7	+	"log/slog"
	8	+	"testing"
	9	+
	10	+	"github.com/bnema/vev/internal/domain"
	11	+	"github.com/stretchr/testify/require"
	12	+	)
	13	+
	14	+	func TestStartupRecoveryCounts(t *testing.T) {
	15	+	var buffer bytes.Buffer
	16	+	log := slog.New(slog.NewJSONHandler(&buffer, nil))
	17	+	records := []domain.CatalogueRecord{
	18	+	{Name: "one", IncarnationID: domain.IncarnationID{1}, RecoveryState: domain.RecoveryHealthy},
	19	+	{Name: "two", IncarnationID: domain.IncarnationID{2}, RecoveryState: domain.RecoveryHealthy},
	20	+	{Name: "new", IncarnationID: domain.IncarnationID{3}, RecoveryState: domain.RecoveryFresh},
	21	+	{Name: "broken", IncarnationID: domain.IncarnationID{4}, RecoveryState: domain.RecoveryDegraded, DegradedReason: "terminal contents must not appear"},
	22	+	}
	23	+	d := New(nil, stubClock{}, log, WithCatalogue(newDurableRecoveryCatalogue(records), records))
	24	+	d.logSessionRestoreComplete(records[0], 7, false)
	25	+	d.logSessionDegraded(records[3], "checkpoint-invalid")
	26	+	d.logStartupRecoveryCounts(0)
	27	+
	28	+	entries := daemonJSONLogs(t, buffer.Bytes())
	29	+	restore := daemonRequireEvent(t, entries, "session_restore_complete")
	30	+	require.Equal(t, "one", restore["session"])
	31	+	require.Equal(t, records[0].IncarnationID.String(), restore["incarnation"])
	32	+	require.EqualValues(t, 7, restore["generation"])
	33	+	degraded := daemonRequireEvent(t, entries, "session_degraded")
	34	+	require.Equal(t, "checkpoint-invalid", degraded["reason_code"])
	35	+	require.NotContains(t, degraded, "reason")
	36	+	startup := daemonRequireEvent(t, entries, "daemon_startup_complete")
	37	+	require.EqualValues(t, 2, startup["healthy"])
	38	+	require.EqualValues(t, 1, startup["fresh"])
	39	+	require.EqualValues(t, 0, startup["restoring"])
	40	+	require.EqualValues(t, 1, startup["degraded"])
	41	+	}
	42	+
	43	+	func daemonJSONLogs(t *testing.T, data []byte) []map[string]any {
	44	+	t.Helper()
	45	+	var entries []map[string]any
	46	+	scanner := bufio.NewScanner(bytes.NewReader(data))
	47	+	for scanner.Scan() {
	48	+	var entry map[string]any
	49	+	require.NoError(t, json.Unmarshal(scanner.Bytes(), &entry))
	50	+	entries = append(entries, entry)
	51	+	}
	52	+	require.NoError(t, scanner.Err())
	53	+	return entries
	54	+	}
	55	+
	56	+	func daemonRequireEvent(t *testing.T, entries []map[string]any, name string) map[string]any {
	57	+	t.Helper()
	58	+	for _, entry := range entries {
	59	+	if entry["msg"] == name {
	60	+	return entry
	61	+	}
	62	+	}
	63	+	t.Fatalf("event %q not found", name)
	64	+	return nil
	65	+	}

M internal/usecase/daemon/snapshot_restore.go +1 -0


@@ -21,6 +21,7 @@ func (d *Daemon) restoreSnapshots(ctx context.Context) {
21	21		return
22	22		}
23	23		defer d.closeRestoreDone()
	24	+	defer d.logStartupRecoveryCounts(0)
24	25		if ns := d.noticeStore; ns != nil {
25	26		claimed, err := ns.Claim()
26	27		if err != nil {


@@ -21,6 +21,7 @@ func (d *Daemon) restoreSnapshots(ctx context.Context) {
21	return	21		return
22	}	22		}
23	defer d.closeRestoreDone()	23		defer d.closeRestoreDone()
		24	+	defer d.logStartupRecoveryCounts(0)
24	if ns := d.noticeStore; ns != nil {	25		if ns := d.noticeStore; ns != nil {
25	claimed, err := ns.Claim()	26		claimed, err := ns.Claim()
26	if err != nil {	27		if err != nil {

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


@@ -159,7 +159,11 @@ func (d *Daemon) finishRecordRestore(record domain.CatalogueRecord, restoreErr e
159	159		}
160	160		d.mu.Unlock()
161	161		if restoreErr != nil {
162		-	d.log.Warn("session checkpoint restore failed", "session", record.Name, "err", restoreErr)
	162	+	reasonCode := "checkpoint-invalid"
	163	+	if errors.Is(restoreErr, context.Canceled) \|\| errors.Is(restoreErr, context.DeadlineExceeded) {
	164	+	reasonCode = "restore-interrupted"
	165	+	}
	166	+	d.logSessionDegraded(record, reasonCode)
163	167		}
164	168		}
165	169

@@ -167,9 +171,11 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
167	171		switch record.RecoveryState {
168	172		case domain.RecoveryFresh:
169	173		d.setStoppedRecovery(record, runtimeFresh)
	174	+	d.logSessionRestoreComplete(record, 0, false)
170	175		return nil
171	176		case domain.RecoveryDegraded:
172	177		d.setStoppedRecovery(record, runtimeDegraded)
	178	+	d.logSessionDegraded(record, "persisted-degraded")
173	179		return nil
174	180		case domain.RecoveryDeleting:
175	181		d.setStoppedRecovery(record, runtimeDeleting)

@@ -223,17 +229,22 @@ func (d *Daemon) restoreRecord(ctx context.Context, record domain.CatalogueRecor
223	229		}
224	230		record = outcome.Record
225	231		if outcome.HEADRepairError != nil {
226		-	d.log.Warn("fallback committed with HEAD repair pending", "session", record.Name, "err", outcome.HEADRepairError)
	232	+	d.log.Warn("snapshot_head_repair_pending", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation, "reason_code", "repair-failed")
	233	+	} else {
	234	+	d.log.Info("fallback_checkpoint_promoted", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation)
	235	+	d.log.Info("snapshot_head_repair_complete", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation)
227	236		}
228		-	d.log.Warn("restored session from fallback checkpoint", "session", record.Name, "generation", selected.Generation)
229	237		} else if err := d.snapshotRepository.RepairHEAD(ctx, record.IncarnationID, selected); err != nil {
230		-	d.log.Warn("checkpoint HEAD repair pending", "session", record.Name, "err", err)
	238	+	d.log.Warn("snapshot_head_repair_pending", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation, "reason_code", "repair-failed")
	239	+	} else {
	240	+	d.log.Info("snapshot_head_repair_complete", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation)
231	241		}
232	242
233	243		if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation, selected); err != nil {
234	244		return err
235	245		}
236	246		d.setStoppedRecovery(record, runtimeHealthy)
	247	+	d.logSessionRestoreComplete(record, selected.Generation, selectedIndex > 0)
237	248		return nil
238	249		}
239	250

 	}
 	d.mu.Unlock()
 	if restoreErr != nil {
-		d.log.Warn("session checkpoint restore failed", "session", record.Name, "err", restoreErr)
+		reasonCode := "checkpoint-invalid"
+		if errors.Is(restoreErr, context.Canceled) || errors.Is(restoreErr, context.DeadlineExceeded) {
+			reasonCode = "restore-interrupted"
+		}
+		d.logSessionDegraded(record, reasonCode)
 	}
 }
 	switch record.RecoveryState {
 	case domain.RecoveryFresh:
 		d.setStoppedRecovery(record, runtimeFresh)
+		d.logSessionRestoreComplete(record, 0, false)
 		return nil
 	case domain.RecoveryDegraded:
 		d.setStoppedRecovery(record, runtimeDegraded)
+		d.logSessionDegraded(record, "persisted-degraded")
 		return nil
 	case domain.RecoveryDeleting:
 		d.setStoppedRecovery(record, runtimeDeleting)
 		}
 		record = outcome.Record
 		if outcome.HEADRepairError != nil {
-			d.log.Warn("fallback committed with HEAD repair pending", "session", record.Name, "err", outcome.HEADRepairError)
+			d.log.Warn("snapshot_head_repair_pending", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation, "reason_code", "repair-failed")
+		} else {
+			d.log.Info("fallback_checkpoint_promoted", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation)
+			d.log.Info("snapshot_head_repair_complete", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation)
 		}
-		d.log.Warn("restored session from fallback checkpoint", "session", record.Name, "generation", selected.Generation)
 	} else if err := d.snapshotRepository.RepairHEAD(ctx, record.IncarnationID, selected); err != nil {
-		d.log.Warn("checkpoint HEAD repair pending", "session", record.Name, "err", err)
+		d.log.Warn("snapshot_head_repair_pending", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation, "reason_code", "repair-failed")
+	} else {
+		d.log.Info("snapshot_head_repair_complete", "session", record.Name, "incarnation", record.IncarnationID.String(), "generation", selected.Generation)
 	}
 	if err := d.restoreSession(ctx, selectedSnapshot, selectedGeneration.Generation, selected); err != nil {
 		return err
 	}
 	d.setStoppedRecovery(record, runtimeHealthy)
+	d.logSessionRestoreComplete(record, selected.Generation, selectedIndex > 0)
 	return nil
 }