fix(app): serialize client and offline lifecycle access

commit: 0fe8a6c5a05f353999ee500c243af96c2975dc2a
author: brice · 2mo ago · 2026-07-26 01:11 UTC
parent: 2fd50e3
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +327 -183

M internal/app/daemon_integration_test.go +130 -0


@@ -7,9 +7,12 @@ import (
7	7		"errors"
8	8		"io"
9	9		"log/slog"
	10	+	"net"
10	11		"os"
11	12		"path/filepath"
12	13		"strings"
	14	+	"sync"
	15	+	"sync/atomic"
13	16		"testing"
14	17		"time"
15	18

@@ -17,8 +20,10 @@ import (
17	20
18	21		"github.com/bnema/vev/internal/adapters/clock"
19	22		"github.com/bnema/vev/internal/adapters/ipc"
	23	+	"github.com/bnema/vev/internal/adapters/lifecycle"
20	24		"github.com/bnema/vev/internal/adapters/pty"
21	25		"github.com/bnema/vev/internal/domain"
	26	+	"github.com/bnema/vev/internal/persist"
22	27		"github.com/bnema/vev/internal/ports"
23	28		"github.com/bnema/vev/internal/usecase/daemon"
24	29		"github.com/bnema/vev/pkg/vt"

@@ -422,6 +427,131 @@ func TestIntegration_NamedSurvivesReattach(t *testing.T) {
422	427		awaitText(t, p2, sz, "MARKER")
423	428		}
424	429
	430	+	func TestMultipleClientsOneLifecycleOwner(t *testing.T) {
	431	+	dir := filepath.Join(t.TempDir(), "runtime")
	432	+	owner, err := lifecycle.TryAcquire(dir)
	433	+	require.NoError(t, err)
	434	+
	435	+	ready := make(chan struct{})
	436	+	transport := &integrationTransport{}
	437	+	dial := func(context.Context, string) (ports.Transport, error) {
	438	+	select {
	439	+	case <-ready:
	440	+	return transport, nil
	441	+	default:
	442	+	return nil, os.ErrNotExist
	443	+	}
	444	+	}
	445	+	var spawns atomic.Int32
	446	+	cfg := backoffConfig{initial: time.Millisecond, max: 2 * time.Millisecond, total: time.Second}
	447	+
	448	+	const clients = 4
	449	+	errs := make(chan error, clients)
	450	+	var wg sync.WaitGroup
	451	+	for range clients {
	452	+	wg.Add(1)
	453	+	go func() {
	454	+	defer wg.Done()
	455	+	_, err := ensureDaemonWithLifecycle(context.Background(), dir, dial, func() error {
	456	+	spawns.Add(1)
	457	+	return nil
	458	+	}, cfg)
	459	+	errs <- err
	460	+	}()
	461	+	}
	462	+	require.Never(t, func() bool { return spawns.Load() != 0 }, 50*time.Millisecond, time.Millisecond)
	463	+	close(ready)
	464	+	require.NoError(t, owner.Release())
	465	+	wg.Wait()
	466	+	close(errs)
	467	+	for err := range errs {
	468	+	require.NoError(t, err)
	469	+	}
	470	+	require.Zero(t, spawns.Load())
	471	+	}
	472	+
	473	+	func TestListWaitsForLifecycleOwner(t *testing.T) {
	474	+	stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
	475	+	t.Setenv("XDG_STATE_HOME", stateRoot)
	476	+	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
	477	+	p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
	478	+	require.NoError(t, p.Close())
	479	+	owner, err := lifecycle.TryAcquire(ipc.SocketDir())
	480	+	require.NoError(t, err)
	481	+
	482	+	done := make(chan error, 1)
	483	+	go func() { done <- runList(context.Background()) }()
	484	+	require.Never(t, func() bool { return len(done) != 0 }, 50*time.Millisecond, time.Millisecond)
	485	+	require.NoError(t, owner.Release())
	486	+	require.NoError(t, <-done)
	487	+	}
	488	+
	489	+	func TestOfflineKillWaitsForLifecycleOwner(t *testing.T) {
	490	+	stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
	491	+	t.Setenv("XDG_STATE_HOME", stateRoot)
	492	+	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
	493	+	p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
	494	+	now := time.Now().UnixNano()
	495	+	require.NoError(t, p.Save(persist.Record{Name: "named", IncarnationID: domain.IncarnationID{1}, CreatedAt: now, UpdatedAt: now, RecoveryState: domain.RecoveryFresh}))
	496	+	require.NoError(t, p.Close())
	497	+	owner, err := lifecycle.TryAcquire(ipc.SocketDir())
	498	+	require.NoError(t, err)
	499	+
	500	+	done := make(chan error, 1)
	501	+	go func() { done <- runKill(context.Background(), "named", false, false) }()
	502	+	require.Never(t, func() bool { return len(done) != 0 }, 50*time.Millisecond, time.Millisecond)
	503	+	require.NoError(t, owner.Release())
	504	+	require.NoError(t, <-done)
	505	+	}
	506	+
	507	+	func TestKillDaemonWaitsForOwnershipTransfer(t *testing.T) {
	508	+	stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
	509	+	t.Setenv("XDG_STATE_HOME", stateRoot)
	510	+	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
	511	+	owner, err := lifecycle.TryAcquire(ipc.SocketDir())
	512	+	require.NoError(t, err)
	513	+	_, served := startDaemonInDir(t, ipc.SocketDir())
	514	+
	515	+	done := make(chan error, 1)
	516	+	go func() { done <- requestDaemonStop(context.Background()) }()
	517	+	select {
	518	+	case err := <-served:
	519	+	require.NoError(t, err)
	520	+	case <-time.After(time.Second):
	521	+	t.Fatal("daemon did not stop")
	522	+	}
	523	+	require.Never(t, func() bool { return len(done) != 0 }, 50*time.Millisecond, time.Millisecond)
	524	+	require.NoError(t, owner.Release())
	525	+	require.NoError(t, <-done)
	526	+	}
	527	+
	528	+	func TestLifecycleSocketCloseCatalogueRace(t *testing.T) {
	529	+	dir := filepath.Join(t.TempDir(), "runtime")
	530	+	old, err := lifecycle.TryAcquire(dir)
	531	+	require.NoError(t, err)
	532	+	var catalogueOpen, listenCalled atomic.Bool
	533	+	newDone := make(chan error, 1)
	534	+	go func() {
	535	+	newDone <- runWithLifecycleOwner(context.Background(), dir, filepath.Join(t.TempDir(), "state"), func(context.Context) error {
	536	+	catalogueOpen.Store(true)
	537	+	listenCalled.Store(true)
	538	+	return nil
	539	+	})
	540	+	}()
	541	+	require.Never(t, func() bool { return catalogueOpen.Load() \|\| listenCalled.Load() }, 50*time.Millisecond, time.Millisecond)
	542	+	require.NoError(t, old.Release())
	543	+	require.Eventually(t, catalogueOpen.Load, time.Second, time.Millisecond)
	544	+	require.NoError(t, <-newDone)
	545	+	}
	546	+
	547	+	type integrationTransport struct{}
	548	+
	549	+	func (*integrationTransport) Send(ports.Frame) error { return nil }
	550	+	func (*integrationTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF }
	551	+	func (*integrationTransport) Close() error { return nil }
	552	+	func (*integrationTransport) LocalAddr() net.Addr { return nil }
	553	+	func (*integrationTransport) RemoteAddr() net.Addr { return nil }
	554	+
425	555		func TestIntegration_KillAllShutsDownDaemon(t *testing.T) {
426	556		sz := domain.Size{Cols: 80, Rows: 24}
427	557		dir, served := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "sleep 30"}))


@@ -7,9 +7,12 @@ import (
7	"errors"	7		"errors"
8	"io"	8		"io"
9	"log/slog"	9		"log/slog"
		10	+	"net"
10	"os"	11		"os"
11	"path/filepath"	12		"path/filepath"
12	"strings"	13		"strings"
		14	+	"sync"
		15	+	"sync/atomic"
13	"testing"	16		"testing"
14	"time"	17		"time"
15		18

@@ -17,8 +20,10 @@ import (
17		20
18	"github.com/bnema/vev/internal/adapters/clock"	21		"github.com/bnema/vev/internal/adapters/clock"
19	"github.com/bnema/vev/internal/adapters/ipc"	22		"github.com/bnema/vev/internal/adapters/ipc"
		23	+	"github.com/bnema/vev/internal/adapters/lifecycle"
20	"github.com/bnema/vev/internal/adapters/pty"	24		"github.com/bnema/vev/internal/adapters/pty"
21	"github.com/bnema/vev/internal/domain"	25		"github.com/bnema/vev/internal/domain"
		26	+	"github.com/bnema/vev/internal/persist"
22	"github.com/bnema/vev/internal/ports"	27		"github.com/bnema/vev/internal/ports"
23	"github.com/bnema/vev/internal/usecase/daemon"	28		"github.com/bnema/vev/internal/usecase/daemon"
24	"github.com/bnema/vev/pkg/vt"	29		"github.com/bnema/vev/pkg/vt"

@@ -422,6 +427,131 @@ func TestIntegration_NamedSurvivesReattach(t *testing.T) {
422	awaitText(t, p2, sz, "MARKER")	427		awaitText(t, p2, sz, "MARKER")
423	}	428		}
424		429
		430	+	func TestMultipleClientsOneLifecycleOwner(t *testing.T) {
		431	+	dir := filepath.Join(t.TempDir(), "runtime")
		432	+	owner, err := lifecycle.TryAcquire(dir)
		433	+	require.NoError(t, err)
		434	+
		435	+	ready := make(chan struct{})
		436	+	transport := &integrationTransport{}
		437	+	dial := func(context.Context, string) (ports.Transport, error) {
		438	+	select {
		439	+	case <-ready:
		440	+	return transport, nil
		441	+	default:
		442	+	return nil, os.ErrNotExist
		443	+	}
		444	+	}
		445	+	var spawns atomic.Int32
		446	+	cfg := backoffConfig{initial: time.Millisecond, max: 2 * time.Millisecond, total: time.Second}
		447	+
		448	+	const clients = 4
		449	+	errs := make(chan error, clients)
		450	+	var wg sync.WaitGroup
		451	+	for range clients {
		452	+	wg.Add(1)
		453	+	go func() {
		454	+	defer wg.Done()
		455	+	_, err := ensureDaemonWithLifecycle(context.Background(), dir, dial, func() error {
		456	+	spawns.Add(1)
		457	+	return nil
		458	+	}, cfg)
		459	+	errs <- err
		460	+	}()
		461	+	}
		462	+	require.Never(t, func() bool { return spawns.Load() != 0 }, 50*time.Millisecond, time.Millisecond)
		463	+	close(ready)
		464	+	require.NoError(t, owner.Release())
		465	+	wg.Wait()
		466	+	close(errs)
		467	+	for err := range errs {
		468	+	require.NoError(t, err)
		469	+	}
		470	+	require.Zero(t, spawns.Load())
		471	+	}
		472	+
		473	+	func TestListWaitsForLifecycleOwner(t *testing.T) {
		474	+	stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
		475	+	t.Setenv("XDG_STATE_HOME", stateRoot)
		476	+	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
		477	+	p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
		478	+	require.NoError(t, p.Close())
		479	+	owner, err := lifecycle.TryAcquire(ipc.SocketDir())
		480	+	require.NoError(t, err)
		481	+
		482	+	done := make(chan error, 1)
		483	+	go func() { done <- runList(context.Background()) }()
		484	+	require.Never(t, func() bool { return len(done) != 0 }, 50*time.Millisecond, time.Millisecond)
		485	+	require.NoError(t, owner.Release())
		486	+	require.NoError(t, <-done)
		487	+	}
		488	+
		489	+	func TestOfflineKillWaitsForLifecycleOwner(t *testing.T) {
		490	+	stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
		491	+	t.Setenv("XDG_STATE_HOME", stateRoot)
		492	+	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
		493	+	p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
		494	+	now := time.Now().UnixNano()
		495	+	require.NoError(t, p.Save(persist.Record{Name: "named", IncarnationID: domain.IncarnationID{1}, CreatedAt: now, UpdatedAt: now, RecoveryState: domain.RecoveryFresh}))
		496	+	require.NoError(t, p.Close())
		497	+	owner, err := lifecycle.TryAcquire(ipc.SocketDir())
		498	+	require.NoError(t, err)
		499	+
		500	+	done := make(chan error, 1)
		501	+	go func() { done <- runKill(context.Background(), "named", false, false) }()
		502	+	require.Never(t, func() bool { return len(done) != 0 }, 50*time.Millisecond, time.Millisecond)
		503	+	require.NoError(t, owner.Release())
		504	+	require.NoError(t, <-done)
		505	+	}
		506	+
		507	+	func TestKillDaemonWaitsForOwnershipTransfer(t *testing.T) {
		508	+	stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
		509	+	t.Setenv("XDG_STATE_HOME", stateRoot)
		510	+	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
		511	+	owner, err := lifecycle.TryAcquire(ipc.SocketDir())
		512	+	require.NoError(t, err)
		513	+	_, served := startDaemonInDir(t, ipc.SocketDir())
		514	+
		515	+	done := make(chan error, 1)
		516	+	go func() { done <- requestDaemonStop(context.Background()) }()
		517	+	select {
		518	+	case err := <-served:
		519	+	require.NoError(t, err)
		520	+	case <-time.After(time.Second):
		521	+	t.Fatal("daemon did not stop")
		522	+	}
		523	+	require.Never(t, func() bool { return len(done) != 0 }, 50*time.Millisecond, time.Millisecond)
		524	+	require.NoError(t, owner.Release())
		525	+	require.NoError(t, <-done)
		526	+	}
		527	+
		528	+	func TestLifecycleSocketCloseCatalogueRace(t *testing.T) {
		529	+	dir := filepath.Join(t.TempDir(), "runtime")
		530	+	old, err := lifecycle.TryAcquire(dir)
		531	+	require.NoError(t, err)
		532	+	var catalogueOpen, listenCalled atomic.Bool
		533	+	newDone := make(chan error, 1)
		534	+	go func() {
		535	+	newDone <- runWithLifecycleOwner(context.Background(), dir, filepath.Join(t.TempDir(), "state"), func(context.Context) error {
		536	+	catalogueOpen.Store(true)
		537	+	listenCalled.Store(true)
		538	+	return nil
		539	+	})
		540	+	}()
		541	+	require.Never(t, func() bool { return catalogueOpen.Load() \|\| listenCalled.Load() }, 50*time.Millisecond, time.Millisecond)
		542	+	require.NoError(t, old.Release())
		543	+	require.Eventually(t, catalogueOpen.Load, time.Second, time.Millisecond)
		544	+	require.NoError(t, <-newDone)
		545	+	}
		546	+
		547	+	type integrationTransport struct{}
		548	+
		549	+	func (*integrationTransport) Send(ports.Frame) error { return nil }
		550	+	func (*integrationTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF }
		551	+	func (*integrationTransport) Close() error { return nil }
		552	+	func (*integrationTransport) LocalAddr() net.Addr { return nil }
		553	+	func (*integrationTransport) RemoteAddr() net.Addr { return nil }
		554	+
425	func TestIntegration_KillAllShutsDownDaemon(t *testing.T) {	555		func TestIntegration_KillAllShutsDownDaemon(t *testing.T) {
426	sz := domain.Size{Cols: 80, Rows: 24}	556		sz := domain.Size{Cols: 80, Rows: 24}
427	dir, served := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "sleep 30"}))	557		dir, served := startDaemon(t, daemon.WithShell("/bin/sh", []string{"-c", "sleep 30"}))

M internal/app/offline_kill_retry_test.go +20 -75


@@ -2,87 +2,32 @@ package app
2	2
3	3		import (
4	4		"context"
5		-	"errors"
	5	+	"os"
6	6		"path/filepath"
7	7		"testing"
8	8		"time"
9	9
	10	+	"github.com/stretchr/testify/require"
	11	+
10	12		"github.com/bnema/vev/internal/domain"
11	13		"github.com/bnema/vev/internal/persist"
12		-	"github.com/stretchr/testify/require"
13	14		)
14	15
15		-	type offlineSnapshotFake struct {
16		-	calls []string
17		-	deleteErr error
18		-	deleteLegacyErr error
19		-	tombstoneErr error
20		-	deleteTombstoneErr error
21		-	}
22		-
23		-	func (f *offlineSnapshotFake) Tombstone(context.Context, string) error {
24		-	f.calls = append(f.calls, "tombstone")
25		-	return f.tombstoneErr
26		-	}
27		-
28		-	func (f *offlineSnapshotFake) Delete(context.Context, string) error {
29		-	f.calls = append(f.calls, "incremental")
30		-	return f.deleteErr
31		-	}
32		-
33		-	func (f *offlineSnapshotFake) DeleteLegacy(context.Context, string) error {
34		-	f.calls = append(f.calls, "legacy")
35		-	return f.deleteLegacyErr
36		-	}
37		-
38		-	func (f *offlineSnapshotFake) DeleteTombstone(context.Context, string) error {
39		-	f.calls = append(f.calls, "delete tombstone")
40		-	return f.deleteTombstoneErr
41		-	}
42		-
43		-	func TestRunKillOfflineRetriesSnapshotDeletion(t *testing.T) {
44		-	for _, tt := range []struct {
45		-	name string
46		-	incrementalErr error
47		-	legacyErr error
48		-	}{
49		-	{name: "legacy failure", legacyErr: errors.New("legacy unlink failed")},
50		-	{name: "incremental failure", incrementalErr: errors.New("incremental quarantine failed")},
51		-	{name: "both failures", incrementalErr: errors.New("incremental quarantine failed"), legacyErr: errors.New("legacy unlink failed")},
52		-	} {
53		-	t.Run(tt.name, func(t *testing.T) {
54		-	stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
55		-	t.Setenv("XDG_STATE_HOME", stateRoot)
56		-	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
57		-
58		-	p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
59		-	now := time.Now().UnixNano()
60		-	require.NoError(t, p.Save(persist.Record{Name: "named", IncarnationID: domain.IncarnationID{1}, Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now, RecoveryState: domain.RecoveryFresh}))
61		-	require.NoError(t, p.Close())
62		-
63		-	fake := &offlineSnapshotFake{deleteErr: tt.incrementalErr, deleteLegacyErr: tt.legacyErr}
64		-	originalFactory := newOfflineSnapshotRepository
65		-	newOfflineSnapshotRepository = func(string) offlineSnapshotSource { return fake }
66		-	t.Cleanup(func() { newOfflineSnapshotRepository = originalFactory })
67		-
68		-	err := runKill(context.Background(), "named", false, false)
69		-	if tt.incrementalErr != nil {
70		-	require.ErrorIs(t, err, tt.incrementalErr)
71		-	}
72		-	if tt.legacyErr != nil {
73		-	require.ErrorIs(t, err, tt.legacyErr)
74		-	}
75		-	require.Equal(t, []string{"tombstone", "incremental", "legacy"}, fake.calls)
76		-	records, err := persist.LoadReadOnly(filepath.Join(stateRoot, "vev"))
77		-	require.NoError(t, err)
78		-	require.Len(t, records, 1, "a failed source delete must retain retry identity")
79		-
80		-	fake.deleteErr, fake.deleteLegacyErr = nil, nil
81		-	require.NoError(t, runKill(context.Background(), "named", false, false))
82		-	require.Equal(t, []string{"tombstone", "incremental", "legacy", "tombstone", "incremental", "legacy", "delete tombstone"}, fake.calls)
83		-	records, err = persist.LoadReadOnly(filepath.Join(stateRoot, "vev"))
84		-	require.NoError(t, err)
85		-	require.Empty(t, records)
86		-	})
87		-	}
	16	+	func TestRunKillOfflineUsesIncarnationDeletionProtocol(t *testing.T) {
	17	+	stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
	18	+	t.Setenv("XDG_STATE_HOME", stateRoot)
	19	+	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
	20	+
	21	+	id := domain.IncarnationID{1}
	22	+	p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
	23	+	now := time.Now().UnixNano()
	24	+	require.NoError(t, p.Save(persist.Record{Name: "named", IncarnationID: id, Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now, RecoveryState: domain.RecoveryFresh}))
	25	+	require.NoError(t, p.Close())
	26	+	source := filepath.Join(stateRoot, "vev", "snapshots", "sessions", id.String())
	27	+	require.NoError(t, os.MkdirAll(source, 0o700))
	28	+	require.NoError(t, os.WriteFile(filepath.Join(source, "sentinel"), []byte("durable"), 0o600))
	29	+
	30	+	require.NoError(t, runKill(context.Background(), "named", false, false))
	31	+	require.NoFileExists(t, filepath.Join(source, "sentinel"))
	32	+	require.FileExists(t, filepath.Join(stateRoot, "vev", "snapshots", "quarantine", id.String(), "snapshot", "sentinel"))
88	33		}


@@ -2,87 +2,32 @@ package app
2			2
3		import (	3		import (
4		"context"	4		"context"
5	-	"errors"	5	+	"os"
6		"path/filepath"	6		"path/filepath"
7		"testing"	7		"testing"
8		"time"	8		"time"
9			9
			10	+	"github.com/stretchr/testify/require"
			11	+
10		"github.com/bnema/vev/internal/domain"	12		"github.com/bnema/vev/internal/domain"
11		"github.com/bnema/vev/internal/persist"	13		"github.com/bnema/vev/internal/persist"
12	-	"github.com/stretchr/testify/require"
13		)	14		)
14			15
15	-	type offlineSnapshotFake struct {	16	+	func TestRunKillOfflineUsesIncarnationDeletionProtocol(t *testing.T) {
16	-	calls []string	17	+	stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
17	-	deleteErr error	18	+	t.Setenv("XDG_STATE_HOME", stateRoot)
18	-	deleteLegacyErr error	19	+	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
19	-	tombstoneErr error	20	+
20	-	deleteTombstoneErr error	21	+	id := domain.IncarnationID{1}
21	-	}	22	+	p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
22	-		23	+	now := time.Now().UnixNano()
23	-	func (f *offlineSnapshotFake) Tombstone(context.Context, string) error {	24	+	require.NoError(t, p.Save(persist.Record{Name: "named", IncarnationID: id, Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now, RecoveryState: domain.RecoveryFresh}))
24	-	f.calls = append(f.calls, "tombstone")	25	+	require.NoError(t, p.Close())
25	-	return f.tombstoneErr	26	+	source := filepath.Join(stateRoot, "vev", "snapshots", "sessions", id.String())
26	-	}	27	+	require.NoError(t, os.MkdirAll(source, 0o700))
27	-		28	+	require.NoError(t, os.WriteFile(filepath.Join(source, "sentinel"), []byte("durable"), 0o600))
28	-	func (f *offlineSnapshotFake) Delete(context.Context, string) error {	29	+
29	-	f.calls = append(f.calls, "incremental")	30	+	require.NoError(t, runKill(context.Background(), "named", false, false))
30	-	return f.deleteErr	31	+	require.NoFileExists(t, filepath.Join(source, "sentinel"))
31	-	}	32	+	require.FileExists(t, filepath.Join(stateRoot, "vev", "snapshots", "quarantine", id.String(), "snapshot", "sentinel"))
32	-
33	-	func (f *offlineSnapshotFake) DeleteLegacy(context.Context, string) error {
34	-	f.calls = append(f.calls, "legacy")
35	-	return f.deleteLegacyErr
36	-	}
37	-
38	-	func (f *offlineSnapshotFake) DeleteTombstone(context.Context, string) error {
39	-	f.calls = append(f.calls, "delete tombstone")
40	-	return f.deleteTombstoneErr
41	-	}
42	-
43	-	func TestRunKillOfflineRetriesSnapshotDeletion(t *testing.T) {
44	-	for _, tt := range []struct {
45	-	name string
46	-	incrementalErr error
47	-	legacyErr error
48	-	}{
49	-	{name: "legacy failure", legacyErr: errors.New("legacy unlink failed")},
50	-	{name: "incremental failure", incrementalErr: errors.New("incremental quarantine failed")},
51	-	{name: "both failures", incrementalErr: errors.New("incremental quarantine failed"), legacyErr: errors.New("legacy unlink failed")},
52	-	} {
53	-	t.Run(tt.name, func(t *testing.T) {
54	-	stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
55	-	t.Setenv("XDG_STATE_HOME", stateRoot)
56	-	t.Setenv("XDG_RUNTIME_DIR", runtimeRoot)
57	-
58	-	p := newTestPersister(t, filepath.Join(stateRoot, "vev"))
59	-	now := time.Now().UnixNano()
60	-	require.NoError(t, p.Save(persist.Record{Name: "named", IncarnationID: domain.IncarnationID{1}, Cwd: t.TempDir(), CreatedAt: now, UpdatedAt: now, RecoveryState: domain.RecoveryFresh}))
61	-	require.NoError(t, p.Close())
62	-
63	-	fake := &offlineSnapshotFake{deleteErr: tt.incrementalErr, deleteLegacyErr: tt.legacyErr}
64	-	originalFactory := newOfflineSnapshotRepository
65	-	newOfflineSnapshotRepository = func(string) offlineSnapshotSource { return fake }
66	-	t.Cleanup(func() { newOfflineSnapshotRepository = originalFactory })
67	-
68	-	err := runKill(context.Background(), "named", false, false)
69	-	if tt.incrementalErr != nil {
70	-	require.ErrorIs(t, err, tt.incrementalErr)
71	-	}
72	-	if tt.legacyErr != nil {
73	-	require.ErrorIs(t, err, tt.legacyErr)
74	-	}
75	-	require.Equal(t, []string{"tombstone", "incremental", "legacy"}, fake.calls)
76	-	records, err := persist.LoadReadOnly(filepath.Join(stateRoot, "vev"))
77	-	require.NoError(t, err)
78	-	require.Len(t, records, 1, "a failed source delete must retain retry identity")
79	-
80	-	fake.deleteErr, fake.deleteLegacyErr = nil, nil
81	-	require.NoError(t, runKill(context.Background(), "named", false, false))
82	-	require.Equal(t, []string{"tombstone", "incremental", "legacy", "tombstone", "incremental", "legacy", "delete tombstone"}, fake.calls)
83	-	records, err = persist.LoadReadOnly(filepath.Join(stateRoot, "vev"))
84	-	require.NoError(t, err)
85	-	require.Empty(t, records)
86	-	})
87	-	}
88		}	33		}

M internal/app/run.go +74 -105


@@ -759,7 +759,7 @@ func (d localDaemonDialer) Dial(ctx context.Context) (ports.Transport, error) {
759	759		return ipc.DialContext(ctx, dir, ipc.WithRuntimeObserver(d.observer))
760	760		}
761	761		}
762		-	return ensureDaemon(ctx, d.dir, dial, realSpawn, defaultBackoff)
	762	+	return ensureDaemonWithLifecycle(ctx, d.dir, dial, realSpawn, defaultBackoff)
763	763		}
764	764
765	765		func detachedLocalHello(name, cwd string) ports.Hello {

@@ -777,7 +777,7 @@ func detachedLocalHello(name, cwd string) ports.Hello {
777	777		}
778	778
779	779		func createDetachedLocalSession(ctx context.Context, name string) error {
780		-	transport, err := ensureDaemon(ctx, ipc.SocketDir(), realDial, realSpawn, defaultBackoff)
	780	+	transport, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), realDial, realSpawn, defaultBackoff)
781	781		if err != nil {
782	782		return err
783	783		}

@@ -840,40 +840,26 @@ func requestDaemonStop(ctx context.Context) error {
840	840		ctx, cancel := context.WithTimeout(ctx, daemonStopTimeout)
841	841		defer cancel()
842	842
843		-	done := make(chan error, 1)
844		-	go func() {
845		-	transport, err := realDial(ctx, ipc.SocketDir())
846		-	if err != nil {
847		-	done <- fmt.Errorf("vev: no daemon running")
848		-	return
849		-	}
850		-	defer func() { _ = transport.Close() }()
851		-	closed := make(chan struct{})
852		-	go func() {
853		-	select {
854		-	case <-ctx.Done():
855		-	_ = transport.Close()
856		-	case <-closed:
857		-	}
858		-	}()
859		-	defer close(closed)
860		-	if err := transport.Send(ports.Frame{Type: ports.MsgKill, Payload: ports.MarshalKill(ports.Kill{All: true})}); err != nil {
861		-	done <- fmt.Errorf("vev: requesting daemon stop: %w", err)
862		-	return
863		-	}
864		-	if _, err := transport.Recv(); err != nil && !errors.Is(err, io.EOF) {
865		-	done <- fmt.Errorf("vev: reading daemon stop reply: %w", err)
866		-	return
867		-	}
868		-	done <- nil
869		-	}()
	843	+	transport, owner, err := waitForDaemonOrLifecycle(ctx, ipc.SocketDir(), realDial, defaultBackoff)
	844	+	if err != nil {
	845	+	return fmt.Errorf("vev: stopping daemon: %w", err)
	846	+	}
	847	+	if owner != nil {
	848	+	return errors.Join(errors.New("vev: no daemon running"), owner.Release())
	849	+	}
	850	+	defer func() { _ = transport.Close() }()
	851	+	if err := transport.Send(ports.Frame{Type: ports.MsgKill, Payload: ports.MarshalKill(ports.Kill{All: true})}); err != nil {
	852	+	return fmt.Errorf("vev: requesting daemon stop: %w", err)
	853	+	}
	854	+	if _, err := transport.Recv(); err != nil && !errors.Is(err, io.EOF) {
	855	+	return fmt.Errorf("vev: reading daemon stop reply: %w", err)
	856	+	}
870	857
871		-	select {
872		-	case err := <-done:
873		-	return err
874		-	case <-ctx.Done():
875		-	return fmt.Errorf("vev: stopping daemon: %w", ctx.Err())
	858	+	owner, err = waitForLifecycleAvailability(ctx, ipc.SocketDir(), defaultBackoff)
	859	+	if err != nil {
	860	+	return fmt.Errorf("vev: waiting for daemon ownership transfer: %w", err)
876	861		}
	862	+	return owner.Release()
877	863		}
878	864
879	865		// runStdio is the hidden remote-side mode used by `ssh host vev _stdio`: it

@@ -895,7 +881,7 @@ func runStdio(ctx context.Context) (retErr error) {
895	881		if observerCloser != nil {
896	882		defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }()
897	883		}
898		-	transport, err := ensureDaemon(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (ports.Transport, error) {
	884	+	transport, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (ports.Transport, error) {
899	885		return ipc.DialContext(ctx, dir, ipc.WithRuntimeObserver(observer))
900	886		}, realSpawn, defaultBackoff)
901	887		if err != nil {

@@ -1010,7 +996,7 @@ func runUDPProxy(ctx context.Context, session string, ready io.Writer) (retErr e
1010	996		return err
1011	997		}
1012	998		defer func() { _ = conn.Close() }()
1013		-	daemonTr, err := ensureDaemon(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (ports.Transport, error) {
	999	+	daemonTr, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (ports.Transport, error) {
1014	1000		return ipc.DialContext(ctx, dir, ipc.WithRuntimeObserver(observer))
1015	1001		}, realSpawn, defaultBackoff)
1016	1002		if err != nil {

@@ -1228,8 +1214,12 @@ func proxyTransports(ctx context.Context, a, b ports.Transport, log *slog.Logger
1228	1214		// runList prints the daemon's session listing. With no daemon running, it
1229	1215		// falls back to the persisted stopped-session records.
1230	1216		func runList(ctx context.Context) error {
1231		-	transport, err := realDial(ctx, ipc.SocketDir())
	1217	+	transport, owner, err := waitForDaemonOrLifecycle(ctx, ipc.SocketDir(), realDial, defaultBackoff)
1232	1218		if err != nil {
	1219	+	return fmt.Errorf("vev: waiting for durable session state: %w", err)
	1220	+	}
	1221	+	if owner != nil {
	1222	+	defer func() { _ = owner.Release() }()
1233	1223		records, loadErr := persist.LoadReadOnly(platform.StateDir())
1234	1224		if loadErr != nil {
1235	1225		return fmt.Errorf("vev: reading stored sessions: %w", loadErr)

@@ -1304,83 +1294,62 @@ func printSessions(w io.Writer, sessions []ports.SessionInfo) {
1304	1294		_ = tw.Flush()
1305	1295		}
1306	1296
1307		-	// offlineSnapshotSource owns the durable source and tombstone transitions for
1308		-	// a daemon-less kill. It is intentionally local to the composition root: the
1309		-	// daemon uses the repository through its narrower ports.
1310		-	type offlineSnapshotSource interface {
1311		-	Tombstone(context.Context, string) error
1312		-	Delete(context.Context, string) error
1313		-	DeleteLegacy(context.Context, string) error
1314		-	DeleteTombstone(context.Context, string) error
1315		-	}
	1297	+	func runOfflineNamedKill(ctx context.Context, name string) (retErr error) {
	1298	+	if _, err := os.Stat(persist.StorePath(platform.StateDir())); errors.Is(err, os.ErrNotExist) {
	1299	+	return fmt.Errorf("vev: no such session: %s", name)
	1300	+	} else if err != nil {
	1301	+	return fmt.Errorf("vev: reading stored sessions: %w", err)
	1302	+	}
	1303	+	records, err := persist.LoadReadOnly(platform.StateDir())
	1304	+	if err != nil {
	1305	+	return fmt.Errorf("vev: reading stored sessions: %w", err)
	1306	+	}
	1307	+	found := false
	1308	+	for _, record := range records {
	1309	+	if record.Name == name {
	1310	+	found = true
	1311	+	break
	1312	+	}
	1313	+	}
	1314	+	if !found {
	1315	+	return fmt.Errorf("vev: no such session: %s", name)
	1316	+	}
1316	1317
1317		-	var newOfflineSnapshotRepository = func(dir string) offlineSnapshotSource {
1318		-	return snapshotadapter.NewRepository(dir)
1319		-	}
	1318	+	repository := snapshotadapter.NewRepository(snapshotDir())
	1319	+	opened, err := persist.OpenOrMigrate(ctx, persist.OpenDeps{
	1320	+	StateDir: platform.StateDir(),
	1321	+	Random: rand.Reader,
	1322	+	SnapshotMigration: repository,
	1323	+	})
	1324	+	if err != nil {
	1325	+	return fmt.Errorf("vev: opening stored sessions: %w", err)
	1326	+	}
	1327	+	defer func() { retErr = errors.Join(retErr, opened.Catalogue.Close()) }()
1320	1328
1321		-	func phaseError(phase string, err error) error {
1322		-	if err == nil {
1323		-	return nil
	1329	+	if _, ok := opened.Catalogue.Record(name); !ok {
	1330	+	return fmt.Errorf("vev: no such session: %s", name)
	1331	+	}
	1332	+	coordinator := recovery.NewCoordinator(opened.Catalogue, repository, recoveryfs.New(platform.StateDir()), rand.Reader)
	1333	+	if err := coordinator.Recover(ctx); err != nil {
	1334	+	return fmt.Errorf("vev: recovering stored session transactions: %w", err)
	1335	+	}
	1336	+	if err := coordinator.Delete(ctx, name); err != nil {
	1337	+	return fmt.Errorf("vev: deleting stored session: %w", err)
1324	1338		}
1325		-	return fmt.Errorf("vev: %s: %w", phase, err)
	1339	+	return nil
1326	1340		}
1327	1341
1328	1342		// runKill asks the daemon to terminate a named session, every session, or the daemon.
1329	1343		func runKill(ctx context.Context, name string, all, daemon bool) error {
1330		-	transport, err := realDial(ctx, ipc.SocketDir())
	1344	+	transport, owner, err := waitForDaemonOrLifecycle(ctx, ipc.SocketDir(), realDial, defaultBackoff)
1331	1345		if err != nil {
	1346	+	return fmt.Errorf("vev: waiting for durable session state: %w", err)
	1347	+	}
	1348	+	if owner != nil {
	1349	+	defer func() { _ = owner.Release() }()
1332	1350		if name != "" && !all && !daemon {
1333		-	storePath := persist.StorePath(platform.StateDir())
1334		-	if _, statErr := os.Stat(storePath); errors.Is(statErr, os.ErrNotExist) {
1335		-	return fmt.Errorf("vev: no such session: %s", name)
1336		-	} else if statErr != nil {
1337		-	return fmt.Errorf("vev: reading stored sessions: %w", statErr)
1338		-	}
1339		-	records, loadErr := persist.LoadReadOnly(platform.StateDir())
1340		-	if loadErr != nil {
1341		-	return fmt.Errorf("vev: reading stored sessions: %w", loadErr)
1342		-	}
1343		-	found := false
1344		-	for _, record := range records {
1345		-	if record.Name == name {
1346		-	found = true
1347		-	break
1348		-	}
1349		-	}
1350		-	if !found {
1351		-	return fmt.Errorf("vev: no such session: %s", name)
1352		-	}
1353		-	p, openErr := persist.Open(platform.StateDir())
1354		-	if openErr != nil {
1355		-	return fmt.Errorf("vev: opening stored sessions: %w", openErr)
1356		-	}
1357		-	snapshots := newOfflineSnapshotRepository(snapshotDir())
1358		-	// The durable marker is the commit point for a logical kill. Keep it and
1359		-	// the persisted record until both independently recoverable sources have
1360		-	// crossed their deletion durability boundaries.
1361		-	if tombstoneErr := snapshots.Tombstone(ctx, name); tombstoneErr != nil {
1362		-	_ = p.Close()
1363		-	return fmt.Errorf("vev: marking session killed: %w", tombstoneErr)
1364		-	}
1365		-	incrementalErr := snapshots.Delete(ctx, name)
1366		-	legacyErr := snapshots.DeleteLegacy(ctx, name)
1367		-	if incrementalErr != nil \|\| legacyErr != nil {
1368		-	_ = p.Close()
1369		-	return errors.Join(
1370		-	phaseError("deleting incremental session snapshot", incrementalErr),
1371		-	phaseError("deleting legacy session snapshot", legacyErr),
1372		-	)
1373		-	}
1374		-	if deleteErr := p.Delete(name); deleteErr != nil {
1375		-	_ = p.Close()
1376		-	return fmt.Errorf("vev: deleting stored session: %w", deleteErr)
1377		-	}
1378		-	if tombstoneErr := snapshots.DeleteTombstone(ctx, name); tombstoneErr != nil {
1379		-	_ = p.Close()
1380		-	return fmt.Errorf("vev: clearing killed session marker: %w", tombstoneErr)
1381		-	}
1382		-	if closeErr := p.Close(); closeErr != nil {
1383		-	return fmt.Errorf("vev: closing stored sessions: %w", closeErr)
	1351	+	if err := runOfflineNamedKill(ctx, name); err != nil {
	1352	+	return err
1384	1353		}
1385	1354		printKillSuccess(name, all, daemon)
1386	1355		return nil

 			return ipc.DialContext(ctx, dir, ipc.WithRuntimeObserver(d.observer))
 		}
 	}
-	return ensureDaemon(ctx, d.dir, dial, realSpawn, defaultBackoff)
+	return ensureDaemonWithLifecycle(ctx, d.dir, dial, realSpawn, defaultBackoff)
 }
 func detachedLocalHello(name, cwd string) ports.Hello {
 }
 func createDetachedLocalSession(ctx context.Context, name string) error {
-	transport, err := ensureDaemon(ctx, ipc.SocketDir(), realDial, realSpawn, defaultBackoff)
+	transport, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), realDial, realSpawn, defaultBackoff)
 	if err != nil {
 		return err
 	}
 	ctx, cancel := context.WithTimeout(ctx, daemonStopTimeout)
 	defer cancel()
-	done := make(chan error, 1)
+	transport, owner, err := waitForDaemonOrLifecycle(ctx, ipc.SocketDir(), realDial, defaultBackoff)
-	go func() {
+	if err != nil {
-		transport, err := realDial(ctx, ipc.SocketDir())
+		return fmt.Errorf("vev: stopping daemon: %w", err)
-		if err != nil {
+	}
-			done <- fmt.Errorf("vev: no daemon running")
+	if owner != nil {
-			return
+		return errors.Join(errors.New("vev: no daemon running"), owner.Release())
-		}
+	}
-		defer func() { _ = transport.Close() }()
+	defer func() { _ = transport.Close() }()
-		closed := make(chan struct{})
+	if err := transport.Send(ports.Frame{Type: ports.MsgKill, Payload: ports.MarshalKill(ports.Kill{All: true})}); err != nil {
-		go func() {
+		return fmt.Errorf("vev: requesting daemon stop: %w", err)
-			select {
+	}
-			case <-ctx.Done():
+	if _, err := transport.Recv(); err != nil && !errors.Is(err, io.EOF) {
-				_ = transport.Close()
+		return fmt.Errorf("vev: reading daemon stop reply: %w", err)
-			case <-closed:
+	}
-			}
-		}()
-		defer close(closed)
-		if err := transport.Send(ports.Frame{Type: ports.MsgKill, Payload: ports.MarshalKill(ports.Kill{All: true})}); err != nil {
-			done <- fmt.Errorf("vev: requesting daemon stop: %w", err)
-			return
-		}
-		if _, err := transport.Recv(); err != nil && !errors.Is(err, io.EOF) {
-			done <- fmt.Errorf("vev: reading daemon stop reply: %w", err)
-			return
-		}
-		done <- nil
-	}()
-	select {
+	owner, err = waitForLifecycleAvailability(ctx, ipc.SocketDir(), defaultBackoff)
-	case err := <-done:
+	if err != nil {
-		return err
+		return fmt.Errorf("vev: waiting for daemon ownership transfer: %w", err)
-	case <-ctx.Done():
-		return fmt.Errorf("vev: stopping daemon: %w", ctx.Err())
 	}
+	return owner.Release()
 }
 // runStdio is the hidden remote-side mode used by `ssh host vev _stdio`: it
 	if observerCloser != nil {
 		defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }()
 	}
-	transport, err := ensureDaemon(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (ports.Transport, error) {
+	transport, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (ports.Transport, error) {
 		return ipc.DialContext(ctx, dir, ipc.WithRuntimeObserver(observer))
 	}, realSpawn, defaultBackoff)
 	if err != nil {
 		return err
 	}
 	defer func() { _ = conn.Close() }()
-	daemonTr, err := ensureDaemon(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (ports.Transport, error) {
+	daemonTr, err := ensureDaemonWithLifecycle(ctx, ipc.SocketDir(), func(ctx context.Context, dir string) (ports.Transport, error) {
 		return ipc.DialContext(ctx, dir, ipc.WithRuntimeObserver(observer))
 	}, realSpawn, defaultBackoff)
 	if err != nil {
 // runList prints the daemon's session listing. With no daemon running, it
 // falls back to the persisted stopped-session records.
 func runList(ctx context.Context) error {
-	transport, err := realDial(ctx, ipc.SocketDir())
+	transport, owner, err := waitForDaemonOrLifecycle(ctx, ipc.SocketDir(), realDial, defaultBackoff)
 	if err != nil {
+		return fmt.Errorf("vev: waiting for durable session state: %w", err)
+	}
+	if owner != nil {
+		defer func() { _ = owner.Release() }()
 		records, loadErr := persist.LoadReadOnly(platform.StateDir())
 		if loadErr != nil {
 			return fmt.Errorf("vev: reading stored sessions: %w", loadErr)
 	_ = tw.Flush()
 }
-// offlineSnapshotSource owns the durable source and tombstone transitions for
+func runOfflineNamedKill(ctx context.Context, name string) (retErr error) {
-// a daemon-less kill. It is intentionally local to the composition root: the
+	if _, err := os.Stat(persist.StorePath(platform.StateDir())); errors.Is(err, os.ErrNotExist) {
-// daemon uses the repository through its narrower ports.
+		return fmt.Errorf("vev: no such session: %s", name)
-type offlineSnapshotSource interface {
+	} else if err != nil {
-	Tombstone(context.Context, string) error
+		return fmt.Errorf("vev: reading stored sessions: %w", err)
-	Delete(context.Context, string) error
+	}
-	DeleteLegacy(context.Context, string) error
+	records, err := persist.LoadReadOnly(platform.StateDir())
-	DeleteTombstone(context.Context, string) error
+	if err != nil {
-}
+		return fmt.Errorf("vev: reading stored sessions: %w", err)
+	}
+	found := false
+	for _, record := range records {
+		if record.Name == name {
+			found = true
+			break
+		}
+	}
+	if !found {
+		return fmt.Errorf("vev: no such session: %s", name)
+	}
-var newOfflineSnapshotRepository = func(dir string) offlineSnapshotSource {
+	repository := snapshotadapter.NewRepository(snapshotDir())
-	return snapshotadapter.NewRepository(dir)
+	opened, err := persist.OpenOrMigrate(ctx, persist.OpenDeps{
-}
+		StateDir:          platform.StateDir(),
+		Random:            rand.Reader,
+		SnapshotMigration: repository,
+	})
+	if err != nil {
+		return fmt.Errorf("vev: opening stored sessions: %w", err)
+	}
+	defer func() { retErr = errors.Join(retErr, opened.Catalogue.Close()) }()
-func phaseError(phase string, err error) error {
+	if _, ok := opened.Catalogue.Record(name); !ok {
-	if err == nil {
+		return fmt.Errorf("vev: no such session: %s", name)
-		return nil
+	}
+	coordinator := recovery.NewCoordinator(opened.Catalogue, repository, recoveryfs.New(platform.StateDir()), rand.Reader)
+	if err := coordinator.Recover(ctx); err != nil {
+		return fmt.Errorf("vev: recovering stored session transactions: %w", err)
+	}
+	if err := coordinator.Delete(ctx, name); err != nil {
+		return fmt.Errorf("vev: deleting stored session: %w", err)
 	}
-	return fmt.Errorf("vev: %s: %w", phase, err)
+	return nil
 }
 // runKill asks the daemon to terminate a named session, every session, or the daemon.
 func runKill(ctx context.Context, name string, all, daemon bool) error {
-	transport, err := realDial(ctx, ipc.SocketDir())
+	transport, owner, err := waitForDaemonOrLifecycle(ctx, ipc.SocketDir(), realDial, defaultBackoff)
 	if err != nil {
+		return fmt.Errorf("vev: waiting for durable session state: %w", err)
+	}
+	if owner != nil {
+		defer func() { _ = owner.Release() }()
 		if name != "" && !all && !daemon {
-			storePath := persist.StorePath(platform.StateDir())
+			if err := runOfflineNamedKill(ctx, name); err != nil {
-			if _, statErr := os.Stat(storePath); errors.Is(statErr, os.ErrNotExist) {
+				return err
-				return fmt.Errorf("vev: no such session: %s", name)
-			} else if statErr != nil {
-				return fmt.Errorf("vev: reading stored sessions: %w", statErr)
-			}
-			records, loadErr := persist.LoadReadOnly(platform.StateDir())
-			if loadErr != nil {
-				return fmt.Errorf("vev: reading stored sessions: %w", loadErr)
-			}
-			found := false
-			for _, record := range records {
-				if record.Name == name {
-					found = true
-					break
-				}
-			}
-			if !found {
-				return fmt.Errorf("vev: no such session: %s", name)
-			}
-			p, openErr := persist.Open(platform.StateDir())
-			if openErr != nil {
-				return fmt.Errorf("vev: opening stored sessions: %w", openErr)
-			}
-			snapshots := newOfflineSnapshotRepository(snapshotDir())
-			// The durable marker is the commit point for a logical kill. Keep it and
-			// the persisted record until both independently recoverable sources have
-			// crossed their deletion durability boundaries.
-			if tombstoneErr := snapshots.Tombstone(ctx, name); tombstoneErr != nil {
-				_ = p.Close()
-				return fmt.Errorf("vev: marking session killed: %w", tombstoneErr)
-			}
-			incrementalErr := snapshots.Delete(ctx, name)
-			legacyErr := snapshots.DeleteLegacy(ctx, name)
-			if incrementalErr != nil || legacyErr != nil {
-				_ = p.Close()
-				return errors.Join(
-					phaseError("deleting incremental session snapshot", incrementalErr),
-					phaseError("deleting legacy session snapshot", legacyErr),
-				)
-			}
-			if deleteErr := p.Delete(name); deleteErr != nil {
-				_ = p.Close()
-				return fmt.Errorf("vev: deleting stored session: %w", deleteErr)
-			}
-			if tombstoneErr := snapshots.DeleteTombstone(ctx, name); tombstoneErr != nil {
-				_ = p.Close()
-				return fmt.Errorf("vev: clearing killed session marker: %w", tombstoneErr)
-			}
-			if closeErr := p.Close(); closeErr != nil {
-				return fmt.Errorf("vev: closing stored sessions: %w", closeErr)
 			}
 			printKillSuccess(name, all, daemon)
 			return nil

M internal/app/spawn.go +103 -3


@@ -13,6 +13,7 @@ import (
13	13		"time"
14	14
15	15		"github.com/bnema/vev/internal/adapters/ipc"
	16	+	"github.com/bnema/vev/internal/adapters/lifecycle"
16	17		"github.com/bnema/vev/internal/platform"
17	18		"github.com/bnema/vev/internal/ports"
18	19		"github.com/bnema/vev/pkg/safedir"

@@ -39,6 +40,18 @@ type dialFunc func(ctx context.Context, dir string) (ports.Transport, error)
39	40		// re-exec a real binary.
40	41		type spawnFunc func() error
41	42
	43	+	type lifecycleProbe interface {
	44	+	TryAcquire(string) (*lifecycle.Owner, error)
	45	+	}
	46	+
	47	+	type osLifecycleProbe struct{}
	48	+
	49	+	func (osLifecycleProbe) TryAcquire(runtimeDir string) (*lifecycle.Owner, error) {
	50	+	return lifecycle.TryAcquire(runtimeDir)
	51	+	}
	52	+
	53	+	var daemonLifecycleProbe lifecycleProbe = osLifecycleProbe{}
	54	+
42	55		// backoffConfig parameterises retry-dial timing so tests can shrink it.
43	56		type backoffConfig struct {
44	57		initial time.Duration

@@ -54,10 +67,97 @@ var defaultBackoff = backoffConfig{
54	67		total: 5 * time.Second,
55	68		}
56	69
	70	+	func waitForLifecycleAvailability(ctx context.Context, runtimeDir string, cfg backoffConfig) (*lifecycle.Owner, error) {
	71	+	deadline := time.Now().Add(cfg.total)
	72	+	backoff := cfg.initial
	73	+	for {
	74	+	if err := ctx.Err(); err != nil {
	75	+	return nil, err
	76	+	}
	77	+	owner, err := daemonLifecycleProbe.TryAcquire(runtimeDir)
	78	+	if err == nil {
	79	+	return owner, nil
	80	+	}
	81	+	if !errors.Is(err, lifecycle.ErrBusy) {
	82	+	return nil, err
	83	+	}
	84	+	if !time.Now().Before(deadline) {
	85	+	return nil, ErrDaemonUnreachable
	86	+	}
	87	+	if err := waitBackoff(ctx, backoff); err != nil {
	88	+	return nil, err
	89	+	}
	90	+	if backoff *= 2; backoff > cfg.max {
	91	+	backoff = cfg.max
	92	+	}
	93	+	}
	94	+	}
	95	+
	96	+	func waitForDaemonOrLifecycle(ctx context.Context, dir string, dial dialFunc, cfg backoffConfig) (ports.Transport, *lifecycle.Owner, error) {
	97	+	deadline := time.Now().Add(cfg.total)
	98	+	backoff := cfg.initial
	99	+	for {
	100	+	if err := ctx.Err(); err != nil {
	101	+	return nil, nil, err
	102	+	}
	103	+	if transport, err := dial(ctx, dir); err == nil {
	104	+	return transport, nil, nil
	105	+	}
	106	+	owner, err := daemonLifecycleProbe.TryAcquire(dir)
	107	+	if err == nil {
	108	+	return nil, owner, nil
	109	+	}
	110	+	if !errors.Is(err, lifecycle.ErrBusy) {
	111	+	return nil, nil, err
	112	+	}
	113	+	if !time.Now().Before(deadline) {
	114	+	return nil, nil, ErrDaemonUnreachable
	115	+	}
	116	+	if err := waitBackoff(ctx, backoff); err != nil {
	117	+	return nil, nil, err
	118	+	}
	119	+	if backoff *= 2; backoff > cfg.max {
	120	+	backoff = cfg.max
	121	+	}
	122	+	}
	123	+	}
	124	+
	125	+	func waitBackoff(ctx context.Context, duration time.Duration) error {
	126	+	timer := time.NewTimer(duration)
	127	+	defer timer.Stop()
	128	+	select {
	129	+	case <-ctx.Done():
	130	+	return ctx.Err()
	131	+	case <-timer.C:
	132	+	return nil
	133	+	}
	134	+	}
	135	+
	136	+	// ensureDaemonWithLifecycle never elects a spawner while another lifecycle
	137	+	// owner may still be initializing or tearing down durable state.
	138	+	func ensureDaemonWithLifecycle(ctx context.Context, dir string, dial dialFunc, spawn spawnFunc, cfg backoffConfig) (ports.Transport, error) {
	139	+	if err := ctx.Err(); err != nil {
	140	+	return nil, err
	141	+	}
	142	+	if transport, err := dial(ctx, dir); err == nil {
	143	+	return transport, nil
	144	+	}
	145	+	transport, owner, err := waitForDaemonOrLifecycle(ctx, dir, dial, cfg)
	146	+	if err != nil {
	147	+	return nil, err
	148	+	}
	149	+	if transport != nil {
	150	+	return transport, nil
	151	+	}
	152	+	if err := owner.Release(); err != nil {
	153	+	return nil, fmt.Errorf("vev: release lifecycle spawn probe: %w", err)
	154	+	}
	155	+	return ensureDaemon(ctx, dir, dial, spawn, cfg)
	156	+	}
	157	+
57	158		// ensureDaemon returns a transport to a running daemon, spawning one if
58		-	// necessary. It first tries a plain dial; on failure it elects a single
59		-	// spawner via an mkdir lock (losers simply wait), then retry-dials with
60		-	// backoff until the socket is live or the budget expires.
	159	+	// necessary after ensureDaemonWithLifecycle has established lifecycle
	160	+	// availability. Tests also exercise this lower-level spawn election directly.
61	161		func ensureDaemon(ctx context.Context, dir string, dial dialFunc, spawn spawnFunc, cfg backoffConfig) (ports.Transport, error) {
62	162		if err := ctx.Err(); err != nil {
63	163		return nil, err


@@ -13,6 +13,7 @@ import (
13		"time"	13		"time"
14			14
15		"github.com/bnema/vev/internal/adapters/ipc"	15		"github.com/bnema/vev/internal/adapters/ipc"
			16	+	"github.com/bnema/vev/internal/adapters/lifecycle"
16		"github.com/bnema/vev/internal/platform"	17		"github.com/bnema/vev/internal/platform"
17		"github.com/bnema/vev/internal/ports"	18		"github.com/bnema/vev/internal/ports"
18		"github.com/bnema/vev/pkg/safedir"	19		"github.com/bnema/vev/pkg/safedir"

@@ -39,6 +40,18 @@ type dialFunc func(ctx context.Context, dir string) (ports.Transport, error)
39		// re-exec a real binary.	40		// re-exec a real binary.
40		type spawnFunc func() error	41		type spawnFunc func() error
41			42
			43	+	type lifecycleProbe interface {
			44	+	TryAcquire(string) (*lifecycle.Owner, error)
			45	+	}
			46	+
			47	+	type osLifecycleProbe struct{}
			48	+
			49	+	func (osLifecycleProbe) TryAcquire(runtimeDir string) (*lifecycle.Owner, error) {
			50	+	return lifecycle.TryAcquire(runtimeDir)
			51	+	}
			52	+
			53	+	var daemonLifecycleProbe lifecycleProbe = osLifecycleProbe{}
			54	+
42		// backoffConfig parameterises retry-dial timing so tests can shrink it.	55		// backoffConfig parameterises retry-dial timing so tests can shrink it.
43		type backoffConfig struct {	56		type backoffConfig struct {
44		initial time.Duration	57		initial time.Duration

@@ -54,10 +67,97 @@ var defaultBackoff = backoffConfig{
54		total: 5 * time.Second,	67		total: 5 * time.Second,
55		}	68		}
56			69
			70	+	func waitForLifecycleAvailability(ctx context.Context, runtimeDir string, cfg backoffConfig) (*lifecycle.Owner, error) {
			71	+	deadline := time.Now().Add(cfg.total)
			72	+	backoff := cfg.initial
			73	+	for {
			74	+	if err := ctx.Err(); err != nil {
			75	+	return nil, err
			76	+	}
			77	+	owner, err := daemonLifecycleProbe.TryAcquire(runtimeDir)
			78	+	if err == nil {
			79	+	return owner, nil
			80	+	}
			81	+	if !errors.Is(err, lifecycle.ErrBusy) {
			82	+	return nil, err
			83	+	}
			84	+	if !time.Now().Before(deadline) {
			85	+	return nil, ErrDaemonUnreachable
			86	+	}
			87	+	if err := waitBackoff(ctx, backoff); err != nil {
			88	+	return nil, err
			89	+	}
			90	+	if backoff *= 2; backoff > cfg.max {
			91	+	backoff = cfg.max
			92	+	}
			93	+	}
			94	+	}
			95	+
			96	+	func waitForDaemonOrLifecycle(ctx context.Context, dir string, dial dialFunc, cfg backoffConfig) (ports.Transport, *lifecycle.Owner, error) {
			97	+	deadline := time.Now().Add(cfg.total)
			98	+	backoff := cfg.initial
			99	+	for {
			100	+	if err := ctx.Err(); err != nil {
			101	+	return nil, nil, err
			102	+	}
			103	+	if transport, err := dial(ctx, dir); err == nil {
			104	+	return transport, nil, nil
			105	+	}
			106	+	owner, err := daemonLifecycleProbe.TryAcquire(dir)
			107	+	if err == nil {
			108	+	return nil, owner, nil
			109	+	}
			110	+	if !errors.Is(err, lifecycle.ErrBusy) {
			111	+	return nil, nil, err
			112	+	}
			113	+	if !time.Now().Before(deadline) {
			114	+	return nil, nil, ErrDaemonUnreachable
			115	+	}
			116	+	if err := waitBackoff(ctx, backoff); err != nil {
			117	+	return nil, nil, err
			118	+	}
			119	+	if backoff *= 2; backoff > cfg.max {
			120	+	backoff = cfg.max
			121	+	}
			122	+	}
			123	+	}
			124	+
			125	+	func waitBackoff(ctx context.Context, duration time.Duration) error {
			126	+	timer := time.NewTimer(duration)
			127	+	defer timer.Stop()
			128	+	select {
			129	+	case <-ctx.Done():
			130	+	return ctx.Err()
			131	+	case <-timer.C:
			132	+	return nil
			133	+	}
			134	+	}
			135	+
			136	+	// ensureDaemonWithLifecycle never elects a spawner while another lifecycle
			137	+	// owner may still be initializing or tearing down durable state.
			138	+	func ensureDaemonWithLifecycle(ctx context.Context, dir string, dial dialFunc, spawn spawnFunc, cfg backoffConfig) (ports.Transport, error) {
			139	+	if err := ctx.Err(); err != nil {
			140	+	return nil, err
			141	+	}
			142	+	if transport, err := dial(ctx, dir); err == nil {
			143	+	return transport, nil
			144	+	}
			145	+	transport, owner, err := waitForDaemonOrLifecycle(ctx, dir, dial, cfg)
			146	+	if err != nil {
			147	+	return nil, err
			148	+	}
			149	+	if transport != nil {
			150	+	return transport, nil
			151	+	}
			152	+	if err := owner.Release(); err != nil {
			153	+	return nil, fmt.Errorf("vev: release lifecycle spawn probe: %w", err)
			154	+	}
			155	+	return ensureDaemon(ctx, dir, dial, spawn, cfg)
			156	+	}
			157	+
57		// ensureDaemon returns a transport to a running daemon, spawning one if	158		// ensureDaemon returns a transport to a running daemon, spawning one if
58	-	// necessary. It first tries a plain dial; on failure it elects a single	159	+	// necessary after ensureDaemonWithLifecycle has established lifecycle
59	-	// spawner via an mkdir lock (losers simply wait), then retry-dials with	160	+	// availability. Tests also exercise this lower-level spawn election directly.
60	-	// backoff until the socket is live or the budget expires.
61		func ensureDaemon(ctx context.Context, dir string, dial dialFunc, spawn spawnFunc, cfg backoffConfig) (ports.Transport, error) {	161		func ensureDaemon(ctx context.Context, dir string, dial dialFunc, spawn spawnFunc, cfg backoffConfig) (ports.Transport, error) {
62		if err := ctx.Err(); err != nil {	162		if err := ctx.Err(); err != nil {
63		return nil, err	163		return nil, err