detached bb3ed31

Merge pull request #198 from bnema/fix/protocol-version-session-reset

fix: reset incompatible sessions and preserve local handoffs
commit
bb3ed31460d5aa6cfb4c4125f6855646e024c784
author
bnema · · 2026-08-11 12:59 UTC
committer
GitHub · · 2026-08-11 12:59 UTC
merge of
2bffe86 + ecc1937 · diff against the first parent
signature
GPG UnverifiedB5690EEEBB952194 The key is unknown
14 files · +373 -57
Unified
UnifiedSplit
M README.md +3 -0
@@ -41,6 +41,9 @@ vev kill <name>|--all kill sessions
4141
4242 The daemon starts on first use and exits after the last session. Numbered sessions survive detach; named sessions also survive daemon restarts. See [durable-session recovery](docs/durable-session-recovery.md).
4343
44+> [!NOTE]
45+> A release that bumps vev's protocol version resets named sessions saved by an older protocol. Session names are retained, but layouts, tabs, terminal history, recovery transcripts, and process-recovery state are discarded. Each local or remote daemon applies this reset to its own sessions when it first starts with the new protocol.
46+
4447 ## Keys and palette
4548
4649 No prefix key; bindings use Alt.
@@ -41,6 +41,9 @@ vev kill <name>|--all kill sessions
41 41
42 The daemon starts on first use and exits after the last session. Numbered sessions survive detach; named sessions also survive daemon restarts. See [durable-session recovery](docs/durable-session-recovery.md).42 The daemon starts on first use and exits after the last session. Numbered sessions survive detach; named sessions also survive daemon restarts. See [durable-session recovery](docs/durable-session-recovery.md).
43 43
44+> [!NOTE]
45+> A release that bumps vev's protocol version resets named sessions saved by an older protocol. Session names are retained, but layouts, tabs, terminal history, recovery transcripts, and process-recovery state are discarded. Each local or remote daemon applies this reset to its own sessions when it first starts with the new protocol.
46+
44 ## Keys and palette47 ## Keys and palette
45 48
46 No prefix key; bindings use Alt.49 No prefix key; bindings use Alt.
M docs/durable-session-recovery.md +7 -1
@@ -12,9 +12,15 @@ vev opens named-session state only while holding `$XDG_RUNTIME_DIR/vev/lifecycle
1212
1313 Discard creates a new incarnation and retains the old record and snapshots under `snapshots/quarantine/` until an explicit later purge.
1414
15+## Protocol upgrades
16+
17+Each catalogue record stores the protocol version that wrote it. At startup, a record from any other protocol—including a newer protocol after a downgrade—is atomically replaced with a fresh incarnation before the daemon publishes its socket. Only the session name is retained; the working directory, layout, tabs, terminal history, recovery transcript, process-recovery state, and checkpoints are discarded. Records created before protocol tracking are treated the same way.
18+
19+This reset is per named session and crash-safe. Once the fresh catalogue authority is committed, leftover snapshot objects are no longer authoritative and startup garbage collection can remove them. Catalogue corruption is not treated as a protocol mismatch and remains fail-closed.
20+
1521 ## Incompatible checkpoints
1622
17-After verifying a VEVM manifest's digest, vev treats any VEVM version mismatch as an incompatible healthy checkpoint. It atomically replaces only that named session's exact healthy checkpoint with a fresh incarnation. The replacement retains the session name and working directory, but has no checkpoint, tabs, terminal history, or recovery transcript.
23+After verifying a VEVM manifest's digest, vev treats any VEVM version mismatch as an incompatible healthy checkpoint. It atomically replaces only that named session's exact healthy checkpoint with a fresh incarnation. Unlike a protocol reset, this replacement retains the session name and working directory, but has no checkpoint, tabs, terminal history, or recovery transcript.
1824
1925 Digest mismatches, corruption, validation failures, I/O errors, and ambiguous failures are never reset or purged. The session remains degraded for explicit recovery.
2026
@@ -12,9 +12,15 @@ vev opens named-session state only while holding `$XDG_RUNTIME_DIR/vev/lifecycle
12 12
13 Discard creates a new incarnation and retains the old record and snapshots under `snapshots/quarantine/` until an explicit later purge.13 Discard creates a new incarnation and retains the old record and snapshots under `snapshots/quarantine/` until an explicit later purge.
14 14
15+## Protocol upgrades
16+
17+Each catalogue record stores the protocol version that wrote it. At startup, a record from any other protocol—including a newer protocol after a downgrade—is atomically replaced with a fresh incarnation before the daemon publishes its socket. Only the session name is retained; the working directory, layout, tabs, terminal history, recovery transcript, process-recovery state, and checkpoints are discarded. Records created before protocol tracking are treated the same way.
18+
19+This reset is per named session and crash-safe. Once the fresh catalogue authority is committed, leftover snapshot objects are no longer authoritative and startup garbage collection can remove them. Catalogue corruption is not treated as a protocol mismatch and remains fail-closed.
20+
15 ## Incompatible checkpoints21 ## Incompatible checkpoints
16 22
17-After verifying a VEVM manifest's digest, vev treats any VEVM version mismatch as an incompatible healthy checkpoint. It atomically replaces only that named session's exact healthy checkpoint with a fresh incarnation. The replacement retains the session name and working directory, but has no checkpoint, tabs, terminal history, or recovery transcript.23+After verifying a VEVM manifest's digest, vev treats any VEVM version mismatch as an incompatible healthy checkpoint. It atomically replaces only that named session's exact healthy checkpoint with a fresh incarnation. Unlike a protocol reset, this replacement retains the session name and working directory, but has no checkpoint, tabs, terminal history, or recovery transcript.
18 24
19 Digest mismatches, corruption, validation failures, I/O errors, and ambiguous failures are never reset or purged. The session remains degraded for explicit recovery.25 Digest mismatches, corruption, validation failures, I/O errors, and ambiguous failures are never reset or purged. The session remains degraded for explicit recovery.
20 26
M docs/remote-resilience.md +10 -8
@@ -37,17 +37,19 @@ shared between clients.
3737 A picker-selected remote target is a direct handoff with one input pump. Exact
3838 lifecycle/name identities are carried separately from display labels, and a
3939 successful daemon-local switch publishes a new committed identity before the
40-client republishes its snapshot. These navigation actions and the strict
41-handshake layout use protocol v27. Attach handshakes may carry an exact target,
42-and successful Welcomes can return the daemon's committed identity.
40+client republishes its snapshot. These navigation actions use the strict,
41+exact-match protocol version. Attach handshakes may carry an exact target, and
42+successful Welcomes can return the daemon's committed identity.
4343
4444 ## Durable record compatibility
4545
46-The catalogue record format is currently version 4 because records now retain
47-stable tab identity metadata. Readers accept version 3 for one-way upgrades, but
48-older binaries reject records written at version 4. Back up the vev state
49-before a rollback; downgrading after an upgrade can leave newer catalogue
50-records unreadable and may require recreating those sessions.
46+Catalogue record format version 5 stores the protocol version alongside stable
47+tab identity metadata. At daemon startup, version 3 and 4 records and version 5
48+records written by another protocol are decoded only to retain their session
49+names; all other session state is reset before socket publication. Compatible
50+version 5 records restore normally. Older binaries reject version 5 records, so
51+a downgrade also resets sessions when the older binary can read the catalogue,
52+or fails closed when it cannot. Back up the vev state before a rollback.
5153
5254 Remote catalogues are bounded and versioned. Complete catalogues contain ordered
5355 tab IDs, names, state, active tab, attachment state, and MRU sequence. Stopped
@@ -37,17 +37,19 @@ shared between clients.
37 A picker-selected remote target is a direct handoff with one input pump. Exact37 A picker-selected remote target is a direct handoff with one input pump. Exact
38 lifecycle/name identities are carried separately from display labels, and a38 lifecycle/name identities are carried separately from display labels, and a
39 successful daemon-local switch publishes a new committed identity before the39 successful daemon-local switch publishes a new committed identity before the
40-client republishes its snapshot. These navigation actions and the strict40+client republishes its snapshot. These navigation actions use the strict,
41-handshake layout use protocol v27. Attach handshakes may carry an exact target,41+exact-match protocol version. Attach handshakes may carry an exact target, and
42-and successful Welcomes can return the daemon's committed identity.42+successful Welcomes can return the daemon's committed identity.
43 43
44 ## Durable record compatibility44 ## Durable record compatibility
45 45
46-The catalogue record format is currently version 4 because records now retain46+Catalogue record format version 5 stores the protocol version alongside stable
47-stable tab identity metadata. Readers accept version 3 for one-way upgrades, but47+tab identity metadata. At daemon startup, version 3 and 4 records and version 5
48-older binaries reject records written at version 4. Back up the vev state48+records written by another protocol are decoded only to retain their session
49-before a rollback; downgrading after an upgrade can leave newer catalogue49+names; all other session state is reset before socket publication. Compatible
50-records unreadable and may require recreating those sessions.50+version 5 records restore normally. Older binaries reject version 5 records, so
51+a downgrade also resets sessions when the older binary can read the catalogue,
52+or fails closed when it cannot. Back up the vev state before a rollback.
51 53
52 Remote catalogues are bounded and versioned. Complete catalogues contain ordered54 Remote catalogues are bounded and versioned. Complete catalogues contain ordered
53 tab IDs, names, state, active tab, attachment state, and MRU sequence. Stopped55 tab IDs, names, state, active tab, attachment state, and MRU sequence. Stopped
M internal/app/run.go +35 -1
@@ -568,6 +568,33 @@ func runDaemon() (retErr error) {
568568 }, lifecycleStartupDependencies(log))
569569 }
570570
571+func resetProtocolIncompatibleRecords(ctx context.Context, log *slog.Logger, coordinator *recovery.Coordinator, records, incompatible []domain.CatalogueRecord) ([]domain.CatalogueRecord, error) {
572+ if len(incompatible) == 0 {
573+ return records, nil
574+ }
575+ byName := make(map[string]int, len(records))
576+ for i := range records {
577+ byName[records[i].Name] = i
578+ }
579+ for _, old := range incompatible {
580+ fresh, committed, err := coordinator.ResetProtocolIncompatible(ctx, old.Name, old.IncarnationID)
581+ if !committed {
582+ if err != nil {
583+ return nil, fmt.Errorf("reset protocol-incompatible session %q: %w", old.Name, err)
584+ }
585+ return nil, fmt.Errorf("reset protocol-incompatible session %q: catalogue authority changed", old.Name)
586+ }
587+ if index, ok := byName[old.Name]; ok {
588+ records[index] = fresh
589+ }
590+ log.Info("protocol_incompatible_session_reset", "session", old.Name, "stored_incarnation", old.IncarnationID.String())
591+ if err != nil {
592+ log.Warn("protocol_incompatible_snapshot_cleanup_failed", "session", old.Name, "err", err)
593+ }
594+ }
595+ return records, nil
596+}
597+
571598 func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, recoveryMode string) {
572599 log.Info("catalogue_validated", "records", len(records), "recovery", recoveryMode)
573600 }
@@ -666,8 +693,15 @@ func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr err
666693 if opened.NewInstall {
667694 recoveryMode = "new-install"
668695 }
669- logCatalogueRecovery(log, opened.Records, recoveryMode)
670696 coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, rand.Reader)
697+ opened.Records, err = resetProtocolIncompatibleRecords(ctx, log, coordinator, opened.Records, opened.IncompatibleRecords)
698+ if err != nil {
699+ return errors.Join(fmt.Errorf("vev: reset protocol-incompatible durable sessions: %w", err), opened.Catalogue.Close())
700+ }
701+ if len(opened.IncompatibleRecords) != 0 {
702+ recoveryMode = "protocol-reset"
703+ }
704+ logCatalogueRecovery(log, opened.Records, recoveryMode)
671705 daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))
672706 daemonOpts = append(daemonOpts, daemon.WithDurableMaintenance(opened.Catalogue, snapshotRepository))
673707 log.Info("session persistence enabled", "path", storePath)
@@ -568,6 +568,33 @@ func runDaemon() (retErr error) {
568 }, lifecycleStartupDependencies(log))568 }, lifecycleStartupDependencies(log))
569 }569 }
570 570
571+func resetProtocolIncompatibleRecords(ctx context.Context, log *slog.Logger, coordinator *recovery.Coordinator, records, incompatible []domain.CatalogueRecord) ([]domain.CatalogueRecord, error) {
572+ if len(incompatible) == 0 {
573+ return records, nil
574+ }
575+ byName := make(map[string]int, len(records))
576+ for i := range records {
577+ byName[records[i].Name] = i
578+ }
579+ for _, old := range incompatible {
580+ fresh, committed, err := coordinator.ResetProtocolIncompatible(ctx, old.Name, old.IncarnationID)
581+ if !committed {
582+ if err != nil {
583+ return nil, fmt.Errorf("reset protocol-incompatible session %q: %w", old.Name, err)
584+ }
585+ return nil, fmt.Errorf("reset protocol-incompatible session %q: catalogue authority changed", old.Name)
586+ }
587+ if index, ok := byName[old.Name]; ok {
588+ records[index] = fresh
589+ }
590+ log.Info("protocol_incompatible_session_reset", "session", old.Name, "stored_incarnation", old.IncarnationID.String())
591+ if err != nil {
592+ log.Warn("protocol_incompatible_snapshot_cleanup_failed", "session", old.Name, "err", err)
593+ }
594+ }
595+ return records, nil
596+}
597+
571 func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, recoveryMode string) {598 func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, recoveryMode string) {
572 log.Info("catalogue_validated", "records", len(records), "recovery", recoveryMode)599 log.Info("catalogue_validated", "records", len(records), "recovery", recoveryMode)
573 }600 }
@@ -666,8 +693,15 @@ func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr err
666 if opened.NewInstall {693 if opened.NewInstall {
667 recoveryMode = "new-install"694 recoveryMode = "new-install"
668 }695 }
669- logCatalogueRecovery(log, opened.Records, recoveryMode)
670 coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, rand.Reader)696 coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, rand.Reader)
697+ opened.Records, err = resetProtocolIncompatibleRecords(ctx, log, coordinator, opened.Records, opened.IncompatibleRecords)
698+ if err != nil {
699+ return errors.Join(fmt.Errorf("vev: reset protocol-incompatible durable sessions: %w", err), opened.Catalogue.Close())
700+ }
701+ if len(opened.IncompatibleRecords) != 0 {
702+ recoveryMode = "protocol-reset"
703+ }
704+ logCatalogueRecovery(log, opened.Records, recoveryMode)
671 daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))705 daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))
672 daemonOpts = append(daemonOpts, daemon.WithDurableMaintenance(opened.Catalogue, snapshotRepository))706 daemonOpts = append(daemonOpts, daemon.WithDurableMaintenance(opened.Catalogue, snapshotRepository))
673 log.Info("session persistence enabled", "path", storePath)707 log.Info("session persistence enabled", "path", storePath)
M internal/persist/catalogue_test.go +48 -1
@@ -1,6 +1,7 @@
11 package persist
22
33 import (
4+ "encoding/binary"
45 "os"
56 "path/filepath"
67 "testing"
@@ -8,6 +9,7 @@ import (
89 "github.com/stretchr/testify/require"
910
1011 "github.com/bnema/vev/internal/domain"
12+ "github.com/bnema/vev/internal/ports"
1113 "github.com/bnema/vev/pkg/kv"
1214 "github.com/bnema/vev/pkg/safedir"
1315 )
@@ -48,6 +50,51 @@ func testCatalogueRecordRoundTrip(t *testing.T) {
4850 }
4951 }
5052
53+func TestOpenOrCreateReportsRecordsFromOlderProtocol(t *testing.T) {
54+ record := domain.CatalogueRecord{
55+ Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/workspace",
56+ CreatedAt: 11, UpdatedAt: 22, LastUsedSeq: 33,
57+ TabNames: []string{"shell", "logs"},
58+ Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}},
59+ }
60+ olderProtocol, err := encodeRecordValueForProtocol(record, ports.ProtocolVersion-1)
61+ require.NoError(t, err)
62+ protocolLess, err := encodeRecordValue(record)
63+ require.NoError(t, err)
64+ binary.BigEndian.PutUint16(protocolLess[len(catalogueMagic):], protocolLessCatalogueRecordVersion)
65+ protocolLess = append(protocolLess[:len(catalogueMagic)+2], protocolLess[len(catalogueMagic)+4:]...)
66+
67+ currentProtocol, err := encodeRecordValue(record)
68+ require.NoError(t, err)
69+ for _, tt := range []struct {
70+ name string
71+ encoded []byte
72+ incompatible bool
73+ }{
74+ {name: "current protocol", encoded: currentProtocol},
75+ {name: "recorded older protocol", encoded: olderProtocol, incompatible: true},
76+ {name: "protocol-less v4 record", encoded: protocolLess, incompatible: true},
77+ } {
78+ t.Run(tt.name, func(t *testing.T) {
79+ dir := privateDir(t)
80+ store, err := kv.Open(StorePath(dir))
81+ require.NoError(t, err)
82+ require.NoError(t, store.Set([]byte(record.Name), tt.encoded))
83+ require.NoError(t, store.Close())
84+
85+ opened, err := OpenOrCreate(dir)
86+ require.NoError(t, err)
87+ defer func() { require.NoError(t, opened.Catalogue.Close()) }()
88+ require.Equal(t, []domain.CatalogueRecord{record}, opened.Records)
89+ if tt.incompatible {
90+ require.Equal(t, []domain.CatalogueRecord{record}, opened.IncompatibleRecords)
91+ } else {
92+ require.Empty(t, opened.IncompatibleRecords)
93+ }
94+ })
95+ }
96+}
97+
5198 func TestDecodeRecordValueRejectsMalformedCheckpointMarker(t *testing.T) {
5299 encoded, err := encodeRecordValue(validRecord("work", 1))
53100 require.NoError(t, err)
@@ -60,7 +107,7 @@ func TestDecodeRecordValueRejectsMalformedCheckpointMarker(t *testing.T) {
60107 func checkpointMarkerOffset(t *testing.T, encoded []byte) int {
61108 t.Helper()
62109 r := valueReader{data: encoded}
63- _, ok := r.take(len(catalogueMagic) + 2 + len(domain.IncarnationID{}))
110+ _, ok := r.take(len(catalogueMagic) + 2 + 2 + len(domain.IncarnationID{}))
64111 require.True(t, ok)
65112 _, ok = r.str()
66113 require.True(t, ok)
@@ -1,6 +1,7 @@
1 package persist1 package persist
2 2
3 import (3 import (
4+ "encoding/binary"
4 "os"5 "os"
5 "path/filepath"6 "path/filepath"
6 "testing"7 "testing"
@@ -8,6 +9,7 @@ import (
8 "github.com/stretchr/testify/require"9 "github.com/stretchr/testify/require"
9 10
10 "github.com/bnema/vev/internal/domain"11 "github.com/bnema/vev/internal/domain"
12+ "github.com/bnema/vev/internal/ports"
11 "github.com/bnema/vev/pkg/kv"13 "github.com/bnema/vev/pkg/kv"
12 "github.com/bnema/vev/pkg/safedir"14 "github.com/bnema/vev/pkg/safedir"
13 )15 )
@@ -48,6 +50,51 @@ func testCatalogueRecordRoundTrip(t *testing.T) {
48 }50 }
49 }51 }
50 52
53+func TestOpenOrCreateReportsRecordsFromOlderProtocol(t *testing.T) {
54+ record := domain.CatalogueRecord{
55+ Name: "work", IncarnationID: domain.IncarnationID{1}, Cwd: "/workspace",
56+ CreatedAt: 11, UpdatedAt: 22, LastUsedSeq: 33,
57+ TabNames: []string{"shell", "logs"},
58+ Committed: &domain.CheckpointRef{Generation: 3, ManifestDigest: [32]byte{3}},
59+ }
60+ olderProtocol, err := encodeRecordValueForProtocol(record, ports.ProtocolVersion-1)
61+ require.NoError(t, err)
62+ protocolLess, err := encodeRecordValue(record)
63+ require.NoError(t, err)
64+ binary.BigEndian.PutUint16(protocolLess[len(catalogueMagic):], protocolLessCatalogueRecordVersion)
65+ protocolLess = append(protocolLess[:len(catalogueMagic)+2], protocolLess[len(catalogueMagic)+4:]...)
66+
67+ currentProtocol, err := encodeRecordValue(record)
68+ require.NoError(t, err)
69+ for _, tt := range []struct {
70+ name string
71+ encoded []byte
72+ incompatible bool
73+ }{
74+ {name: "current protocol", encoded: currentProtocol},
75+ {name: "recorded older protocol", encoded: olderProtocol, incompatible: true},
76+ {name: "protocol-less v4 record", encoded: protocolLess, incompatible: true},
77+ } {
78+ t.Run(tt.name, func(t *testing.T) {
79+ dir := privateDir(t)
80+ store, err := kv.Open(StorePath(dir))
81+ require.NoError(t, err)
82+ require.NoError(t, store.Set([]byte(record.Name), tt.encoded))
83+ require.NoError(t, store.Close())
84+
85+ opened, err := OpenOrCreate(dir)
86+ require.NoError(t, err)
87+ defer func() { require.NoError(t, opened.Catalogue.Close()) }()
88+ require.Equal(t, []domain.CatalogueRecord{record}, opened.Records)
89+ if tt.incompatible {
90+ require.Equal(t, []domain.CatalogueRecord{record}, opened.IncompatibleRecords)
91+ } else {
92+ require.Empty(t, opened.IncompatibleRecords)
93+ }
94+ })
95+ }
96+}
97+
51 func TestDecodeRecordValueRejectsMalformedCheckpointMarker(t *testing.T) {98 func TestDecodeRecordValueRejectsMalformedCheckpointMarker(t *testing.T) {
52 encoded, err := encodeRecordValue(validRecord("work", 1))99 encoded, err := encodeRecordValue(validRecord("work", 1))
53 require.NoError(t, err)100 require.NoError(t, err)
@@ -60,7 +107,7 @@ func TestDecodeRecordValueRejectsMalformedCheckpointMarker(t *testing.T) {
60 func checkpointMarkerOffset(t *testing.T, encoded []byte) int {107 func checkpointMarkerOffset(t *testing.T, encoded []byte) int {
61 t.Helper()108 t.Helper()
62 r := valueReader{data: encoded}109 r := valueReader{data: encoded}
63- _, ok := r.take(len(catalogueMagic) + 2 + len(domain.IncarnationID{}))110+ _, ok := r.take(len(catalogueMagic) + 2 + 2 + len(domain.IncarnationID{}))
64 require.True(t, ok)111 require.True(t, ok)
65 _, ok = r.str()112 _, ok = r.str()
66 require.True(t, ok)113 require.True(t, ok)
M internal/persist/persist.go +36 -4
@@ -28,9 +28,10 @@ var ErrCatalogueUnreadable = errors.New("persist: catalogue unreadable")
2828
2929 // OpenResult is the outcome of opening durable session state at startup.
3030 type OpenResult struct {
31- Catalogue *Persister
32- Records []domain.CatalogueRecord
33- NewInstall bool
31+ Catalogue *Persister
32+ Records []domain.CatalogueRecord
33+ IncompatibleRecords []domain.CatalogueRecord
34+ NewInstall bool
3435 }
3536
3637 func StorePath(dir string) string { return filepath.Join(dir, filename) }
@@ -87,7 +88,11 @@ func OpenOrCreate(dir string) (OpenResult, error) {
8788 if err != nil {
8889 return OpenResult{}, fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err)
8990 }
90- return OpenResult{Catalogue: catalogue, Records: records, NewInstall: !existed}, nil
91+ incompatible, err := catalogue.loadIncompatibleRecords()
92+ if err != nil {
93+ return OpenResult{}, errors.Join(fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err), catalogue.Close())
94+ }
95+ return OpenResult{Catalogue: catalogue, Records: records, IncompatibleRecords: incompatible, NewInstall: !existed}, nil
9196 }
9297
9398 func New(store ports.Store) *Persister { return &Persister{store: store} }
@@ -431,6 +436,33 @@ func (p *Persister) Delete(name string) error {
431436 return p.Apply(map[string]*domain.CatalogueRecord{name: nil})
432437 }
433438 func (p *Persister) LoadAll() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() }
439+
440+func (p *Persister) loadIncompatibleRecords() ([]domain.CatalogueRecord, error) {
441+ if p == nil || p.store == nil {
442+ return nil, errPersistenceUnavailable
443+ }
444+ p.mu.Lock()
445+ defer p.mu.Unlock()
446+ if err := p.terminalLocked(); err != nil {
447+ return nil, err
448+ }
449+ var records []domain.CatalogueRecord
450+ var decodeErr error
451+ p.store.Range(func(key, value []byte) bool {
452+ stored, err := decodeStoredRecordValue(string(key), value)
453+ if err != nil {
454+ decodeErr = err
455+ return false
456+ }
457+ if stored.protocolVersion != ports.ProtocolVersion {
458+ records = append(records, stored.record)
459+ }
460+ return true
461+ })
462+ sort.Slice(records, func(i, j int) bool { return records[i].Name < records[j].Name })
463+ return records, decodeErr
464+}
465+
434466 func (p *Persister) LoadCatalogue() ([]domain.CatalogueRecord, error) {
435467 if p == nil {
436468 return []domain.CatalogueRecord{}, nil
@@ -28,9 +28,10 @@ var ErrCatalogueUnreadable = errors.New("persist: catalogue unreadable")
28 28
29 // OpenResult is the outcome of opening durable session state at startup.29 // OpenResult is the outcome of opening durable session state at startup.
30 type OpenResult struct {30 type OpenResult struct {
31- Catalogue *Persister31+ Catalogue *Persister
32- Records []domain.CatalogueRecord32+ Records []domain.CatalogueRecord
33- NewInstall bool33+ IncompatibleRecords []domain.CatalogueRecord
34+ NewInstall bool
34 }35 }
35 36
36 func StorePath(dir string) string { return filepath.Join(dir, filename) }37 func StorePath(dir string) string { return filepath.Join(dir, filename) }
@@ -87,7 +88,11 @@ func OpenOrCreate(dir string) (OpenResult, error) {
87 if err != nil {88 if err != nil {
88 return OpenResult{}, fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err)89 return OpenResult{}, fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err)
89 }90 }
90- return OpenResult{Catalogue: catalogue, Records: records, NewInstall: !existed}, nil91+ incompatible, err := catalogue.loadIncompatibleRecords()
92+ if err != nil {
93+ return OpenResult{}, errors.Join(fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err), catalogue.Close())
94+ }
95+ return OpenResult{Catalogue: catalogue, Records: records, IncompatibleRecords: incompatible, NewInstall: !existed}, nil
91 }96 }
92 97
93 func New(store ports.Store) *Persister { return &Persister{store: store} }98 func New(store ports.Store) *Persister { return &Persister{store: store} }
@@ -431,6 +436,33 @@ func (p *Persister) Delete(name string) error {
431 return p.Apply(map[string]*domain.CatalogueRecord{name: nil})436 return p.Apply(map[string]*domain.CatalogueRecord{name: nil})
432 }437 }
433 func (p *Persister) LoadAll() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() }438 func (p *Persister) LoadAll() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() }
439+
440+func (p *Persister) loadIncompatibleRecords() ([]domain.CatalogueRecord, error) {
441+ if p == nil || p.store == nil {
442+ return nil, errPersistenceUnavailable
443+ }
444+ p.mu.Lock()
445+ defer p.mu.Unlock()
446+ if err := p.terminalLocked(); err != nil {
447+ return nil, err
448+ }
449+ var records []domain.CatalogueRecord
450+ var decodeErr error
451+ p.store.Range(func(key, value []byte) bool {
452+ stored, err := decodeStoredRecordValue(string(key), value)
453+ if err != nil {
454+ decodeErr = err
455+ return false
456+ }
457+ if stored.protocolVersion != ports.ProtocolVersion {
458+ records = append(records, stored.record)
459+ }
460+ return true
461+ })
462+ sort.Slice(records, func(i, j int) bool { return records[i].Name < records[j].Name })
463+ return records, decodeErr
464+}
465+
434 func (p *Persister) LoadCatalogue() ([]domain.CatalogueRecord, error) {466 func (p *Persister) LoadCatalogue() ([]domain.CatalogueRecord, error) {
435 if p == nil {467 if p == nil {
436 return []domain.CatalogueRecord{}, nil468 return []domain.CatalogueRecord{}, nil
M internal/persist/record.go +43 -19
@@ -7,11 +7,13 @@ import (
77 "math"
88
99 "github.com/bnema/vev/internal/domain"
10+ "github.com/bnema/vev/internal/ports"
1011 )
1112
1213 const (
13- catalogueRecordVersion uint16 = 4
14- legacyCatalogueRecordVersion uint16 = 3
14+ catalogueRecordVersion uint16 = 5
15+ protocolLessCatalogueRecordVersion uint16 = 4
16+ legacyCatalogueRecordVersion uint16 = 3
1517 )
1618
1719 var errMalformedRecord = errors.New("persist: malformed catalogue record")
@@ -22,12 +24,17 @@ var catalogueMagic = [4]byte{'V', 'E', 'V', 'C'}
2224 type Record = domain.CatalogueRecord
2325
2426 func encodeRecordValue(record domain.CatalogueRecord) ([]byte, error) {
27+ return encodeRecordValueForProtocol(record, ports.ProtocolVersion)
28+}
29+
30+func encodeRecordValueForProtocol(record domain.CatalogueRecord, protocolVersion uint16) ([]byte, error) {
2531 if err := record.Validate(); err != nil {
2632 return nil, fmt.Errorf("persist: invalid catalogue record: %w", err)
2733 }
2834 buf := make([]byte, 0, 128)
2935 buf = append(buf, catalogueMagic[:]...)
3036 buf = binary.BigEndian.AppendUint16(buf, catalogueRecordVersion)
37+ buf = binary.BigEndian.AppendUint16(buf, protocolVersion)
3138 buf = append(buf, record.IncarnationID[:]...)
3239 var err error
3340 buf, err = appendCheckedString(buf, record.Cwd)
@@ -66,45 +73,62 @@ func encodeRecordValue(record domain.CatalogueRecord) ([]byte, error) {
6673 return buf, err
6774 }
6875
76+type storedRecord struct {
77+ record domain.CatalogueRecord
78+ protocolVersion uint16
79+}
80+
6981 func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error) {
82+ stored, err := decodeStoredRecordValue(name, value)
83+ return stored.record, err
84+}
85+
86+func decodeStoredRecordValue(name string, value []byte) (storedRecord, error) {
7087 if err := domain.ValidateSessionName(name); err != nil {
71- return domain.CatalogueRecord{}, err
88+ return storedRecord{}, err
7289 }
7390 reader := valueReader{data: value}
7491 magic, ok := reader.take(4)
7592 if !ok || string(magic) != string(catalogueMagic[:]) {
76- return domain.CatalogueRecord{}, errMalformedRecord
93+ return storedRecord{}, errMalformedRecord
7794 }
7895 version, ok := reader.u16()
79- if !ok || (version != catalogueRecordVersion && version != legacyCatalogueRecordVersion) {
80- return domain.CatalogueRecord{}, errMalformedRecord
96+ if !ok || (version != catalogueRecordVersion && version != protocolLessCatalogueRecordVersion && version != legacyCatalogueRecordVersion) {
97+ return storedRecord{}, errMalformedRecord
98+ }
99+ var protocolVersion uint16
100+ if version == catalogueRecordVersion {
101+ protocolVersion, ok = reader.u16()
102+ if !ok {
103+ return storedRecord{}, errMalformedRecord
104+ }
81105 }
82106 id, ok := reader.take(16)
83107 if !ok {
84- return domain.CatalogueRecord{}, errMalformedRecord
108+ return storedRecord{}, errMalformedRecord
85109 }
86110 record := domain.CatalogueRecord{Name: name}
87111 copy(record.IncarnationID[:], id)
88112 var valid bool
89113 if record.Cwd, valid = reader.str(); !valid {
90- return domain.CatalogueRecord{}, errMalformedRecord
114+ return storedRecord{}, errMalformedRecord
91115 }
92116 created, ok := reader.u64()
93117 if !ok {
94- return domain.CatalogueRecord{}, errMalformedRecord
118+ return storedRecord{}, errMalformedRecord
95119 }
96120 record.CreatedAt = int64(created)
97121 updated, ok := reader.u64()
98122 if !ok {
99- return domain.CatalogueRecord{}, errMalformedRecord
123+ return storedRecord{}, errMalformedRecord
100124 }
101125 record.UpdatedAt = int64(updated)
102126 if record.LastUsedSeq, ok = reader.u64(); !ok {
103- return domain.CatalogueRecord{}, errMalformedRecord
127+ return storedRecord{}, errMalformedRecord
104128 }
105129 count, ok := reader.u32()
106130 if !ok || uint64(count) > uint64(reader.remaining())/4 {
107- return domain.CatalogueRecord{}, errMalformedRecord
131+ return storedRecord{}, errMalformedRecord
108132 }
109133 if count > 0 {
110134 record.TabNames = make([]string, 0, int(count))
@@ -112,7 +136,7 @@ func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error
112136 for range count {
113137 tab, ok := reader.str()
114138 if !ok {
115- return domain.CatalogueRecord{}, errMalformedRecord
139+ return storedRecord{}, errMalformedRecord
116140 }
117141 record.TabNames = append(record.TabNames, tab)
118142 }
@@ -126,7 +150,7 @@ func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error
126150 } else {
127151 tabRecordCount, ok := reader.u32()
128152 if !ok || uint64(tabRecordCount) > uint64(reader.remaining())/8 {
129- return domain.CatalogueRecord{}, errMalformedRecord
153+ return storedRecord{}, errMalformedRecord
130154 }
131155 if tabRecordCount > 0 {
132156 record.TabRecords = make([]domain.CatalogueTabRecord, 0, int(tabRecordCount))
@@ -135,21 +159,21 @@ func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error
135159 stableID, stableOK := reader.str()
136160 name, nameOK := reader.str()
137161 if !stableOK || !nameOK {
138- return domain.CatalogueRecord{}, errMalformedRecord
162+ return storedRecord{}, errMalformedRecord
139163 }
140164 record.TabRecords = append(record.TabRecords, domain.CatalogueTabRecord{StableID: domain.TabStableID(stableID), Name: name})
141165 }
142166 }
143167 if record.Committed, ok = reader.ref(); !ok {
144- return domain.CatalogueRecord{}, errMalformedRecord
168+ return storedRecord{}, errMalformedRecord
145169 }
146170 if record.DegradedReason, ok = reader.str(); !ok || reader.remaining() != 0 {
147- return domain.CatalogueRecord{}, errMalformedRecord
171+ return storedRecord{}, errMalformedRecord
148172 }
149173 if err := record.Validate(); err != nil {
150- return domain.CatalogueRecord{}, errors.Join(errMalformedRecord, fmt.Errorf("persist: invalid catalogue record: %w", err))
174+ return storedRecord{}, errors.Join(errMalformedRecord, fmt.Errorf("persist: invalid catalogue record: %w", err))
151175 }
152- return record, nil
176+ return storedRecord{record: record, protocolVersion: protocolVersion}, nil
153177 }
154178
155179 func appendCheckedString(buf []byte, value string) ([]byte, error) {
@@ -7,11 +7,13 @@ import (
7 "math"7 "math"
8 8
9 "github.com/bnema/vev/internal/domain"9 "github.com/bnema/vev/internal/domain"
10+ "github.com/bnema/vev/internal/ports"
10 )11 )
11 12
12 const (13 const (
13- catalogueRecordVersion uint16 = 414+ catalogueRecordVersion uint16 = 5
14- legacyCatalogueRecordVersion uint16 = 315+ protocolLessCatalogueRecordVersion uint16 = 4
16+ legacyCatalogueRecordVersion uint16 = 3
15 )17 )
16 18
17 var errMalformedRecord = errors.New("persist: malformed catalogue record")19 var errMalformedRecord = errors.New("persist: malformed catalogue record")
@@ -22,12 +24,17 @@ var catalogueMagic = [4]byte{'V', 'E', 'V', 'C'}
22 type Record = domain.CatalogueRecord24 type Record = domain.CatalogueRecord
23 25
24 func encodeRecordValue(record domain.CatalogueRecord) ([]byte, error) {26 func encodeRecordValue(record domain.CatalogueRecord) ([]byte, error) {
27+ return encodeRecordValueForProtocol(record, ports.ProtocolVersion)
28+}
29+
30+func encodeRecordValueForProtocol(record domain.CatalogueRecord, protocolVersion uint16) ([]byte, error) {
25 if err := record.Validate(); err != nil {31 if err := record.Validate(); err != nil {
26 return nil, fmt.Errorf("persist: invalid catalogue record: %w", err)32 return nil, fmt.Errorf("persist: invalid catalogue record: %w", err)
27 }33 }
28 buf := make([]byte, 0, 128)34 buf := make([]byte, 0, 128)
29 buf = append(buf, catalogueMagic[:]...)35 buf = append(buf, catalogueMagic[:]...)
30 buf = binary.BigEndian.AppendUint16(buf, catalogueRecordVersion)36 buf = binary.BigEndian.AppendUint16(buf, catalogueRecordVersion)
37+ buf = binary.BigEndian.AppendUint16(buf, protocolVersion)
31 buf = append(buf, record.IncarnationID[:]...)38 buf = append(buf, record.IncarnationID[:]...)
32 var err error39 var err error
33 buf, err = appendCheckedString(buf, record.Cwd)40 buf, err = appendCheckedString(buf, record.Cwd)
@@ -66,45 +73,62 @@ func encodeRecordValue(record domain.CatalogueRecord) ([]byte, error) {
66 return buf, err73 return buf, err
67 }74 }
68 75
76+type storedRecord struct {
77+ record domain.CatalogueRecord
78+ protocolVersion uint16
79+}
80+
69 func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error) {81 func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error) {
82+ stored, err := decodeStoredRecordValue(name, value)
83+ return stored.record, err
84+}
85+
86+func decodeStoredRecordValue(name string, value []byte) (storedRecord, error) {
70 if err := domain.ValidateSessionName(name); err != nil {87 if err := domain.ValidateSessionName(name); err != nil {
71- return domain.CatalogueRecord{}, err88+ return storedRecord{}, err
72 }89 }
73 reader := valueReader{data: value}90 reader := valueReader{data: value}
74 magic, ok := reader.take(4)91 magic, ok := reader.take(4)
75 if !ok || string(magic) != string(catalogueMagic[:]) {92 if !ok || string(magic) != string(catalogueMagic[:]) {
76- return domain.CatalogueRecord{}, errMalformedRecord93+ return storedRecord{}, errMalformedRecord
77 }94 }
78 version, ok := reader.u16()95 version, ok := reader.u16()
79- if !ok || (version != catalogueRecordVersion && version != legacyCatalogueRecordVersion) {96+ if !ok || (version != catalogueRecordVersion && version != protocolLessCatalogueRecordVersion && version != legacyCatalogueRecordVersion) {
80- return domain.CatalogueRecord{}, errMalformedRecord97+ return storedRecord{}, errMalformedRecord
98+ }
99+ var protocolVersion uint16
100+ if version == catalogueRecordVersion {
101+ protocolVersion, ok = reader.u16()
102+ if !ok {
103+ return storedRecord{}, errMalformedRecord
104+ }
81 }105 }
82 id, ok := reader.take(16)106 id, ok := reader.take(16)
83 if !ok {107 if !ok {
84- return domain.CatalogueRecord{}, errMalformedRecord108+ return storedRecord{}, errMalformedRecord
85 }109 }
86 record := domain.CatalogueRecord{Name: name}110 record := domain.CatalogueRecord{Name: name}
87 copy(record.IncarnationID[:], id)111 copy(record.IncarnationID[:], id)
88 var valid bool112 var valid bool
89 if record.Cwd, valid = reader.str(); !valid {113 if record.Cwd, valid = reader.str(); !valid {
90- return domain.CatalogueRecord{}, errMalformedRecord114+ return storedRecord{}, errMalformedRecord
91 }115 }
92 created, ok := reader.u64()116 created, ok := reader.u64()
93 if !ok {117 if !ok {
94- return domain.CatalogueRecord{}, errMalformedRecord118+ return storedRecord{}, errMalformedRecord
95 }119 }
96 record.CreatedAt = int64(created)120 record.CreatedAt = int64(created)
97 updated, ok := reader.u64()121 updated, ok := reader.u64()
98 if !ok {122 if !ok {
99- return domain.CatalogueRecord{}, errMalformedRecord123+ return storedRecord{}, errMalformedRecord
100 }124 }
101 record.UpdatedAt = int64(updated)125 record.UpdatedAt = int64(updated)
102 if record.LastUsedSeq, ok = reader.u64(); !ok {126 if record.LastUsedSeq, ok = reader.u64(); !ok {
103- return domain.CatalogueRecord{}, errMalformedRecord127+ return storedRecord{}, errMalformedRecord
104 }128 }
105 count, ok := reader.u32()129 count, ok := reader.u32()
106 if !ok || uint64(count) > uint64(reader.remaining())/4 {130 if !ok || uint64(count) > uint64(reader.remaining())/4 {
107- return domain.CatalogueRecord{}, errMalformedRecord131+ return storedRecord{}, errMalformedRecord
108 }132 }
109 if count > 0 {133 if count > 0 {
110 record.TabNames = make([]string, 0, int(count))134 record.TabNames = make([]string, 0, int(count))
@@ -112,7 +136,7 @@ func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error
112 for range count {136 for range count {
113 tab, ok := reader.str()137 tab, ok := reader.str()
114 if !ok {138 if !ok {
115- return domain.CatalogueRecord{}, errMalformedRecord139+ return storedRecord{}, errMalformedRecord
116 }140 }
117 record.TabNames = append(record.TabNames, tab)141 record.TabNames = append(record.TabNames, tab)
118 }142 }
@@ -126,7 +150,7 @@ func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error
126 } else {150 } else {
127 tabRecordCount, ok := reader.u32()151 tabRecordCount, ok := reader.u32()
128 if !ok || uint64(tabRecordCount) > uint64(reader.remaining())/8 {152 if !ok || uint64(tabRecordCount) > uint64(reader.remaining())/8 {
129- return domain.CatalogueRecord{}, errMalformedRecord153+ return storedRecord{}, errMalformedRecord
130 }154 }
131 if tabRecordCount > 0 {155 if tabRecordCount > 0 {
132 record.TabRecords = make([]domain.CatalogueTabRecord, 0, int(tabRecordCount))156 record.TabRecords = make([]domain.CatalogueTabRecord, 0, int(tabRecordCount))
@@ -135,21 +159,21 @@ func decodeRecordValue(name string, value []byte) (domain.CatalogueRecord, error
135 stableID, stableOK := reader.str()159 stableID, stableOK := reader.str()
136 name, nameOK := reader.str()160 name, nameOK := reader.str()
137 if !stableOK || !nameOK {161 if !stableOK || !nameOK {
138- return domain.CatalogueRecord{}, errMalformedRecord162+ return storedRecord{}, errMalformedRecord
139 }163 }
140 record.TabRecords = append(record.TabRecords, domain.CatalogueTabRecord{StableID: domain.TabStableID(stableID), Name: name})164 record.TabRecords = append(record.TabRecords, domain.CatalogueTabRecord{StableID: domain.TabStableID(stableID), Name: name})
141 }165 }
142 }166 }
143 if record.Committed, ok = reader.ref(); !ok {167 if record.Committed, ok = reader.ref(); !ok {
144- return domain.CatalogueRecord{}, errMalformedRecord168+ return storedRecord{}, errMalformedRecord
145 }169 }
146 if record.DegradedReason, ok = reader.str(); !ok || reader.remaining() != 0 {170 if record.DegradedReason, ok = reader.str(); !ok || reader.remaining() != 0 {
147- return domain.CatalogueRecord{}, errMalformedRecord171+ return storedRecord{}, errMalformedRecord
148 }172 }
149 if err := record.Validate(); err != nil {173 if err := record.Validate(); err != nil {
150- return domain.CatalogueRecord{}, errors.Join(errMalformedRecord, fmt.Errorf("persist: invalid catalogue record: %w", err))174+ return storedRecord{}, errors.Join(errMalformedRecord, fmt.Errorf("persist: invalid catalogue record: %w", err))
151 }175 }
152- return record, nil176+ return storedRecord{record: record, protocolVersion: protocolVersion}, nil
153 }177 }
154 178
155 func appendCheckedString(buf []byte, value string) ([]byte, error) {179 func appendCheckedString(buf []byte, value string) ([]byte, error) {
M internal/usecase/client/client_test.go +41 -0
@@ -463,6 +463,47 @@ func TestAttachPublishesCommittedRouteSnapshot(t *testing.T) {
463463 require.Empty(t, snapshot.Entries, "the active route is metadata-only")
464464 }
465465
466+func TestLocalOnlyHandoffBetweenLocalSessionsKeepsClientRunning(t *testing.T) {
467+ term := newRunTerminal()
468+ defer term.in.unblock()
469+
470+ firstLifecycle := domain.SessionLifecycleID{1}
471+ secondLifecycle := domain.SessionLifecycleID{2}
472+ welcome := func(name string, lifecycle domain.SessionLifecycleID) ports.Frame {
473+ return frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{
474+ SessionID: name, SessionName: name, ResumeToken: 1, Capabilities: ports.CapabilityResume,
475+ CommittedIdentity: &ports.CommittedRouteIdentity{
476+ Target: ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: name},
477+ },
478+ }))
479+ }
480+ secondTarget := ports.ExactSessionTarget{LifecycleID: secondLifecycle, SessionName: "second"}
481+ first := &recordingTransport{recvs: []recvItem{
482+ {f: welcome("first", firstLifecycle)},
483+ {f: frameOf(ports.MsgAttachTarget, ports.MarshalAttachTarget(ports.AttachTarget{
484+ Session: "second", Intent: ports.IntentAttach, ExactTarget: &secondTarget,
485+ EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
486+ }))},
487+ }}
488+ second := &recordingTransport{recvs: []recvItem{
489+ {f: welcome("second", secondLifecycle)},
490+ {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
491+ }}
492+ dialer := &sequenceDialer{trs: []ports.Transport{first, second}}
493+
494+ err := runTestClient(context.Background(), testDependencies(dialer, term, realClock{}, nil, nil), client.AttachRequest{
495+ Intent: ports.IntentAttach, SessionName: "first",
496+ Origin: ports.RouteOriginLocal, OriginKey: "local",
497+ EnvironmentPolicy: ports.EnvironmentPolicyClientOwned,
498+ })
499+
500+ require.NoError(t, err)
501+ hello := helloFromSend(t, second)
502+ require.Equal(t, "second", hello.Name)
503+ require.Equal(t, ports.EnvironmentPolicyClientOwned, hello.EnvironmentPolicy)
504+ require.Nil(t, hello.RemoteTarget)
505+}
506+
466507 func TestAttachHelloPreservesCompleteAttachRequest(t *testing.T) {
467508 term := newRunTerminal()
468509 defer term.in.unblock()
@@ -463,6 +463,47 @@ func TestAttachPublishesCommittedRouteSnapshot(t *testing.T) {
463 require.Empty(t, snapshot.Entries, "the active route is metadata-only")463 require.Empty(t, snapshot.Entries, "the active route is metadata-only")
464 }464 }
465 465
466+func TestLocalOnlyHandoffBetweenLocalSessionsKeepsClientRunning(t *testing.T) {
467+ term := newRunTerminal()
468+ defer term.in.unblock()
469+
470+ firstLifecycle := domain.SessionLifecycleID{1}
471+ secondLifecycle := domain.SessionLifecycleID{2}
472+ welcome := func(name string, lifecycle domain.SessionLifecycleID) ports.Frame {
473+ return frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{
474+ SessionID: name, SessionName: name, ResumeToken: 1, Capabilities: ports.CapabilityResume,
475+ CommittedIdentity: &ports.CommittedRouteIdentity{
476+ Target: ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: name},
477+ },
478+ }))
479+ }
480+ secondTarget := ports.ExactSessionTarget{LifecycleID: secondLifecycle, SessionName: "second"}
481+ first := &recordingTransport{recvs: []recvItem{
482+ {f: welcome("first", firstLifecycle)},
483+ {f: frameOf(ports.MsgAttachTarget, ports.MarshalAttachTarget(ports.AttachTarget{
484+ Session: "second", Intent: ports.IntentAttach, ExactTarget: &secondTarget,
485+ EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
486+ }))},
487+ }}
488+ second := &recordingTransport{recvs: []recvItem{
489+ {f: welcome("second", secondLifecycle)},
490+ {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
491+ }}
492+ dialer := &sequenceDialer{trs: []ports.Transport{first, second}}
493+
494+ err := runTestClient(context.Background(), testDependencies(dialer, term, realClock{}, nil, nil), client.AttachRequest{
495+ Intent: ports.IntentAttach, SessionName: "first",
496+ Origin: ports.RouteOriginLocal, OriginKey: "local",
497+ EnvironmentPolicy: ports.EnvironmentPolicyClientOwned,
498+ })
499+
500+ require.NoError(t, err)
501+ hello := helloFromSend(t, second)
502+ require.Equal(t, "second", hello.Name)
503+ require.Equal(t, ports.EnvironmentPolicyClientOwned, hello.EnvironmentPolicy)
504+ require.Nil(t, hello.RemoteTarget)
505+}
506+
466 func TestAttachHelloPreservesCompleteAttachRequest(t *testing.T) {507 func TestAttachHelloPreservesCompleteAttachRequest(t *testing.T) {
467 term := newRunTerminal()508 term := newRunTerminal()
468 defer term.in.unblock()509 defer term.in.unblock()
M internal/usecase/client/routes.go +14 -2
@@ -537,6 +537,10 @@ func (l *routeLedger) snapshot() ports.RecentRouteSnapshot {
537537 func (l *routeLedger) samePeerHandoff(active AttachRequest, target ports.AttachTarget) AttachRequest {
538538 origin := normalizeRouteOrigin(active.Origin, active.Remote)
539539 originKey := normalizeRouteOriginKey(active.OriginKey, origin)
540+ environmentPolicy := target.EnvironmentPolicy
541+ if origin == ports.RouteOriginLocal {
542+ environmentPolicy = ports.EnvironmentPolicyClientOwned
543+ }
540544
541545 l.mu.RLock()
542546 defer l.mu.RUnlock()
@@ -553,7 +557,11 @@ func (l *routeLedger) samePeerHandoff(active AttachRequest, target ports.AttachT
553557 // original discovery target would make its Hello request a second,
554558 // incorrect remote handoff instead of a same-peer session switch.
555559 request.RemoteTarget = nil
556- request.EnvironmentPolicy = target.EnvironmentPolicy
560+ if request.EnvironmentPolicy != environmentPolicy && environmentPolicy == ports.EnvironmentPolicyClientOwned {
561+ request.NavigationCapabilities = 0
562+ request.StartupOverlay = ports.StartupOverlayNone
563+ }
564+ request.EnvironmentPolicy = environmentPolicy
557565 request.ExactTarget = target.ExactTarget
558566 if target.ExactTarget == nil {
559567 request.PreferredTabID = ""
@@ -567,7 +575,11 @@ func (l *routeLedger) samePeerHandoff(active AttachRequest, target ports.AttachT
567575 request.ExactTarget = target.ExactTarget
568576 request.PreferredTabID = ""
569577 request.RemoteTarget = nil
570- request.EnvironmentPolicy = target.EnvironmentPolicy
578+ if request.EnvironmentPolicy != environmentPolicy && environmentPolicy == ports.EnvironmentPolicyClientOwned {
579+ request.NavigationCapabilities = 0
580+ request.StartupOverlay = ports.StartupOverlayNone
581+ }
582+ request.EnvironmentPolicy = environmentPolicy
571583 return request
572584 }
573585
@@ -537,6 +537,10 @@ func (l *routeLedger) snapshot() ports.RecentRouteSnapshot {
537 func (l *routeLedger) samePeerHandoff(active AttachRequest, target ports.AttachTarget) AttachRequest {537 func (l *routeLedger) samePeerHandoff(active AttachRequest, target ports.AttachTarget) AttachRequest {
538 origin := normalizeRouteOrigin(active.Origin, active.Remote)538 origin := normalizeRouteOrigin(active.Origin, active.Remote)
539 originKey := normalizeRouteOriginKey(active.OriginKey, origin)539 originKey := normalizeRouteOriginKey(active.OriginKey, origin)
540+ environmentPolicy := target.EnvironmentPolicy
541+ if origin == ports.RouteOriginLocal {
542+ environmentPolicy = ports.EnvironmentPolicyClientOwned
543+ }
540 544
541 l.mu.RLock()545 l.mu.RLock()
542 defer l.mu.RUnlock()546 defer l.mu.RUnlock()
@@ -553,7 +557,11 @@ func (l *routeLedger) samePeerHandoff(active AttachRequest, target ports.AttachT
553 // original discovery target would make its Hello request a second,557 // original discovery target would make its Hello request a second,
554 // incorrect remote handoff instead of a same-peer session switch.558 // incorrect remote handoff instead of a same-peer session switch.
555 request.RemoteTarget = nil559 request.RemoteTarget = nil
556- request.EnvironmentPolicy = target.EnvironmentPolicy560+ if request.EnvironmentPolicy != environmentPolicy && environmentPolicy == ports.EnvironmentPolicyClientOwned {
561+ request.NavigationCapabilities = 0
562+ request.StartupOverlay = ports.StartupOverlayNone
563+ }
564+ request.EnvironmentPolicy = environmentPolicy
557 request.ExactTarget = target.ExactTarget565 request.ExactTarget = target.ExactTarget
558 if target.ExactTarget == nil {566 if target.ExactTarget == nil {
559 request.PreferredTabID = ""567 request.PreferredTabID = ""
@@ -567,7 +575,11 @@ func (l *routeLedger) samePeerHandoff(active AttachRequest, target ports.AttachT
567 request.ExactTarget = target.ExactTarget575 request.ExactTarget = target.ExactTarget
568 request.PreferredTabID = ""576 request.PreferredTabID = ""
569 request.RemoteTarget = nil577 request.RemoteTarget = nil
570- request.EnvironmentPolicy = target.EnvironmentPolicy578+ if request.EnvironmentPolicy != environmentPolicy && environmentPolicy == ports.EnvironmentPolicyClientOwned {
579+ request.NavigationCapabilities = 0
580+ request.StartupOverlay = ports.StartupOverlayNone
581+ }
582+ request.EnvironmentPolicy = environmentPolicy
571 return request583 return request
572 }584 }
573 585
M internal/usecase/client/routes_test.go +41 -1
@@ -143,7 +143,47 @@ func TestRouteLedgerSamePeerHandoffRestoresRouteTab(t *testing.T) {
143143 require.Equal(t, first.target.SessionName, request.SessionName)
144144 require.Equal(t, tt.wantExactTarget, request.ExactTarget)
145145 require.Equal(t, tt.wantTab, request.PreferredTabID)
146- require.Equal(t, ports.EnvironmentPolicyDaemonOwned, request.EnvironmentPolicy)
146+ require.Equal(t, ports.EnvironmentPolicyClientOwned, request.EnvironmentPolicy)
147+ })
148+ }
149+}
150+
151+func TestRouteLedgerSamePeerHandoffClearsDaemonOwnedNavigationForLocalRoutes(t *testing.T) {
152+ active := routeTestCandidate(2, ports.RouteOriginLocal)
153+ active.originKey = "local"
154+ active.request.EnvironmentPolicy = ports.EnvironmentPolicyDaemonOwned
155+ active.request.NavigationCapabilities = ports.NavigationCapabilityHomePicker
156+ active.request.StartupOverlay = ports.StartupOverlaySessionPicker
157+
158+ for _, tt := range []struct {
159+ name string
160+ matched bool
161+ }{
162+ {name: "matched route", matched: true},
163+ {name: "fallback route"},
164+ } {
165+ t.Run(tt.name, func(t *testing.T) {
166+ ledger := newRouteLedger()
167+ target := routeTestTarget(1)
168+ if tt.matched {
169+ stored := routeTestCandidate(1, ports.RouteOriginLocal)
170+ stored.originKey = "local"
171+ stored.request.EnvironmentPolicy = ports.EnvironmentPolicyDaemonOwned
172+ stored.request.NavigationCapabilities = ports.NavigationCapabilityHomePicker
173+ stored.request.StartupOverlay = ports.StartupOverlaySessionPicker
174+ _, err := ledger.commit(stored)
175+ require.NoError(t, err)
176+ }
177+
178+ request := ledger.samePeerHandoff(active.request, ports.AttachTarget{
179+ Session: target.SessionName, Intent: ports.IntentAttach, ExactTarget: &target,
180+ EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
181+ })
182+
183+ require.Equal(t, ports.EnvironmentPolicyClientOwned, request.EnvironmentPolicy)
184+ require.Zero(t, request.NavigationCapabilities)
185+ require.Equal(t, ports.StartupOverlayNone, request.StartupOverlay)
186+ require.NoError(t, validateAttachRequest(request))
147187 })
148188 }
149189 }
@@ -143,7 +143,47 @@ func TestRouteLedgerSamePeerHandoffRestoresRouteTab(t *testing.T) {
143 require.Equal(t, first.target.SessionName, request.SessionName)143 require.Equal(t, first.target.SessionName, request.SessionName)
144 require.Equal(t, tt.wantExactTarget, request.ExactTarget)144 require.Equal(t, tt.wantExactTarget, request.ExactTarget)
145 require.Equal(t, tt.wantTab, request.PreferredTabID)145 require.Equal(t, tt.wantTab, request.PreferredTabID)
146- require.Equal(t, ports.EnvironmentPolicyDaemonOwned, request.EnvironmentPolicy)146+ require.Equal(t, ports.EnvironmentPolicyClientOwned, request.EnvironmentPolicy)
147+ })
148+ }
149+}
150+
151+func TestRouteLedgerSamePeerHandoffClearsDaemonOwnedNavigationForLocalRoutes(t *testing.T) {
152+ active := routeTestCandidate(2, ports.RouteOriginLocal)
153+ active.originKey = "local"
154+ active.request.EnvironmentPolicy = ports.EnvironmentPolicyDaemonOwned
155+ active.request.NavigationCapabilities = ports.NavigationCapabilityHomePicker
156+ active.request.StartupOverlay = ports.StartupOverlaySessionPicker
157+
158+ for _, tt := range []struct {
159+ name string
160+ matched bool
161+ }{
162+ {name: "matched route", matched: true},
163+ {name: "fallback route"},
164+ } {
165+ t.Run(tt.name, func(t *testing.T) {
166+ ledger := newRouteLedger()
167+ target := routeTestTarget(1)
168+ if tt.matched {
169+ stored := routeTestCandidate(1, ports.RouteOriginLocal)
170+ stored.originKey = "local"
171+ stored.request.EnvironmentPolicy = ports.EnvironmentPolicyDaemonOwned
172+ stored.request.NavigationCapabilities = ports.NavigationCapabilityHomePicker
173+ stored.request.StartupOverlay = ports.StartupOverlaySessionPicker
174+ _, err := ledger.commit(stored)
175+ require.NoError(t, err)
176+ }
177+
178+ request := ledger.samePeerHandoff(active.request, ports.AttachTarget{
179+ Session: target.SessionName, Intent: ports.IntentAttach, ExactTarget: &target,
180+ EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
181+ })
182+
183+ require.Equal(t, ports.EnvironmentPolicyClientOwned, request.EnvironmentPolicy)
184+ require.Zero(t, request.NavigationCapabilities)
185+ require.Equal(t, ports.StartupOverlayNone, request.StartupOverlay)
186+ require.NoError(t, validateAttachRequest(request))
147 })187 })
148 }188 }
149 }189 }
M internal/usecase/daemon/navigation_attachment_handoff_test.go +22 -16
@@ -14,26 +14,27 @@ import (
1414
1515 func TestNavigationHandoffsDropReplacedInitiatorWithoutMutation(t *testing.T) {
1616 tests := []struct {
17- name string
18- action string
19- run func(*Daemon, *session, *session, *attachedClient, *attachmentEffectTicket) error
20- check func(*testing.T, *Daemon, *session, *session)
17+ name string
18+ releasedAction string
19+ wantErr bool
20+ run func(*Daemon, *session, *session, *attachedClient, *attachmentEffectTicket) error
21+ check func(*testing.T, *Daemon, *session, *session)
2122 }{
2223 {
23- name: "picker selection", action: "picker-select",
24+ name: "picker selection", releasedAction: "detach",
2425 run: func(d *Daemon, source, target *session, ac *attachedClient, effect *attachmentEffectTicket) error {
2526 d.enterPicker(source, ac)
2627 return d.switchToTargetForAttachment(effect.connectionToken(), picker.Target{Session: target.id, TabIndex: 1}, sessionHandoffGuard{}, "picker-select")
2728 },
2829 },
2930 {
30- name: "palette session", action: "palette-session",
31+ name: "palette session", releasedAction: "detach",
3132 run: func(d *Daemon, _ *session, target *session, _ *attachedClient, effect *attachmentEffectTicket) error {
3233 return d.switchToTargetForAttachment(effect.connectionToken(), picker.Target{Session: target.id, TabIndex: 1}, sessionHandoffGuard{}, "palette-session")
3334 },
3435 },
3536 {
36- name: "session overflow", action: "overflow-session",
37+ name: "session overflow", releasedAction: "detach",
3738 run: func(d *Daemon, source, target *session, ac *attachedClient, effect *attachmentEffectTicket) error {
3839 source.mu.Lock()
3940 source.name = "alpha"
@@ -46,11 +47,11 @@ func TestNavigationHandoffsDropReplacedInitiatorWithoutMutation(t *testing.T) {
4647 },
4748 },
4849 {
49- name: "stopped session resume", action: "stopped-session",
50+ name: "stopped session resume", releasedAction: "detach",
5051 run: func(d *Daemon, _ *session, target *session, _ *attachedClient, effect *attachmentEffectTicket) error {
5152 d.mu.Lock()
5253 delete(d.sessions, target.id)
53- d.stopped["stopped"] = stoppedSession{name: "stopped", cwd: "/tmp", createdAt: 9, state: ports.SessionDown}
54+ d.stopped["stopped"] = stoppedSession{name: "stopped", cwd: "/tmp", createdAt: 9, incarnation: domain.IncarnationID{2}, state: ports.SessionDown}
5455 d.mu.Unlock()
5556 expectedCreatedAt := int64(9)
5657 return d.switchToTargetForAttachment(effect.connectionToken(), picker.Target{Name: "stopped", Stopped: true, ExpectedCreatedAt: &expectedCreatedAt}, sessionHandoffGuard{}, "stopped-session")
@@ -65,7 +66,7 @@ func TestNavigationHandoffsDropReplacedInitiatorWithoutMutation(t *testing.T) {
6566 },
6667 },
6768 {
68- name: "transition prompt creation", action: "create-session",
69+ name: "transition prompt creation", releasedAction: "create-session", wantErr: true,
6970 run: func(d *Daemon, _ *session, _ *session, _ *attachedClient, effect *attachmentEffectTicket) error {
7071 return d.createSessionAndSwitchForAttachment(effect.connectionToken(), "created")
7172 },
@@ -83,7 +84,7 @@ func TestNavigationHandoffsDropReplacedInitiatorWithoutMutation(t *testing.T) {
8384 d, source, old, _, releases := newManualTabSession(t, 2)
8485 defer releaseAll(releases)
8586 d.ptys = newFactorySeq(t)
86- target := &session{sessionCore: sessionCore{id: "target", name: "target"}, ctx: source.ctx, cancel: func() {}, tabs: []*tab{
87+ target := &session{sessionCore: sessionCore{id: "target", name: "target", incarnation: domain.SessionLifecycleID{1}}, ctx: source.ctx, cancel: func() {}, tabs: []*tab{
8788 newTab(nil, domain.Size{Cols: 80, Rows: 23}),
8889 newTab(nil, domain.Size{Cols: 80, Rows: 23}),
8990 }}
@@ -102,7 +103,7 @@ func TestNavigationHandoffsDropReplacedInitiatorWithoutMutation(t *testing.T) {
102103 releaseAction := make(chan struct{})
103104 var admissionEndedOnce sync.Once
104105 d.afterActionAttachmentEffectEnded = func(action string) {
105- if action == tt.action {
106+ if action == tt.releasedAction {
106107 admissionEndedOnce.Do(func() { close(admissionEnded) })
107108 <-releaseAction
108109 }
@@ -122,7 +123,12 @@ func TestNavigationHandoffsDropReplacedInitiatorWithoutMutation(t *testing.T) {
122123 })
123124 require.NoError(t, err)
124125 close(releaseAction)
125- require.Error(t, <-actionDone)
126+ actionErr := <-actionDone
127+ if tt.wantErr {
128+ require.Error(t, actionErr)
129+ } else {
130+ require.NoError(t, actionErr)
131+ }
126132
127133 require.Same(t, source, old.currentSession())
128134 source.mu.Lock()
@@ -143,7 +149,7 @@ func TestNavigationHandoffsDropReplacedInitiatorWithoutMutation(t *testing.T) {
143149 func TestStoppedSessionHandoffDoesNotResumeAfterInitiatorReplacement(t *testing.T) {
144150 d, source, old, _ := newManualSessionWithPTYs(t, nil)
145151 d.mu.Lock()
146- d.stopped["stopped"] = stoppedSession{name: "stopped", cwd: "/tmp", createdAt: 7, state: ports.SessionDown}
152+ d.stopped["stopped"] = stoppedSession{name: "stopped", cwd: "/tmp", createdAt: 7, incarnation: domain.IncarnationID{3}, state: ports.SessionDown}
147153 d.mu.Unlock()
148154
149155 rc := d.attachCoordinator(source, nil, old, true)
@@ -157,7 +163,7 @@ func TestStoppedSessionHandoffDoesNotResumeAfterInitiatorReplacement(t *testing.
157163 release := make(chan struct{})
158164 var endedOnce sync.Once
159165 d.afterActionAttachmentEffectEnded = func(action string) {
160- if action == "picker-stopped" {
166+ if action == "detach" {
161167 endedOnce.Do(func() { close(ended) })
162168 <-release
163169 }
@@ -180,7 +186,7 @@ func TestStoppedSessionHandoffDoesNotResumeAfterInitiatorReplacement(t *testing.
180186 })
181187 require.NoError(t, err)
182188 close(release)
183- require.Error(t, <-done)
189+ require.NoError(t, <-done)
184190
185191 d.mu.Lock()
186192 _, stillStopped := d.stopped["stopped"]
@@ -14,26 +14,27 @@ import (
14 14
15 func TestNavigationHandoffsDropReplacedInitiatorWithoutMutation(t *testing.T) {15 func TestNavigationHandoffsDropReplacedInitiatorWithoutMutation(t *testing.T) {
16 tests := []struct {16 tests := []struct {
17- name string17+ name string
18- action string18+ releasedAction string
19- run func(*Daemon, *session, *session, *attachedClient, *attachmentEffectTicket) error19+ wantErr bool
20- check func(*testing.T, *Daemon, *session, *session)20+ run func(*Daemon, *session, *session, *attachedClient, *attachmentEffectTicket) error
21+ check func(*testing.T, *Daemon, *session, *session)
21 }{22 }{
22 {23 {
23- name: "picker selection", action: "picker-select",24+ name: "picker selection", releasedAction: "detach",
24 run: func(d *Daemon, source, target *session, ac *attachedClient, effect *attachmentEffectTicket) error {25 run: func(d *Daemon, source, target *session, ac *attachedClient, effect *attachmentEffectTicket) error {
25 d.enterPicker(source, ac)26 d.enterPicker(source, ac)
26 return d.switchToTargetForAttachment(effect.connectionToken(), picker.Target{Session: target.id, TabIndex: 1}, sessionHandoffGuard{}, "picker-select")27 return d.switchToTargetForAttachment(effect.connectionToken(), picker.Target{Session: target.id, TabIndex: 1}, sessionHandoffGuard{}, "picker-select")
27 },28 },
28 },29 },
29 {30 {
30- name: "palette session", action: "palette-session",31+ name: "palette session", releasedAction: "detach",
31 run: func(d *Daemon, _ *session, target *session, _ *attachedClient, effect *attachmentEffectTicket) error {32 run: func(d *Daemon, _ *session, target *session, _ *attachedClient, effect *attachmentEffectTicket) error {
32 return d.switchToTargetForAttachment(effect.connectionToken(), picker.Target{Session: target.id, TabIndex: 1}, sessionHandoffGuard{}, "palette-session")33 return d.switchToTargetForAttachment(effect.connectionToken(), picker.Target{Session: target.id, TabIndex: 1}, sessionHandoffGuard{}, "palette-session")
33 },34 },
34 },35 },
35 {36 {
36- name: "session overflow", action: "overflow-session",37+ name: "session overflow", releasedAction: "detach",
37 run: func(d *Daemon, source, target *session, ac *attachedClient, effect *attachmentEffectTicket) error {38 run: func(d *Daemon, source, target *session, ac *attachedClient, effect *attachmentEffectTicket) error {
38 source.mu.Lock()39 source.mu.Lock()
39 source.name = "alpha"40 source.name = "alpha"
@@ -46,11 +47,11 @@ func TestNavigationHandoffsDropReplacedInitiatorWithoutMutation(t *testing.T) {
46 },47 },
47 },48 },
48 {49 {
49- name: "stopped session resume", action: "stopped-session",50+ name: "stopped session resume", releasedAction: "detach",
50 run: func(d *Daemon, _ *session, target *session, _ *attachedClient, effect *attachmentEffectTicket) error {51 run: func(d *Daemon, _ *session, target *session, _ *attachedClient, effect *attachmentEffectTicket) error {
51 d.mu.Lock()52 d.mu.Lock()
52 delete(d.sessions, target.id)53 delete(d.sessions, target.id)
53- d.stopped["stopped"] = stoppedSession{name: "stopped", cwd: "/tmp", createdAt: 9, state: ports.SessionDown}54+ d.stopped["stopped"] = stoppedSession{name: "stopped", cwd: "/tmp", createdAt: 9, incarnation: domain.IncarnationID{2}, state: ports.SessionDown}
54 d.mu.Unlock()55 d.mu.Unlock()
55 expectedCreatedAt := int64(9)56 expectedCreatedAt := int64(9)
56 return d.switchToTargetForAttachment(effect.connectionToken(), picker.Target{Name: "stopped", Stopped: true, ExpectedCreatedAt: &expectedCreatedAt}, sessionHandoffGuard{}, "stopped-session")57 return d.switchToTargetForAttachment(effect.connectionToken(), picker.Target{Name: "stopped", Stopped: true, ExpectedCreatedAt: &expectedCreatedAt}, sessionHandoffGuard{}, "stopped-session")
@@ -65,7 +66,7 @@ func TestNavigationHandoffsDropReplacedInitiatorWithoutMutation(t *testing.T) {
65 },66 },
66 },67 },
67 {68 {
68- name: "transition prompt creation", action: "create-session",69+ name: "transition prompt creation", releasedAction: "create-session", wantErr: true,
69 run: func(d *Daemon, _ *session, _ *session, _ *attachedClient, effect *attachmentEffectTicket) error {70 run: func(d *Daemon, _ *session, _ *session, _ *attachedClient, effect *attachmentEffectTicket) error {
70 return d.createSessionAndSwitchForAttachment(effect.connectionToken(), "created")71 return d.createSessionAndSwitchForAttachment(effect.connectionToken(), "created")
71 },72 },
@@ -83,7 +84,7 @@ func TestNavigationHandoffsDropReplacedInitiatorWithoutMutation(t *testing.T) {
83 d, source, old, _, releases := newManualTabSession(t, 2)84 d, source, old, _, releases := newManualTabSession(t, 2)
84 defer releaseAll(releases)85 defer releaseAll(releases)
85 d.ptys = newFactorySeq(t)86 d.ptys = newFactorySeq(t)
86- target := &session{sessionCore: sessionCore{id: "target", name: "target"}, ctx: source.ctx, cancel: func() {}, tabs: []*tab{87+ target := &session{sessionCore: sessionCore{id: "target", name: "target", incarnation: domain.SessionLifecycleID{1}}, ctx: source.ctx, cancel: func() {}, tabs: []*tab{
87 newTab(nil, domain.Size{Cols: 80, Rows: 23}),88 newTab(nil, domain.Size{Cols: 80, Rows: 23}),
88 newTab(nil, domain.Size{Cols: 80, Rows: 23}),89 newTab(nil, domain.Size{Cols: 80, Rows: 23}),
89 }}90 }}
@@ -102,7 +103,7 @@ func TestNavigationHandoffsDropReplacedInitiatorWithoutMutation(t *testing.T) {
102 releaseAction := make(chan struct{})103 releaseAction := make(chan struct{})
103 var admissionEndedOnce sync.Once104 var admissionEndedOnce sync.Once
104 d.afterActionAttachmentEffectEnded = func(action string) {105 d.afterActionAttachmentEffectEnded = func(action string) {
105- if action == tt.action {106+ if action == tt.releasedAction {
106 admissionEndedOnce.Do(func() { close(admissionEnded) })107 admissionEndedOnce.Do(func() { close(admissionEnded) })
107 <-releaseAction108 <-releaseAction
108 }109 }
@@ -122,7 +123,12 @@ func TestNavigationHandoffsDropReplacedInitiatorWithoutMutation(t *testing.T) {
122 })123 })
123 require.NoError(t, err)124 require.NoError(t, err)
124 close(releaseAction)125 close(releaseAction)
125- require.Error(t, <-actionDone)126+ actionErr := <-actionDone
127+ if tt.wantErr {
128+ require.Error(t, actionErr)
129+ } else {
130+ require.NoError(t, actionErr)
131+ }
126 132
127 require.Same(t, source, old.currentSession())133 require.Same(t, source, old.currentSession())
128 source.mu.Lock()134 source.mu.Lock()
@@ -143,7 +149,7 @@ func TestNavigationHandoffsDropReplacedInitiatorWithoutMutation(t *testing.T) {
143 func TestStoppedSessionHandoffDoesNotResumeAfterInitiatorReplacement(t *testing.T) {149 func TestStoppedSessionHandoffDoesNotResumeAfterInitiatorReplacement(t *testing.T) {
144 d, source, old, _ := newManualSessionWithPTYs(t, nil)150 d, source, old, _ := newManualSessionWithPTYs(t, nil)
145 d.mu.Lock()151 d.mu.Lock()
146- d.stopped["stopped"] = stoppedSession{name: "stopped", cwd: "/tmp", createdAt: 7, state: ports.SessionDown}152+ d.stopped["stopped"] = stoppedSession{name: "stopped", cwd: "/tmp", createdAt: 7, incarnation: domain.IncarnationID{3}, state: ports.SessionDown}
147 d.mu.Unlock()153 d.mu.Unlock()
148 154
149 rc := d.attachCoordinator(source, nil, old, true)155 rc := d.attachCoordinator(source, nil, old, true)
@@ -157,7 +163,7 @@ func TestStoppedSessionHandoffDoesNotResumeAfterInitiatorReplacement(t *testing.
157 release := make(chan struct{})163 release := make(chan struct{})
158 var endedOnce sync.Once164 var endedOnce sync.Once
159 d.afterActionAttachmentEffectEnded = func(action string) {165 d.afterActionAttachmentEffectEnded = func(action string) {
160- if action == "picker-stopped" {166+ if action == "detach" {
161 endedOnce.Do(func() { close(ended) })167 endedOnce.Do(func() { close(ended) })
162 <-release168 <-release
163 }169 }
@@ -180,7 +186,7 @@ func TestStoppedSessionHandoffDoesNotResumeAfterInitiatorReplacement(t *testing.
180 })186 })
181 require.NoError(t, err)187 require.NoError(t, err)
182 close(release)188 close(release)
183- require.Error(t, <-done)189+ require.NoError(t, <-done)
184 190
185 d.mu.Lock()191 d.mu.Lock()
186 _, stillStopped := d.stopped["stopped"]192 _, stillStopped := d.stopped["stopped"]
M internal/usecase/daemon/navigation_attachment_incarnation_test.go +4 -4
@@ -8,7 +8,7 @@ import (
88 "github.com/stretchr/testify/require"
99 )
1010
11-func TestNavigationHandoffRejectsResumedOrReboundInitiatorIncarnation(t *testing.T) {
11+func TestNavigationHandoffDoesNotMutateAfterInitiatorIncarnationChanges(t *testing.T) {
1212 tests := []struct {
1313 name string
1414 rebind func(*testing.T, *Daemon, *session, *attachedClient)
@@ -39,7 +39,7 @@ func TestNavigationHandoffRejectsResumedOrReboundInitiatorIncarnation(t *testing
3939 for _, tt := range tests {
4040 t.Run(tt.name, func(t *testing.T) {
4141 d, source, ac, _ := newManualSessionWithPTYs(t, nil)
42- target := &session{sessionCore: sessionCore{id: "target", name: "target"}, ctx: source.ctx, cancel: func() {}, tabs: []*tab{
42+ target := &session{sessionCore: sessionCore{id: "target", name: "target", incarnation: domain.SessionLifecycleID{1}}, ctx: source.ctx, cancel: func() {}, tabs: []*tab{
4343 newTab(nil, domain.Size{Cols: 80, Rows: 23}),
4444 newTab(nil, domain.Size{Cols: 80, Rows: 23}),
4545 }}
@@ -57,7 +57,7 @@ func TestNavigationHandoffRejectsResumedOrReboundInitiatorIncarnation(t *testing
5757 ended := make(chan struct{})
5858 release := make(chan struct{})
5959 d.afterActionAttachmentEffectEnded = func(action string) {
60- if action == "test-handoff" {
60+ if action == "detach" {
6161 close(ended)
6262 <-release
6363 }
@@ -69,7 +69,7 @@ func TestNavigationHandoffRejectsResumedOrReboundInitiatorIncarnation(t *testing
6969 awaitTestCompletion(t, ended, "test handoff did not release its role ticket")
7070 tt.rebind(t, d, source, ac)
7171 close(release)
72- require.ErrorIs(t, awaitTestValue(t, done, "test handoff did not finish"), errAttachmentTransition)
72+ require.NoError(t, awaitTestValue(t, done, "test handoff did not finish"))
7373
7474 require.Same(t, source, ac.currentSession())
7575 target.mu.Lock()
@@ -8,7 +8,7 @@ import (
8 "github.com/stretchr/testify/require"8 "github.com/stretchr/testify/require"
9 )9 )
10 10
11-func TestNavigationHandoffRejectsResumedOrReboundInitiatorIncarnation(t *testing.T) {11+func TestNavigationHandoffDoesNotMutateAfterInitiatorIncarnationChanges(t *testing.T) {
12 tests := []struct {12 tests := []struct {
13 name string13 name string
14 rebind func(*testing.T, *Daemon, *session, *attachedClient)14 rebind func(*testing.T, *Daemon, *session, *attachedClient)
@@ -39,7 +39,7 @@ func TestNavigationHandoffRejectsResumedOrReboundInitiatorIncarnation(t *testing
39 for _, tt := range tests {39 for _, tt := range tests {
40 t.Run(tt.name, func(t *testing.T) {40 t.Run(tt.name, func(t *testing.T) {
41 d, source, ac, _ := newManualSessionWithPTYs(t, nil)41 d, source, ac, _ := newManualSessionWithPTYs(t, nil)
42- target := &session{sessionCore: sessionCore{id: "target", name: "target"}, ctx: source.ctx, cancel: func() {}, tabs: []*tab{42+ target := &session{sessionCore: sessionCore{id: "target", name: "target", incarnation: domain.SessionLifecycleID{1}}, ctx: source.ctx, cancel: func() {}, tabs: []*tab{
43 newTab(nil, domain.Size{Cols: 80, Rows: 23}),43 newTab(nil, domain.Size{Cols: 80, Rows: 23}),
44 newTab(nil, domain.Size{Cols: 80, Rows: 23}),44 newTab(nil, domain.Size{Cols: 80, Rows: 23}),
45 }}45 }}
@@ -57,7 +57,7 @@ func TestNavigationHandoffRejectsResumedOrReboundInitiatorIncarnation(t *testing
57 ended := make(chan struct{})57 ended := make(chan struct{})
58 release := make(chan struct{})58 release := make(chan struct{})
59 d.afterActionAttachmentEffectEnded = func(action string) {59 d.afterActionAttachmentEffectEnded = func(action string) {
60- if action == "test-handoff" {60+ if action == "detach" {
61 close(ended)61 close(ended)
62 <-release62 <-release
63 }63 }
@@ -69,7 +69,7 @@ func TestNavigationHandoffRejectsResumedOrReboundInitiatorIncarnation(t *testing
69 awaitTestCompletion(t, ended, "test handoff did not release its role ticket")69 awaitTestCompletion(t, ended, "test handoff did not release its role ticket")
70 tt.rebind(t, d, source, ac)70 tt.rebind(t, d, source, ac)
71 close(release)71 close(release)
72- require.ErrorIs(t, awaitTestValue(t, done, "test handoff did not finish"), errAttachmentTransition)72+ require.NoError(t, awaitTestValue(t, done, "test handoff did not finish"))
73 73
74 require.Same(t, source, ac.currentSession())74 require.Same(t, source, ac.currentSession())
75 target.mu.Lock()75 target.mu.Lock()
M internal/usecase/recovery/reset.go +39 -0
@@ -8,6 +8,45 @@ import (
88 "github.com/bnema/vev/internal/domain"
99 )
1010
11+// ResetProtocolIncompatible replaces one exact record with a fresh incarnation
12+// that retains only its name. A protocol boundary never restores metadata or
13+// checkpoints whose semantics may have changed.
14+func (c *Coordinator) ResetProtocolIncompatible(ctx context.Context, name string, expectedIncarnation domain.IncarnationID) (domain.CatalogueRecord, bool, error) {
15+ if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil || c.random == nil {
16+ return domain.CatalogueRecord{}, false, errors.New("recovery: incomplete protocol-reset dependencies")
17+ }
18+ if err := ctx.Err(); err != nil {
19+ return domain.CatalogueRecord{}, false, err
20+ }
21+ c.mutationMu.Lock()
22+ defer c.mutationMu.Unlock()
23+ unlock := c.locks.Lock([]string{name})
24+ defer unlock()
25+
26+ record, ok, err := c.catalogue.Record(name)
27+ if err != nil {
28+ return domain.CatalogueRecord{}, false, err
29+ }
30+ if !ok || record.IncarnationID != expectedIncarnation {
31+ return domain.CatalogueRecord{}, false, nil
32+ }
33+ next, err := domain.NewIncarnationID(c.random)
34+ if err != nil {
35+ return domain.CatalogueRecord{}, false, fmt.Errorf("recovery: generate protocol-reset incarnation: %w", err)
36+ }
37+ fresh := domain.CatalogueRecord{Name: record.Name, IncarnationID: next}
38+ if err := fresh.Validate(); err != nil {
39+ return domain.CatalogueRecord{}, false, fmt.Errorf("recovery: invalid protocol-reset replacement: %w", err)
40+ }
41+ if err := c.catalogue.Replace(record.Name, fresh); err != nil {
42+ return domain.CatalogueRecord{}, false, err
43+ }
44+ if err := c.repository.DeleteIncarnation(ctx, record.IncarnationID); err != nil {
45+ return fresh, true, err
46+ }
47+ return fresh, true, nil
48+}
49+
1150 // ResetIncompatible replaces an exact healthy checkpoint authority with a
1251 // fresh incarnation. Stale observations are harmless no-ops.
1352 func (c *Coordinator) ResetIncompatible(ctx context.Context, name string, expectedIncarnation domain.IncarnationID, expectedCheckpoint domain.CheckpointRef) (domain.CatalogueRecord, bool, error) {
@@ -8,6 +8,45 @@ import (
8 "github.com/bnema/vev/internal/domain"8 "github.com/bnema/vev/internal/domain"
9 )9 )
10 10
11+// ResetProtocolIncompatible replaces one exact record with a fresh incarnation
12+// that retains only its name. A protocol boundary never restores metadata or
13+// checkpoints whose semantics may have changed.
14+func (c *Coordinator) ResetProtocolIncompatible(ctx context.Context, name string, expectedIncarnation domain.IncarnationID) (domain.CatalogueRecord, bool, error) {
15+ if c == nil || c.catalogue == nil || c.repository == nil || c.locks == nil || c.random == nil {
16+ return domain.CatalogueRecord{}, false, errors.New("recovery: incomplete protocol-reset dependencies")
17+ }
18+ if err := ctx.Err(); err != nil {
19+ return domain.CatalogueRecord{}, false, err
20+ }
21+ c.mutationMu.Lock()
22+ defer c.mutationMu.Unlock()
23+ unlock := c.locks.Lock([]string{name})
24+ defer unlock()
25+
26+ record, ok, err := c.catalogue.Record(name)
27+ if err != nil {
28+ return domain.CatalogueRecord{}, false, err
29+ }
30+ if !ok || record.IncarnationID != expectedIncarnation {
31+ return domain.CatalogueRecord{}, false, nil
32+ }
33+ next, err := domain.NewIncarnationID(c.random)
34+ if err != nil {
35+ return domain.CatalogueRecord{}, false, fmt.Errorf("recovery: generate protocol-reset incarnation: %w", err)
36+ }
37+ fresh := domain.CatalogueRecord{Name: record.Name, IncarnationID: next}
38+ if err := fresh.Validate(); err != nil {
39+ return domain.CatalogueRecord{}, false, fmt.Errorf("recovery: invalid protocol-reset replacement: %w", err)
40+ }
41+ if err := c.catalogue.Replace(record.Name, fresh); err != nil {
42+ return domain.CatalogueRecord{}, false, err
43+ }
44+ if err := c.repository.DeleteIncarnation(ctx, record.IncarnationID); err != nil {
45+ return fresh, true, err
46+ }
47+ return fresh, true, nil
48+}
49+
11 // ResetIncompatible replaces an exact healthy checkpoint authority with a50 // ResetIncompatible replaces an exact healthy checkpoint authority with a
12 // fresh incarnation. Stale observations are harmless no-ops.51 // fresh incarnation. Stale observations are harmless no-ops.
13 func (c *Coordinator) ResetIncompatible(ctx context.Context, name string, expectedIncarnation domain.IncarnationID, expectedCheckpoint domain.CheckpointRef) (domain.CatalogueRecord, bool, error) {52 func (c *Coordinator) ResetIncompatible(ctx context.Context, name string, expectedIncarnation domain.IncarnationID, expectedCheckpoint domain.CheckpointRef) (domain.CatalogueRecord, bool, error) {
M internal/usecase/recovery/transactions_test.go +30 -0
@@ -235,6 +235,36 @@ func healthyTransactionRecord() domain.CatalogueRecord {
235235 }
236236 }
237237
238+func TestResetProtocolIncompatibleKeepsOnlySessionName(t *testing.T) {
239+ t.Parallel()
240+ old := healthyTransactionRecord()
241+ old.TabRecords = []domain.CatalogueTabRecord{{StableID: "tab-1", Name: "shell"}}
242+ old.DegradedReason = ""
243+ catalogue := newTransactionCatalogue(old)
244+ repository := &transactionRepository{}
245+ coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
246+
247+ fresh, committed, err := coordinator.ResetProtocolIncompatible(context.Background(), old.Name, old.IncarnationID)
248+
249+ require.NoError(t, err)
250+ require.True(t, committed)
251+ require.Equal(t, old.Name, fresh.Name)
252+ require.Equal(t, domain.IncarnationID(bytes.Repeat([]byte{2}, 16)), fresh.IncarnationID)
253+ require.Empty(t, fresh.Cwd)
254+ require.Zero(t, fresh.CreatedAt)
255+ require.Zero(t, fresh.UpdatedAt)
256+ require.Zero(t, fresh.LastUsedSeq)
257+ require.Empty(t, fresh.TabNames)
258+ require.Empty(t, fresh.TabRecords)
259+ require.Nil(t, fresh.Committed)
260+ require.Empty(t, fresh.DegradedReason)
261+ stored, ok, recordErr := catalogue.Record(old.Name)
262+ require.NoError(t, recordErr)
263+ require.True(t, ok)
264+ require.True(t, fresh.Equal(stored))
265+ require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deleteCalls)
266+}
267+
238268 func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) {
239269 t.Parallel()
240270 old := healthyTransactionRecord()
@@ -235,6 +235,36 @@ func healthyTransactionRecord() domain.CatalogueRecord {
235 }235 }
236 }236 }
237 237
238+func TestResetProtocolIncompatibleKeepsOnlySessionName(t *testing.T) {
239+ t.Parallel()
240+ old := healthyTransactionRecord()
241+ old.TabRecords = []domain.CatalogueTabRecord{{StableID: "tab-1", Name: "shell"}}
242+ old.DegradedReason = ""
243+ catalogue := newTransactionCatalogue(old)
244+ repository := &transactionRepository{}
245+ coordinator := NewCoordinator(catalogue, repository, bytes.NewReader(bytes.Repeat([]byte{2}, 16)))
246+
247+ fresh, committed, err := coordinator.ResetProtocolIncompatible(context.Background(), old.Name, old.IncarnationID)
248+
249+ require.NoError(t, err)
250+ require.True(t, committed)
251+ require.Equal(t, old.Name, fresh.Name)
252+ require.Equal(t, domain.IncarnationID(bytes.Repeat([]byte{2}, 16)), fresh.IncarnationID)
253+ require.Empty(t, fresh.Cwd)
254+ require.Zero(t, fresh.CreatedAt)
255+ require.Zero(t, fresh.UpdatedAt)
256+ require.Zero(t, fresh.LastUsedSeq)
257+ require.Empty(t, fresh.TabNames)
258+ require.Empty(t, fresh.TabRecords)
259+ require.Nil(t, fresh.Committed)
260+ require.Empty(t, fresh.DegradedReason)
261+ stored, ok, recordErr := catalogue.Record(old.Name)
262+ require.NoError(t, recordErr)
263+ require.True(t, ok)
264+ require.True(t, fresh.Equal(stored))
265+ require.Equal(t, []domain.IncarnationID{old.IncarnationID}, repository.deleteCalls)
266+}
267+
238 func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) {268 func TestResetIncompatibleReplacesExactHealthyCheckpoint(t *testing.T) {
239 t.Parallel()269 t.Parallel()
240 old := healthyTransactionRecord()270 old := healthyTransactionRecord()

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close