detached 8b24e21

feat(remote): reduce same-peer switch latency

commit
8b24e21a302659948309a85a02b78a523cd991f2
author
brice · · 2026-08-23 09:48 UTC
parent
83b15cd
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
26 files · +858 -56
Split
UnifiedSplit
M docs/performance.md +19 -0
@@ -87,6 +87,25 @@ For a bounded local check, add `--scenario 1x4-idle-local`. The harness still
8787 validates the complete manifest before selecting that scenario. Do not commit
8888 result directories; raw traces and run manifests are the measurement evidence.
8989
90+## Output snapshot encoding
91+
92+Full state-bearing output snapshots larger than 1 KiB are encoded with
93+standard-library zlib only when the compressed representation is smaller.
94+Incremental output, small snapshots, and incompressible snapshots remain
95+uncompressed. The wire carries a closed compression kind and exact decoded
96+length; decoding rejects unknown kinds, truncation, trailing bytes, integrity
97+failures, and output beyond the frame bound.
98+
99+Measure the representative snapshot encoder with:
100+
101+```sh
102+go test ./internal/ports -run '^$' -bench '^BenchmarkMarshalOutput$' -benchmem
103+```
104+
105+The benchmark reports source throughput, encoded bytes, allocations, and time
106+for empty and styled `120×40` snapshots plus a styled `200×60` snapshot. It is
107+an encoder benchmark, not a mobile-link latency measurement.
108+
90109 ## In-process checks
91110
92111 These commands measure local daemon or VT work only, not end-to-end transport
@@ -87,6 +87,25 @@ For a bounded local check, add `--scenario 1x4-idle-local`. The harness still
87 validates the complete manifest before selecting that scenario. Do not commit87 validates the complete manifest before selecting that scenario. Do not commit
88 result directories; raw traces and run manifests are the measurement evidence.88 result directories; raw traces and run manifests are the measurement evidence.
89 89
90+## Output snapshot encoding
91+
92+Full state-bearing output snapshots larger than 1 KiB are encoded with
93+standard-library zlib only when the compressed representation is smaller.
94+Incremental output, small snapshots, and incompressible snapshots remain
95+uncompressed. The wire carries a closed compression kind and exact decoded
96+length; decoding rejects unknown kinds, truncation, trailing bytes, integrity
97+failures, and output beyond the frame bound.
98+
99+Measure the representative snapshot encoder with:
100+
101+```sh
102+go test ./internal/ports -run '^$' -bench '^BenchmarkMarshalOutput$' -benchmem
103+```
104+
105+The benchmark reports source throughput, encoded bytes, allocations, and time
106+for empty and styled `120×40` snapshots plus a styled `200×60` snapshot. It is
107+an encoder benchmark, not a mobile-link latency measurement.
108+
90 ## In-process checks109 ## In-process checks
91 110
92 These commands measure local daemon or VT work only, not end-to-end transport111 These commands measure local daemon or VT work only, not end-to-end transport
M docs/remote-resilience.md +15 -0
@@ -41,6 +41,16 @@ client republishes its snapshot. These navigation actions use the strict,
4141 exact-match protocol version. Attach handshakes may carry an exact target, and
4242 successful Welcomes can return the daemon's committed identity.
4343
44+Selecting an ordinary active session row on the daemon already serving the
45+attachment uses an in-band switch. The daemon first offers an endpoint-empty,
46+exact lifecycle target; the client confirms it with its remembered stable tab
47+cursor while holding raw input. The daemon either commits the fenced attachment
48+transition and publishes the committed identity before the rebased full paint,
49+or sends a typed pre-commit rejection and leaves the source attachment usable.
50+No hostname, label, DNS result, or SSH alias authorizes reuse. Stopped,
51+explicit-tab, cross-origin, and remote-to-remote selections retain direct
52+close-and-dial handoff.
53+
4454 ## Durable record compatibility
4555
4656 Catalogue record format version 5 stores the protocol version alongside stable
@@ -84,6 +94,11 @@ picker selection changes.
8494 - The UDP carriage authenticates and retransmits protocol frames, preserves
8595 frame order, and reports bounded link-health state without logging payloads,
8696 addresses, keys, or identities.
97+- Large full output snapshots use bounded zlib compression when it reduces the
98+ wire payload. Incremental output and non-beneficial snapshots retain their
99+ canonical encoding. The protocol validates the compression kind, declared
100+ decoded length, stream integrity, and trailing bytes before terminal output
101+ is applied.
87102 - The attachment reconnects with its resume token after a path change. The
88103 remote session and its PTYs remain in place while the attachment is offline.
89104
@@ -41,6 +41,16 @@ client republishes its snapshot. These navigation actions use the strict,
41 exact-match protocol version. Attach handshakes may carry an exact target, and41 exact-match protocol version. Attach handshakes may carry an exact target, and
42 successful Welcomes can return the daemon's committed identity.42 successful Welcomes can return the daemon's committed identity.
43 43
44+Selecting an ordinary active session row on the daemon already serving the
45+attachment uses an in-band switch. The daemon first offers an endpoint-empty,
46+exact lifecycle target; the client confirms it with its remembered stable tab
47+cursor while holding raw input. The daemon either commits the fenced attachment
48+transition and publishes the committed identity before the rebased full paint,
49+or sends a typed pre-commit rejection and leaves the source attachment usable.
50+No hostname, label, DNS result, or SSH alias authorizes reuse. Stopped,
51+explicit-tab, cross-origin, and remote-to-remote selections retain direct
52+close-and-dial handoff.
53+
44 ## Durable record compatibility54 ## Durable record compatibility
45 55
46 Catalogue record format version 5 stores the protocol version alongside stable56 Catalogue record format version 5 stores the protocol version alongside stable
@@ -84,6 +94,11 @@ picker selection changes.
84 - The UDP carriage authenticates and retransmits protocol frames, preserves94 - The UDP carriage authenticates and retransmits protocol frames, preserves
85 frame order, and reports bounded link-health state without logging payloads,95 frame order, and reports bounded link-health state without logging payloads,
86 addresses, keys, or identities.96 addresses, keys, or identities.
97+- Large full output snapshots use bounded zlib compression when it reduces the
98+ wire payload. Incremental output and non-beneficial snapshots retain their
99+ canonical encoding. The protocol validates the compression kind, declared
100+ decoded length, stream integrity, and trailing bytes before terminal output
101+ is applied.
87 - The attachment reconnects with its resume token after a path change. The102 - The attachment reconnects with its resume token after a path change. The
88 remote session and its PTYs remain in place while the attachment is offline.103 remote session and its PTYs remain in place while the attachment is offline.
89 104
M internal/adapters/dgram/network_sim_test.go +5 -2
@@ -876,8 +876,11 @@ func TestFloodOutputPayloadIsIncrementalStateBearingOutput(t *testing.T) {
876876 t.Fatalf("output state = %d -> %d, want %d -> %d", output.Base, output.New, state, state+1)
877877 }
878878 minimumBytes := floodRecordMTUs * mtu
879- if len(output.Data) < minimumBytes || len(payload) < minimumBytes {
880- t.Fatalf("state %d: data bytes=%d encoded bytes=%d, want each at least %d", state, len(output.Data), len(payload), minimumBytes)
879+ if len(output.Data) < minimumBytes {
880+ t.Fatalf("state %d: data bytes=%d, want at least %d", state, len(output.Data), minimumBytes)
881+ }
882+ if !output.Full && len(payload) < minimumBytes {
883+ t.Fatalf("state %d: incremental encoded bytes=%d, want at least %d", state, len(payload), minimumBytes)
881884 }
882885 }
883886 }
@@ -876,8 +876,11 @@ func TestFloodOutputPayloadIsIncrementalStateBearingOutput(t *testing.T) {
876 t.Fatalf("output state = %d -> %d, want %d -> %d", output.Base, output.New, state, state+1)876 t.Fatalf("output state = %d -> %d, want %d -> %d", output.Base, output.New, state, state+1)
877 }877 }
878 minimumBytes := floodRecordMTUs * mtu878 minimumBytes := floodRecordMTUs * mtu
879- if len(output.Data) < minimumBytes || len(payload) < minimumBytes {879+ if len(output.Data) < minimumBytes {
880- t.Fatalf("state %d: data bytes=%d encoded bytes=%d, want each at least %d", state, len(output.Data), len(payload), minimumBytes)880+ t.Fatalf("state %d: data bytes=%d, want at least %d", state, len(output.Data), minimumBytes)
881+ }
882+ if !output.Full && len(payload) < minimumBytes {
883+ t.Fatalf("state %d: incremental encoded bytes=%d, want at least %d", state, len(payload), minimumBytes)
881 }884 }
882 }885 }
883 }886 }
M internal/ports/frame.go +6 -4
@@ -12,7 +12,7 @@ import (
1212 const HandshakeTimeout = 15 * time.Second
1313
1414 // ProtocolVersion is the current vev IPC wire protocol version.
15-const ProtocolVersion uint16 = 32
15+const ProtocolVersion uint16 = 33
1616
1717 // MaxFrameLen is the largest permitted frame length, including the type byte
1818 // and excluding the four-byte length prefix.
@@ -22,9 +22,9 @@ const MaxFrameLen = 16 << 20
2222 type MsgType uint8
2323
2424 // Frame message types use non-contiguous allocations: the legacy client range
25-// is 1–13 and 15, and newer client subscriptions use 32. Server-originated
26-// messages occupy 16–23 and 25–31. Values 14 and 24 remain reserved for future
27-// extensions.
25+// is 1–13 and 15, and newer client controls use 32–33. Server-originated
26+// messages occupy 16–23, 25–31, and 34. Values 14 and 24 remain reserved for
27+// future extensions.
2828 const (
2929 MsgHello MsgType = 1
3030 MsgInput MsgType = 2
@@ -41,6 +41,7 @@ const (
4141 MsgOutputResetRequest MsgType = 13
4242 MsgRemotePreviewRequest MsgType = 15
4343 MsgRouteAttentionSubscription MsgType = 32
44+ MsgSamePeerSwitchRequest MsgType = 33
4445
4546 MsgWelcome MsgType = 16
4647 MsgError MsgType = 17
@@ -59,6 +60,7 @@ const (
5960 MsgNavigateRecentRoute MsgType = 29
6061 MsgRouteNavigationFailure MsgType = 30
6162 MsgRoutePosition MsgType = 31
63+ MsgSamePeerSwitchFailure MsgType = 34
6264 )
6365
6466 // Frame is the unit of exchange over a Transport: a typed, length-delimited
@@ -12,7 +12,7 @@ import (
12 const HandshakeTimeout = 15 * time.Second12 const HandshakeTimeout = 15 * time.Second
13 13
14 // ProtocolVersion is the current vev IPC wire protocol version.14 // ProtocolVersion is the current vev IPC wire protocol version.
15-const ProtocolVersion uint16 = 3215+const ProtocolVersion uint16 = 33
16 16
17 // MaxFrameLen is the largest permitted frame length, including the type byte17 // MaxFrameLen is the largest permitted frame length, including the type byte
18 // and excluding the four-byte length prefix.18 // and excluding the four-byte length prefix.
@@ -22,9 +22,9 @@ const MaxFrameLen = 16 << 20
22 type MsgType uint822 type MsgType uint8
23 23
24 // Frame message types use non-contiguous allocations: the legacy client range24 // Frame message types use non-contiguous allocations: the legacy client range
25-// is 1–13 and 15, and newer client subscriptions use 32. Server-originated25+// is 1–13 and 15, and newer client controls use 32–33. Server-originated
26-// messages occupy 16–23 and 25–31. Values 14 and 24 remain reserved for future26+// messages occupy 16–23, 25–31, and 34. Values 14 and 24 remain reserved for
27-// extensions.27+// future extensions.
28 const (28 const (
29 MsgHello MsgType = 129 MsgHello MsgType = 1
30 MsgInput MsgType = 230 MsgInput MsgType = 2
@@ -41,6 +41,7 @@ const (
41 MsgOutputResetRequest MsgType = 1341 MsgOutputResetRequest MsgType = 13
42 MsgRemotePreviewRequest MsgType = 1542 MsgRemotePreviewRequest MsgType = 15
43 MsgRouteAttentionSubscription MsgType = 3243 MsgRouteAttentionSubscription MsgType = 32
44+ MsgSamePeerSwitchRequest MsgType = 33
44 45
45 MsgWelcome MsgType = 1646 MsgWelcome MsgType = 16
46 MsgError MsgType = 1747 MsgError MsgType = 17
@@ -59,6 +60,7 @@ const (
59 MsgNavigateRecentRoute MsgType = 2960 MsgNavigateRecentRoute MsgType = 29
60 MsgRouteNavigationFailure MsgType = 3061 MsgRouteNavigationFailure MsgType = 30
61 MsgRoutePosition MsgType = 3162 MsgRoutePosition MsgType = 31
63+ MsgSamePeerSwitchFailure MsgType = 34
62 )64 )
63 65
64 // Frame is the unit of exchange over a Transport: a typed, length-delimited66 // Frame is the unit of exchange over a Transport: a typed, length-delimited
M internal/ports/frame_test.go +2 -2
@@ -3,8 +3,8 @@ package ports
33 import "testing"
44
55 func TestProtocolVersion(t *testing.T) {
6- if ProtocolVersion != 32 {
7- t.Fatalf("ProtocolVersion = %d, want 32", ProtocolVersion)
6+ if ProtocolVersion != 33 {
7+ t.Fatalf("ProtocolVersion = %d, want 33", ProtocolVersion)
88 }
99 }
1010
@@ -3,8 +3,8 @@ package ports
3 import "testing"3 import "testing"
4 4
5 func TestProtocolVersion(t *testing.T) {5 func TestProtocolVersion(t *testing.T) {
6- if ProtocolVersion != 32 {6+ if ProtocolVersion != 33 {
7- t.Fatalf("ProtocolVersion = %d, want 32", ProtocolVersion)7+ t.Fatalf("ProtocolVersion = %d, want 33", ProtocolVersion)
8 }8 }
9 }9 }
10 10
M internal/ports/navigation_test.go +1 -1
@@ -62,7 +62,7 @@ func TestHelloNavigationValidationTable(t *testing.T) {
6262 {name: "client-owned home picker rejected", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityHomePicker}, valid: false},
6363 {name: "daemon-owned back picker rejected", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityBack, StartupOverlay: StartupOverlaySessionPicker, EnvironmentPolicy: EnvironmentPolicyDaemonOwned}, valid: false},
6464 {name: "remote-target back picker rejected", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Name: "work", Size: domain.Size{Cols: 80, Rows: 24}, RemoteTarget: remoteTarget, NavigationCapabilities: NavigationCapabilityBack, StartupOverlay: StartupOverlaySessionPicker, EnvironmentPolicy: EnvironmentPolicyDaemonOwned}, valid: false},
65- {name: "back with startup picker", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityBack, StartupOverlay: StartupOverlaySessionPicker}, valid: true, capability: NavigationCapabilityBack, overlay: StartupOverlaySessionPicker, wantPayload: "002002" + "00000000000000000000000000000000" + "0000000000000000" + "0000" + "00500018" + "0000" + "0000" + "00" + "00" + "00000000" + "00" + "00" + "00" + "0000" + "0201"},
65+ {name: "back with startup picker", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityBack, StartupOverlay: StartupOverlaySessionPicker}, valid: true, capability: NavigationCapabilityBack, overlay: StartupOverlaySessionPicker, wantPayload: "002102" + "00000000000000000000000000000000" + "0000000000000000" + "0000" + "00500018" + "0000" + "0000" + "00" + "00" + "00000000" + "00" + "00" + "00" + "0000" + "0201"},
6666 {name: "new rejects navigation", hello: Hello{Version: ProtocolVersion, Intent: IntentNew, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityBack, StartupOverlay: StartupOverlaySessionPicker}, valid: false},
6767 {name: "unknown capability", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: 4}, valid: false},
6868 {name: "back without picker", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityBack}, valid: false},
@@ -62,7 +62,7 @@ func TestHelloNavigationValidationTable(t *testing.T) {
62 {name: "client-owned home picker rejected", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityHomePicker}, valid: false},62 {name: "client-owned home picker rejected", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityHomePicker}, valid: false},
63 {name: "daemon-owned back picker rejected", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityBack, StartupOverlay: StartupOverlaySessionPicker, EnvironmentPolicy: EnvironmentPolicyDaemonOwned}, valid: false},63 {name: "daemon-owned back picker rejected", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityBack, StartupOverlay: StartupOverlaySessionPicker, EnvironmentPolicy: EnvironmentPolicyDaemonOwned}, valid: false},
64 {name: "remote-target back picker rejected", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Name: "work", Size: domain.Size{Cols: 80, Rows: 24}, RemoteTarget: remoteTarget, NavigationCapabilities: NavigationCapabilityBack, StartupOverlay: StartupOverlaySessionPicker, EnvironmentPolicy: EnvironmentPolicyDaemonOwned}, valid: false},64 {name: "remote-target back picker rejected", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Name: "work", Size: domain.Size{Cols: 80, Rows: 24}, RemoteTarget: remoteTarget, NavigationCapabilities: NavigationCapabilityBack, StartupOverlay: StartupOverlaySessionPicker, EnvironmentPolicy: EnvironmentPolicyDaemonOwned}, valid: false},
65- {name: "back with startup picker", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityBack, StartupOverlay: StartupOverlaySessionPicker}, valid: true, capability: NavigationCapabilityBack, overlay: StartupOverlaySessionPicker, wantPayload: "002002" + "00000000000000000000000000000000" + "0000000000000000" + "0000" + "00500018" + "0000" + "0000" + "00" + "00" + "00000000" + "00" + "00" + "00" + "0000" + "0201"},65+ {name: "back with startup picker", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityBack, StartupOverlay: StartupOverlaySessionPicker}, valid: true, capability: NavigationCapabilityBack, overlay: StartupOverlaySessionPicker, wantPayload: "002102" + "00000000000000000000000000000000" + "0000000000000000" + "0000" + "00500018" + "0000" + "0000" + "00" + "00" + "00000000" + "00" + "00" + "00" + "0000" + "0201"},
66 {name: "new rejects navigation", hello: Hello{Version: ProtocolVersion, Intent: IntentNew, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityBack, StartupOverlay: StartupOverlaySessionPicker}, valid: false},66 {name: "new rejects navigation", hello: Hello{Version: ProtocolVersion, Intent: IntentNew, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityBack, StartupOverlay: StartupOverlaySessionPicker}, valid: false},
67 {name: "unknown capability", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: 4}, valid: false},67 {name: "unknown capability", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: 4}, valid: false},
68 {name: "back without picker", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityBack}, valid: false},68 {name: "back without picker", hello: Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: NavigationCapabilityBack}, valid: false},
A internal/ports/output_compression_test.go +124 -0
@@ -0,0 +1,124 @@
1+package ports
2+
3+import (
4+ "bytes"
5+ "encoding/binary"
6+ "testing"
7+
8+ "github.com/bnema/vev/internal/domain"
9+)
10+
11+func TestOutputCompression(t *testing.T) {
12+ full := Output{
13+ Epoch: 1, Base: 0, New: 1, Size: domain.Size{Cols: 120, Rows: 40}, Full: true,
14+ Data: bytes.Repeat([]byte("\x1b[38;2;120;180;240mstyled viewport row\x1b[0m\r\n"), 128),
15+ }
16+ incremental := full
17+ incremental.Base = 1
18+ incremental.New = 2
19+ incremental.Full = false
20+
21+ for _, tt := range []struct {
22+ name string
23+ output Output
24+ wantKind byte
25+ }{
26+ {name: "large full snapshot", output: full, wantKind: outputCompressionZlib},
27+ {name: "incremental output", output: incremental, wantKind: outputCompressionNone},
28+ {name: "small full snapshot", output: Output{Epoch: 1, Base: 0, New: 1, Size: domain.Size{Cols: 80, Rows: 24}, Full: true, Data: []byte("small")}, wantKind: outputCompressionNone},
29+ } {
30+ t.Run(tt.name, func(t *testing.T) {
31+ payload, err := MarshalOutput(tt.output)
32+ if err != nil {
33+ t.Fatal(err)
34+ }
35+ if got := payload[outputHeaderLen]; got != tt.wantKind {
36+ t.Fatalf("compression kind = %d, want %d", got, tt.wantKind)
37+ }
38+ decoded, err := UnmarshalOutput(payload)
39+ if err != nil {
40+ t.Fatal(err)
41+ }
42+ if !bytes.Equal(decoded.Data, tt.output.Data) {
43+ t.Fatalf("decoded data differs\n got: %q\nwant: %q", decoded.Data, tt.output.Data)
44+ }
45+ })
46+ }
47+}
48+
49+func TestCompressedOutputRejectsMalformedPayloads(t *testing.T) {
50+ output := Output{
51+ Epoch: 1, Base: 0, New: 1, Size: domain.Size{Cols: 120, Rows: 40}, Full: true,
52+ Data: bytes.Repeat([]byte("\x1b[31mcompressed\x1b[0m\r\n"), 256),
53+ }
54+ payload, err := MarshalOutput(output)
55+ if err != nil {
56+ t.Fatal(err)
57+ }
58+ if payload[outputHeaderLen] != outputCompressionZlib {
59+ t.Fatal("test fixture was not compressed")
60+ }
61+
62+ for _, tt := range []struct {
63+ name string
64+ mutate func([]byte)
65+ }{
66+ {name: "unknown kind", mutate: func(b []byte) { b[outputHeaderLen] = 99 }},
67+ {name: "decoded length mismatch", mutate: func(b []byte) {
68+ binary.BigEndian.PutUint32(b[outputHeaderLen+1:outputHeaderLen+5], uint32(len(output.Data)-1))
69+ }},
70+ {name: "compressed incremental", mutate: func(b []byte) {
71+ b[outputHeaderLen-1] = 0
72+ binary.BigEndian.PutUint64(b[16:24], 2)
73+ binary.BigEndian.PutUint64(b[8:16], 1)
74+ }},
75+ {name: "corrupt stream", mutate: func(b []byte) { b[len(b)-1] ^= 0xff }},
76+ } {
77+ t.Run(tt.name, func(t *testing.T) {
78+ bad := append([]byte(nil), payload...)
79+ tt.mutate(bad)
80+ if _, err := UnmarshalOutput(bad); err == nil {
81+ t.Fatal("UnmarshalOutput accepted malformed compressed payload")
82+ }
83+ })
84+ }
85+ trailingCompressed := append(append([]byte(nil), payload...), 0)
86+ binary.BigEndian.PutUint32(trailingCompressed[outputHeaderLen+5:outputHeaderLen+9], uint32(len(trailingCompressed)-(outputHeaderLen+9)))
87+ if _, err := UnmarshalOutput(trailingCompressed); err == nil {
88+ t.Fatal("UnmarshalOutput accepted trailing compressed bytes")
89+ }
90+ assertAllPrefixesFail(t, payload, UnmarshalOutput)
91+ assertTrailingGarbageFails(t, payload, UnmarshalOutput)
92+}
93+
94+func BenchmarkMarshalOutput(b *testing.B) {
95+ fixtures := map[string]Output{
96+ "empty-120x40": {
97+ Epoch: 1, Base: 0, New: 1, Size: domain.Size{Cols: 120, Rows: 40}, Full: true,
98+ Data: bytes.Repeat([]byte(" "), 120*40),
99+ },
100+ "styled-120x40": {
101+ Epoch: 1, Base: 0, New: 1, Size: domain.Size{Cols: 120, Rows: 40}, Full: true,
102+ Data: bytes.Repeat([]byte("\x1b[38;2;120;180;240mstyled viewport row\x1b[0m\r\n"), 128),
103+ },
104+ "styled-200x60": {
105+ Epoch: 1, Base: 0, New: 1, Size: domain.Size{Cols: 200, Rows: 60}, Full: true,
106+ Data: bytes.Repeat([]byte("\x1b[48;2;24;32;48m\x1b[38;2;200;220;255mlarge styled viewport row\x1b[0m\r\n"), 256),
107+ },
108+ }
109+ for name, output := range fixtures {
110+ b.Run(name, func(b *testing.B) {
111+ b.ReportAllocs()
112+ b.SetBytes(int64(len(output.Data)))
113+ var encoded []byte
114+ for b.Loop() {
115+ var err error
116+ encoded, err = MarshalOutput(output)
117+ if err != nil {
118+ b.Fatal(err)
119+ }
120+ }
121+ b.ReportMetric(float64(len(encoded)), "encoded-bytes/op")
122+ })
123+ }
124+}
@@ -0,0 +1,124 @@
1+package ports
2+
3+import (
4+ "bytes"
5+ "encoding/binary"
6+ "testing"
7+
8+ "github.com/bnema/vev/internal/domain"
9+)
10+
11+func TestOutputCompression(t *testing.T) {
12+ full := Output{
13+ Epoch: 1, Base: 0, New: 1, Size: domain.Size{Cols: 120, Rows: 40}, Full: true,
14+ Data: bytes.Repeat([]byte("\x1b[38;2;120;180;240mstyled viewport row\x1b[0m\r\n"), 128),
15+ }
16+ incremental := full
17+ incremental.Base = 1
18+ incremental.New = 2
19+ incremental.Full = false
20+
21+ for _, tt := range []struct {
22+ name string
23+ output Output
24+ wantKind byte
25+ }{
26+ {name: "large full snapshot", output: full, wantKind: outputCompressionZlib},
27+ {name: "incremental output", output: incremental, wantKind: outputCompressionNone},
28+ {name: "small full snapshot", output: Output{Epoch: 1, Base: 0, New: 1, Size: domain.Size{Cols: 80, Rows: 24}, Full: true, Data: []byte("small")}, wantKind: outputCompressionNone},
29+ } {
30+ t.Run(tt.name, func(t *testing.T) {
31+ payload, err := MarshalOutput(tt.output)
32+ if err != nil {
33+ t.Fatal(err)
34+ }
35+ if got := payload[outputHeaderLen]; got != tt.wantKind {
36+ t.Fatalf("compression kind = %d, want %d", got, tt.wantKind)
37+ }
38+ decoded, err := UnmarshalOutput(payload)
39+ if err != nil {
40+ t.Fatal(err)
41+ }
42+ if !bytes.Equal(decoded.Data, tt.output.Data) {
43+ t.Fatalf("decoded data differs\n got: %q\nwant: %q", decoded.Data, tt.output.Data)
44+ }
45+ })
46+ }
47+}
48+
49+func TestCompressedOutputRejectsMalformedPayloads(t *testing.T) {
50+ output := Output{
51+ Epoch: 1, Base: 0, New: 1, Size: domain.Size{Cols: 120, Rows: 40}, Full: true,
52+ Data: bytes.Repeat([]byte("\x1b[31mcompressed\x1b[0m\r\n"), 256),
53+ }
54+ payload, err := MarshalOutput(output)
55+ if err != nil {
56+ t.Fatal(err)
57+ }
58+ if payload[outputHeaderLen] != outputCompressionZlib {
59+ t.Fatal("test fixture was not compressed")
60+ }
61+
62+ for _, tt := range []struct {
63+ name string
64+ mutate func([]byte)
65+ }{
66+ {name: "unknown kind", mutate: func(b []byte) { b[outputHeaderLen] = 99 }},
67+ {name: "decoded length mismatch", mutate: func(b []byte) {
68+ binary.BigEndian.PutUint32(b[outputHeaderLen+1:outputHeaderLen+5], uint32(len(output.Data)-1))
69+ }},
70+ {name: "compressed incremental", mutate: func(b []byte) {
71+ b[outputHeaderLen-1] = 0
72+ binary.BigEndian.PutUint64(b[16:24], 2)
73+ binary.BigEndian.PutUint64(b[8:16], 1)
74+ }},
75+ {name: "corrupt stream", mutate: func(b []byte) { b[len(b)-1] ^= 0xff }},
76+ } {
77+ t.Run(tt.name, func(t *testing.T) {
78+ bad := append([]byte(nil), payload...)
79+ tt.mutate(bad)
80+ if _, err := UnmarshalOutput(bad); err == nil {
81+ t.Fatal("UnmarshalOutput accepted malformed compressed payload")
82+ }
83+ })
84+ }
85+ trailingCompressed := append(append([]byte(nil), payload...), 0)
86+ binary.BigEndian.PutUint32(trailingCompressed[outputHeaderLen+5:outputHeaderLen+9], uint32(len(trailingCompressed)-(outputHeaderLen+9)))
87+ if _, err := UnmarshalOutput(trailingCompressed); err == nil {
88+ t.Fatal("UnmarshalOutput accepted trailing compressed bytes")
89+ }
90+ assertAllPrefixesFail(t, payload, UnmarshalOutput)
91+ assertTrailingGarbageFails(t, payload, UnmarshalOutput)
92+}
93+
94+func BenchmarkMarshalOutput(b *testing.B) {
95+ fixtures := map[string]Output{
96+ "empty-120x40": {
97+ Epoch: 1, Base: 0, New: 1, Size: domain.Size{Cols: 120, Rows: 40}, Full: true,
98+ Data: bytes.Repeat([]byte(" "), 120*40),
99+ },
100+ "styled-120x40": {
101+ Epoch: 1, Base: 0, New: 1, Size: domain.Size{Cols: 120, Rows: 40}, Full: true,
102+ Data: bytes.Repeat([]byte("\x1b[38;2;120;180;240mstyled viewport row\x1b[0m\r\n"), 128),
103+ },
104+ "styled-200x60": {
105+ Epoch: 1, Base: 0, New: 1, Size: domain.Size{Cols: 200, Rows: 60}, Full: true,
106+ Data: bytes.Repeat([]byte("\x1b[48;2;24;32;48m\x1b[38;2;200;220;255mlarge styled viewport row\x1b[0m\r\n"), 256),
107+ },
108+ }
109+ for name, output := range fixtures {
110+ b.Run(name, func(b *testing.B) {
111+ b.ReportAllocs()
112+ b.SetBytes(int64(len(output.Data)))
113+ var encoded []byte
114+ for b.Loop() {
115+ var err error
116+ encoded, err = MarshalOutput(output)
117+ if err != nil {
118+ b.Fatal(err)
119+ }
120+ }
121+ b.ReportMetric(float64(len(encoded)), "encoded-bytes/op")
122+ })
123+ }
124+}
M internal/ports/routes.go +50 -0
@@ -230,6 +230,56 @@ type RouteNavigationAction struct {
230230 Generation uint64
231231 }
232232
233+// SamePeerSwitchRequest confirms a daemon-offered endpoint-empty target. It
234+// carries only the exact lifecycle identity and the client-owned tab cursor;
235+// transport origin remains proven by the existing authenticated connection.
236+type SamePeerSwitchRequest struct {
237+ RequestID uint64
238+ Target ExactSessionTarget
239+ PreferredTabID domain.TabStableID
240+}
241+
242+func (r SamePeerSwitchRequest) Validate() error {
243+ if r.RequestID == 0 {
244+ return ErrInvalidRouteWire
245+ }
246+ if err := r.Target.Validate(); err != nil {
247+ return ErrInvalidRouteWire
248+ }
249+ if r.PreferredTabID != "" {
250+ if err := domain.ValidateTabStableID(r.PreferredTabID); err != nil {
251+ return ErrInvalidRouteWire
252+ }
253+ }
254+ return nil
255+}
256+
257+// SamePeerSwitchFailureCode is a closed pre-commit rejection taxonomy.
258+type SamePeerSwitchFailureCode uint8
259+
260+const (
261+ SamePeerSwitchStaleTarget SamePeerSwitchFailureCode = iota + 1
262+ SamePeerSwitchUnavailable
263+)
264+
265+func (c SamePeerSwitchFailureCode) valid() bool {
266+ return c == SamePeerSwitchStaleTarget || c == SamePeerSwitchUnavailable
267+}
268+
269+// SamePeerSwitchFailure leaves the source attachment unchanged. RequestID
270+// rejects a delayed failure after a later user selection.
271+type SamePeerSwitchFailure struct {
272+ RequestID uint64
273+ Code SamePeerSwitchFailureCode
274+}
275+
276+func (f SamePeerSwitchFailure) Validate() error {
277+ if f.RequestID == 0 || !f.Code.valid() {
278+ return ErrInvalidRouteWire
279+ }
280+ return nil
281+}
282+
233283 // RouteNavigationFailure is a bounded taxonomy for a rejected or stale route
234284 // action. User-facing text remains local to the receiving client.
235285 type RouteNavigationFailure struct {
@@ -230,6 +230,56 @@ type RouteNavigationAction struct {
230 Generation uint64230 Generation uint64
231 }231 }
232 232
233+// SamePeerSwitchRequest confirms a daemon-offered endpoint-empty target. It
234+// carries only the exact lifecycle identity and the client-owned tab cursor;
235+// transport origin remains proven by the existing authenticated connection.
236+type SamePeerSwitchRequest struct {
237+ RequestID uint64
238+ Target ExactSessionTarget
239+ PreferredTabID domain.TabStableID
240+}
241+
242+func (r SamePeerSwitchRequest) Validate() error {
243+ if r.RequestID == 0 {
244+ return ErrInvalidRouteWire
245+ }
246+ if err := r.Target.Validate(); err != nil {
247+ return ErrInvalidRouteWire
248+ }
249+ if r.PreferredTabID != "" {
250+ if err := domain.ValidateTabStableID(r.PreferredTabID); err != nil {
251+ return ErrInvalidRouteWire
252+ }
253+ }
254+ return nil
255+}
256+
257+// SamePeerSwitchFailureCode is a closed pre-commit rejection taxonomy.
258+type SamePeerSwitchFailureCode uint8
259+
260+const (
261+ SamePeerSwitchStaleTarget SamePeerSwitchFailureCode = iota + 1
262+ SamePeerSwitchUnavailable
263+)
264+
265+func (c SamePeerSwitchFailureCode) valid() bool {
266+ return c == SamePeerSwitchStaleTarget || c == SamePeerSwitchUnavailable
267+}
268+
269+// SamePeerSwitchFailure leaves the source attachment unchanged. RequestID
270+// rejects a delayed failure after a later user selection.
271+type SamePeerSwitchFailure struct {
272+ RequestID uint64
273+ Code SamePeerSwitchFailureCode
274+}
275+
276+func (f SamePeerSwitchFailure) Validate() error {
277+ if f.RequestID == 0 || !f.Code.valid() {
278+ return ErrInvalidRouteWire
279+ }
280+ return nil
281+}
282+
233 // RouteNavigationFailure is a bounded taxonomy for a rejected or stale route283 // RouteNavigationFailure is a bounded taxonomy for a rejected or stale route
234 // action. User-facing text remains local to the receiving client.284 // action. User-facing text remains local to the receiving client.
235 type RouteNavigationFailure struct {285 type RouteNavigationFailure struct {
M internal/ports/routes_wire.go +71 -0
@@ -98,6 +98,77 @@ func UnmarshalExactSessionTarget(b []byte) (ExactSessionTarget, error) {
9898 return target, nil
9999 }
100100
101+// MarshalSamePeerSwitchRequest encodes the strict in-band session switch
102+// confirmation. The target is non-optional and carries no display origin.
103+func MarshalSamePeerSwitchRequest(request SamePeerSwitchRequest) ([]byte, error) {
104+ if err := request.Validate(); err != nil {
105+ return nil, err
106+ }
107+ w := payloadWriter{}
108+ w.putUint64(request.RequestID)
109+ marshalExactSessionTarget(&w, request.Target)
110+ w.putString(string(request.PreferredTabID))
111+ return w.b, nil
112+}
113+
114+// UnmarshalSamePeerSwitchRequest decodes one complete in-band switch request.
115+func UnmarshalSamePeerSwitchRequest(b []byte) (SamePeerSwitchRequest, error) {
116+ r := payloadReader{b: b}
117+ var request SamePeerSwitchRequest
118+ var err error
119+ if request.RequestID, err = r.getUint64(); err != nil {
120+ return SamePeerSwitchRequest{}, err
121+ }
122+ if request.Target, err = unmarshalExactSessionTarget(&r); err != nil {
123+ return SamePeerSwitchRequest{}, err
124+ }
125+ preferred, err := r.getString()
126+ if err != nil {
127+ return SamePeerSwitchRequest{}, err
128+ }
129+ request.PreferredTabID = domain.TabStableID(preferred)
130+ if err := r.done(); err != nil {
131+ return SamePeerSwitchRequest{}, err
132+ }
133+ if err := request.Validate(); err != nil {
134+ return SamePeerSwitchRequest{}, err
135+ }
136+ return request, nil
137+}
138+
139+// MarshalSamePeerSwitchFailure encodes a bounded pre-commit rejection.
140+func MarshalSamePeerSwitchFailure(failure SamePeerSwitchFailure) ([]byte, error) {
141+ if err := failure.Validate(); err != nil {
142+ return nil, err
143+ }
144+ w := payloadWriter{}
145+ w.putUint64(failure.RequestID)
146+ w.putUint8(uint8(failure.Code))
147+ return w.b, nil
148+}
149+
150+// UnmarshalSamePeerSwitchFailure decodes a complete bounded rejection.
151+func UnmarshalSamePeerSwitchFailure(b []byte) (SamePeerSwitchFailure, error) {
152+ r := payloadReader{b: b}
153+ var failure SamePeerSwitchFailure
154+ var err error
155+ if failure.RequestID, err = r.getUint64(); err != nil {
156+ return SamePeerSwitchFailure{}, err
157+ }
158+ code, err := r.getUint8()
159+ if err != nil {
160+ return SamePeerSwitchFailure{}, err
161+ }
162+ failure.Code = SamePeerSwitchFailureCode(code)
163+ if err := r.done(); err != nil {
164+ return SamePeerSwitchFailure{}, err
165+ }
166+ if err := failure.Validate(); err != nil {
167+ return SamePeerSwitchFailure{}, err
168+ }
169+ return failure, nil
170+}
171+
101172 func validateCommittedRouteIdentity(identity CommittedRouteIdentity) error {
102173 if err := identity.Validate(); err != nil {
103174 return fmt.Errorf("%w: committed identity: %v", ErrInvalidRouteWire, err)
@@ -98,6 +98,77 @@ func UnmarshalExactSessionTarget(b []byte) (ExactSessionTarget, error) {
98 return target, nil98 return target, nil
99 }99 }
100 100
101+// MarshalSamePeerSwitchRequest encodes the strict in-band session switch
102+// confirmation. The target is non-optional and carries no display origin.
103+func MarshalSamePeerSwitchRequest(request SamePeerSwitchRequest) ([]byte, error) {
104+ if err := request.Validate(); err != nil {
105+ return nil, err
106+ }
107+ w := payloadWriter{}
108+ w.putUint64(request.RequestID)
109+ marshalExactSessionTarget(&w, request.Target)
110+ w.putString(string(request.PreferredTabID))
111+ return w.b, nil
112+}
113+
114+// UnmarshalSamePeerSwitchRequest decodes one complete in-band switch request.
115+func UnmarshalSamePeerSwitchRequest(b []byte) (SamePeerSwitchRequest, error) {
116+ r := payloadReader{b: b}
117+ var request SamePeerSwitchRequest
118+ var err error
119+ if request.RequestID, err = r.getUint64(); err != nil {
120+ return SamePeerSwitchRequest{}, err
121+ }
122+ if request.Target, err = unmarshalExactSessionTarget(&r); err != nil {
123+ return SamePeerSwitchRequest{}, err
124+ }
125+ preferred, err := r.getString()
126+ if err != nil {
127+ return SamePeerSwitchRequest{}, err
128+ }
129+ request.PreferredTabID = domain.TabStableID(preferred)
130+ if err := r.done(); err != nil {
131+ return SamePeerSwitchRequest{}, err
132+ }
133+ if err := request.Validate(); err != nil {
134+ return SamePeerSwitchRequest{}, err
135+ }
136+ return request, nil
137+}
138+
139+// MarshalSamePeerSwitchFailure encodes a bounded pre-commit rejection.
140+func MarshalSamePeerSwitchFailure(failure SamePeerSwitchFailure) ([]byte, error) {
141+ if err := failure.Validate(); err != nil {
142+ return nil, err
143+ }
144+ w := payloadWriter{}
145+ w.putUint64(failure.RequestID)
146+ w.putUint8(uint8(failure.Code))
147+ return w.b, nil
148+}
149+
150+// UnmarshalSamePeerSwitchFailure decodes a complete bounded rejection.
151+func UnmarshalSamePeerSwitchFailure(b []byte) (SamePeerSwitchFailure, error) {
152+ r := payloadReader{b: b}
153+ var failure SamePeerSwitchFailure
154+ var err error
155+ if failure.RequestID, err = r.getUint64(); err != nil {
156+ return SamePeerSwitchFailure{}, err
157+ }
158+ code, err := r.getUint8()
159+ if err != nil {
160+ return SamePeerSwitchFailure{}, err
161+ }
162+ failure.Code = SamePeerSwitchFailureCode(code)
163+ if err := r.done(); err != nil {
164+ return SamePeerSwitchFailure{}, err
165+ }
166+ if err := failure.Validate(); err != nil {
167+ return SamePeerSwitchFailure{}, err
168+ }
169+ return failure, nil
170+}
171+
101 func validateCommittedRouteIdentity(identity CommittedRouteIdentity) error {172 func validateCommittedRouteIdentity(identity CommittedRouteIdentity) error {
102 if err := identity.Validate(); err != nil {173 if err := identity.Validate(); err != nil {
103 return fmt.Errorf("%w: committed identity: %v", ErrInvalidRouteWire, err)174 return fmt.Errorf("%w: committed identity: %v", ErrInvalidRouteWire, err)
A internal/ports/same_peer_switch_test.go +72 -0
@@ -0,0 +1,72 @@
1+package ports
2+
3+import (
4+ "bytes"
5+ "testing"
6+
7+ "github.com/bnema/vev/internal/domain"
8+)
9+
10+func samePeerSwitchTarget() ExactSessionTarget {
11+ var lifecycle domain.SessionLifecycleID
12+ lifecycle[0] = 9
13+ return ExactSessionTarget{LifecycleID: lifecycle, SessionName: "work"}
14+}
15+
16+func TestSamePeerSwitchWireStrict(t *testing.T) {
17+ request := SamePeerSwitchRequest{RequestID: 7, Target: samePeerSwitchTarget(), PreferredTabID: "tab-2"}
18+ payload, err := MarshalSamePeerSwitchRequest(request)
19+ if err != nil {
20+ t.Fatal(err)
21+ }
22+ want := append([]byte{0, 0, 0, 0, 0, 0, 0, 7, 9}, make([]byte, 15)...)
23+ want = append(want, 0, 4, 'w', 'o', 'r', 'k', 0, 5, 't', 'a', 'b', '-', '2')
24+ if !bytes.Equal(payload, want) {
25+ t.Fatalf("request payload = %x, want %x", payload, want)
26+ }
27+ decoded, err := UnmarshalSamePeerSwitchRequest(payload)
28+ if err != nil || decoded != request {
29+ t.Fatalf("request = %+v, error = %v, want %+v", decoded, err, request)
30+ }
31+ assertAllPrefixesFail(t, payload, UnmarshalSamePeerSwitchRequest)
32+ assertTrailingGarbageFails(t, payload, UnmarshalSamePeerSwitchRequest)
33+
34+ failure := SamePeerSwitchFailure{RequestID: 7, Code: SamePeerSwitchStaleTarget}
35+ failurePayload, err := MarshalSamePeerSwitchFailure(failure)
36+ if err != nil {
37+ t.Fatal(err)
38+ }
39+ if want := []byte{0, 0, 0, 0, 0, 0, 0, 7, 1}; !bytes.Equal(failurePayload, want) {
40+ t.Fatalf("failure payload = %x, want %x", failurePayload, want)
41+ }
42+ decodedFailure, err := UnmarshalSamePeerSwitchFailure(failurePayload)
43+ if err != nil || decodedFailure != failure {
44+ t.Fatalf("failure = %+v, error = %v, want %+v", decodedFailure, err, failure)
45+ }
46+ assertAllPrefixesFail(t, failurePayload, UnmarshalSamePeerSwitchFailure)
47+ assertTrailingGarbageFails(t, failurePayload, UnmarshalSamePeerSwitchFailure)
48+}
49+
50+func TestSamePeerSwitchWireRejectsInvalidValues(t *testing.T) {
51+ valid := SamePeerSwitchRequest{RequestID: 1, Target: samePeerSwitchTarget()}
52+ for _, request := range []SamePeerSwitchRequest{
53+ {},
54+ {RequestID: 1},
55+ {RequestID: 1, Target: samePeerSwitchTarget(), PreferredTabID: "bad tab"},
56+ } {
57+ if payload, err := MarshalSamePeerSwitchRequest(request); err == nil || payload != nil {
58+ t.Fatalf("MarshalSamePeerSwitchRequest(%+v) = %x, %v; want error", request, payload, err)
59+ }
60+ }
61+ payload, err := MarshalSamePeerSwitchRequest(valid)
62+ if err != nil {
63+ t.Fatal(err)
64+ }
65+ payload[7] = 0
66+ if _, err := UnmarshalSamePeerSwitchRequest(payload); err == nil {
67+ t.Fatal("UnmarshalSamePeerSwitchRequest accepted an invalid request ID")
68+ }
69+ if _, err := UnmarshalSamePeerSwitchFailure([]byte{0, 0, 0, 0, 0, 0, 0, 1, 99}); err == nil {
70+ t.Fatal("UnmarshalSamePeerSwitchFailure accepted an unknown code")
71+ }
72+}
@@ -0,0 +1,72 @@
1+package ports
2+
3+import (
4+ "bytes"
5+ "testing"
6+
7+ "github.com/bnema/vev/internal/domain"
8+)
9+
10+func samePeerSwitchTarget() ExactSessionTarget {
11+ var lifecycle domain.SessionLifecycleID
12+ lifecycle[0] = 9
13+ return ExactSessionTarget{LifecycleID: lifecycle, SessionName: "work"}
14+}
15+
16+func TestSamePeerSwitchWireStrict(t *testing.T) {
17+ request := SamePeerSwitchRequest{RequestID: 7, Target: samePeerSwitchTarget(), PreferredTabID: "tab-2"}
18+ payload, err := MarshalSamePeerSwitchRequest(request)
19+ if err != nil {
20+ t.Fatal(err)
21+ }
22+ want := append([]byte{0, 0, 0, 0, 0, 0, 0, 7, 9}, make([]byte, 15)...)
23+ want = append(want, 0, 4, 'w', 'o', 'r', 'k', 0, 5, 't', 'a', 'b', '-', '2')
24+ if !bytes.Equal(payload, want) {
25+ t.Fatalf("request payload = %x, want %x", payload, want)
26+ }
27+ decoded, err := UnmarshalSamePeerSwitchRequest(payload)
28+ if err != nil || decoded != request {
29+ t.Fatalf("request = %+v, error = %v, want %+v", decoded, err, request)
30+ }
31+ assertAllPrefixesFail(t, payload, UnmarshalSamePeerSwitchRequest)
32+ assertTrailingGarbageFails(t, payload, UnmarshalSamePeerSwitchRequest)
33+
34+ failure := SamePeerSwitchFailure{RequestID: 7, Code: SamePeerSwitchStaleTarget}
35+ failurePayload, err := MarshalSamePeerSwitchFailure(failure)
36+ if err != nil {
37+ t.Fatal(err)
38+ }
39+ if want := []byte{0, 0, 0, 0, 0, 0, 0, 7, 1}; !bytes.Equal(failurePayload, want) {
40+ t.Fatalf("failure payload = %x, want %x", failurePayload, want)
41+ }
42+ decodedFailure, err := UnmarshalSamePeerSwitchFailure(failurePayload)
43+ if err != nil || decodedFailure != failure {
44+ t.Fatalf("failure = %+v, error = %v, want %+v", decodedFailure, err, failure)
45+ }
46+ assertAllPrefixesFail(t, failurePayload, UnmarshalSamePeerSwitchFailure)
47+ assertTrailingGarbageFails(t, failurePayload, UnmarshalSamePeerSwitchFailure)
48+}
49+
50+func TestSamePeerSwitchWireRejectsInvalidValues(t *testing.T) {
51+ valid := SamePeerSwitchRequest{RequestID: 1, Target: samePeerSwitchTarget()}
52+ for _, request := range []SamePeerSwitchRequest{
53+ {},
54+ {RequestID: 1},
55+ {RequestID: 1, Target: samePeerSwitchTarget(), PreferredTabID: "bad tab"},
56+ } {
57+ if payload, err := MarshalSamePeerSwitchRequest(request); err == nil || payload != nil {
58+ t.Fatalf("MarshalSamePeerSwitchRequest(%+v) = %x, %v; want error", request, payload, err)
59+ }
60+ }
61+ payload, err := MarshalSamePeerSwitchRequest(valid)
62+ if err != nil {
63+ t.Fatal(err)
64+ }
65+ payload[7] = 0
66+ if _, err := UnmarshalSamePeerSwitchRequest(payload); err == nil {
67+ t.Fatal("UnmarshalSamePeerSwitchRequest accepted an invalid request ID")
68+ }
69+ if _, err := UnmarshalSamePeerSwitchFailure([]byte{0, 0, 0, 0, 0, 0, 0, 1, 99}); err == nil {
70+ t.Fatal("UnmarshalSamePeerSwitchFailure accepted an unknown code")
71+ }
72+}
M internal/ports/wire.go +111 -13
@@ -7,10 +7,14 @@
77 package ports
88
99 import (
10+ "bytes"
11+ "compress/zlib"
1012 "encoding/binary"
1113 "errors"
1214 "fmt"
15+ "io"
1316 "math"
17+ "sync"
1418
1519 renderer "github.com/bnema/vev-vt"
1620 "github.com/bnema/vev/internal/domain"
@@ -29,10 +33,26 @@ var errInvalidBoolean = errors.New("ports: invalid boolean flag")
2933 var errInvalidEnum = errors.New("ports: invalid enum value")
3034
3135 const (
32- outputHeaderLen = 5*8 + 2*2 + 1
33- outputPayloadOverhead = outputHeaderLen + 4
36+ outputHeaderLen = 5*8 + 2*2 + 1
37+ outputCompressionHeaderLen = 1 + 4
38+ outputPayloadOverhead = outputHeaderLen + outputCompressionHeaderLen + 4
39+ // outputCompressionThreshold keeps small snapshots on the canonical path.
40+ outputCompressionThreshold = 1024
3441 )
3542
43+const (
44+ outputCompressionNone byte = iota
45+ outputCompressionZlib
46+)
47+
48+var outputCompressorPool = sync.Pool{New: func() any {
49+ writer, err := zlib.NewWriterLevel(io.Discard, zlib.BestSpeed)
50+ if err != nil {
51+ panic(err)
52+ }
53+ return writer
54+}}
55+
3656 // ErrInvalidSize reports a terminal size that cannot be represented by the
3757 // protocol or would exceed the bounded screen area.
3858 var ErrInvalidSize = errors.New("ports: invalid terminal size")
@@ -1402,11 +1422,16 @@ func UnmarshalErrorMsg(b []byte) (ErrorMsg, error) {
14021422 }
14031423
14041424 // MarshalOutput encodes m into the epoch/base/new/echo/viewRevision/
1405-// size/full/data message layout.
1425+// size/full/compression/decoded-length/data message layout. Compression is
1426+// limited to large, full snapshots and is retained only when it saves bytes.
14061427 func MarshalOutput(m Output) ([]byte, error) {
14071428 if err := ValidateOutput(m); err != nil {
14081429 return nil, err
14091430 }
1431+ kind, data, err := compressOutputData(m)
1432+ if err != nil {
1433+ return nil, err
1434+ }
14101435 w := payloadWriter{}
14111436 w.putUint64(m.Epoch)
14121437 w.putUint64(m.Base)
@@ -1416,12 +1441,37 @@ func MarshalOutput(m Output) ([]byte, error) {
14161441 w.putUint16(uint16(m.Size.Cols))
14171442 w.putUint16(uint16(m.Size.Rows))
14181443 w.putBool(m.Full)
1419- w.putLongBytes(m.Data)
1444+ w.putUint8(kind)
1445+ w.putUint32(uint32(len(m.Data)))
1446+ w.putLongBytes(data)
14201447 return w.b, nil
14211448 }
14221449
1450+func compressOutputData(m Output) (byte, []byte, error) {
1451+ if !m.Full || len(m.Data) < outputCompressionThreshold {
1452+ return outputCompressionNone, m.Data, nil
1453+ }
1454+ var compressed bytes.Buffer
1455+ writer := outputCompressorPool.Get().(*zlib.Writer)
1456+ writer.Reset(&compressed)
1457+ defer func() {
1458+ writer.Reset(io.Discard)
1459+ outputCompressorPool.Put(writer)
1460+ }()
1461+ if _, err := writer.Write(m.Data); err != nil {
1462+ return 0, nil, err
1463+ }
1464+ if err := writer.Close(); err != nil {
1465+ return 0, nil, err
1466+ }
1467+ if compressed.Len()+outputCompressionHeaderLen >= len(m.Data) {
1468+ return outputCompressionNone, m.Data, nil
1469+ }
1470+ return outputCompressionZlib, compressed.Bytes(), nil
1471+}
1472+
14231473 // UnmarshalOutput strictly decodes one final Output payload. Header semantics
1424-// are validated before the length-prefixed data is copied.
1474+// are validated before its bounded decoded data allocation.
14251475 func UnmarshalOutput(b []byte) (Output, error) {
14261476 if len(b) > MaxFrameLen-1 {
14271477 return Output{}, ErrInvalidOutput
@@ -1462,15 +1512,16 @@ func UnmarshalOutput(b []byte) (Output, error) {
14621512 if err := ValidateOutput(m); err != nil {
14631513 return Output{}, err
14641514 }
1465- if len(r.b) < 4 {
1466- return Output{}, errShortPayload
1515+ kind, err := r.getUint8()
1516+ if err != nil {
1517+ return Output{}, err
14671518 }
1468- dataLen := binary.BigEndian.Uint32(r.b)
1469- if uint64(dataLen) > uint64(len(r.b)-4) {
1470- return Output{}, errShortPayload
1519+ decodedLen, err := r.getUint32()
1520+ if err != nil {
1521+ return Output{}, err
14711522 }
1472- if uint64(dataLen) < uint64(len(r.b)-4) {
1473- return Output{}, errTrailingBytes
1523+ if int64(decodedLen) > int64(MaxFrameLen-1-outputPayloadOverhead) {
1524+ return Output{}, ErrInvalidOutput
14741525 }
14751526 data, err := r.getLongBytes()
14761527 if err != nil {
@@ -1479,10 +1530,57 @@ func UnmarshalOutput(b []byte) (Output, error) {
14791530 if err := r.done(); err != nil {
14801531 return Output{}, err
14811532 }
1482- m.Data = data
1533+ switch kind {
1534+ case outputCompressionNone:
1535+ if uint32(len(data)) != decodedLen {
1536+ return Output{}, ErrInvalidOutput
1537+ }
1538+ m.Data = data
1539+ case outputCompressionZlib:
1540+ if !m.Full {
1541+ return Output{}, ErrInvalidOutput
1542+ }
1543+ m.Data, err = decompressOutputData(data, int(decodedLen))
1544+ if err != nil {
1545+ return Output{}, ErrInvalidOutput
1546+ }
1547+ default:
1548+ return Output{}, errInvalidEnum
1549+ }
1550+ if err := ValidateOutput(m); err != nil {
1551+ return Output{}, err
1552+ }
14831553 return m, nil
14841554 }
14851555
1556+func decompressOutputData(data []byte, decodedLen int) ([]byte, error) {
1557+ source := bytes.NewReader(data)
1558+ reader, err := zlib.NewReader(source)
1559+ if err != nil {
1560+ return nil, err
1561+ }
1562+ decoded := make([]byte, decodedLen)
1563+ if _, err := io.ReadFull(reader, decoded); err != nil {
1564+ _ = reader.Close()
1565+ return nil, err
1566+ }
1567+ var extra [1]byte
1568+ if n, err := reader.Read(extra[:]); n != 0 || err != io.EOF {
1569+ _ = reader.Close()
1570+ if err == nil {
1571+ return nil, errors.New("ports: compressed output exceeds declared length")
1572+ }
1573+ return nil, err
1574+ }
1575+ if err := reader.Close(); err != nil {
1576+ return nil, err
1577+ }
1578+ if source.Len() != 0 {
1579+ return nil, errors.New("ports: compressed output has trailing bytes")
1580+ }
1581+ return decoded, nil
1582+}
1583+
14861584 // MarshalAttachTarget encodes a strict server attach-target payload.
14871585 func MarshalAttachTarget(m AttachTarget) []byte {
14881586 if err := ValidateAttachTarget(m); err != nil {
@@ -7,10 +7,14 @@
7 package ports7 package ports
8 8
9 import (9 import (
10+ "bytes"
11+ "compress/zlib"
10 "encoding/binary"12 "encoding/binary"
11 "errors"13 "errors"
12 "fmt"14 "fmt"
15+ "io"
13 "math"16 "math"
17+ "sync"
14 18
15 renderer "github.com/bnema/vev-vt"19 renderer "github.com/bnema/vev-vt"
16 "github.com/bnema/vev/internal/domain"20 "github.com/bnema/vev/internal/domain"
@@ -29,10 +33,26 @@ var errInvalidBoolean = errors.New("ports: invalid boolean flag")
29 var errInvalidEnum = errors.New("ports: invalid enum value")33 var errInvalidEnum = errors.New("ports: invalid enum value")
30 34
31 const (35 const (
32- outputHeaderLen = 5*8 + 2*2 + 136+ outputHeaderLen = 5*8 + 2*2 + 1
33- outputPayloadOverhead = outputHeaderLen + 437+ outputCompressionHeaderLen = 1 + 4
38+ outputPayloadOverhead = outputHeaderLen + outputCompressionHeaderLen + 4
39+ // outputCompressionThreshold keeps small snapshots on the canonical path.
40+ outputCompressionThreshold = 1024
34 )41 )
35 42
43+const (
44+ outputCompressionNone byte = iota
45+ outputCompressionZlib
46+)
47+
48+var outputCompressorPool = sync.Pool{New: func() any {
49+ writer, err := zlib.NewWriterLevel(io.Discard, zlib.BestSpeed)
50+ if err != nil {
51+ panic(err)
52+ }
53+ return writer
54+}}
55+
36 // ErrInvalidSize reports a terminal size that cannot be represented by the56 // ErrInvalidSize reports a terminal size that cannot be represented by the
37 // protocol or would exceed the bounded screen area.57 // protocol or would exceed the bounded screen area.
38 var ErrInvalidSize = errors.New("ports: invalid terminal size")58 var ErrInvalidSize = errors.New("ports: invalid terminal size")
@@ -1402,11 +1422,16 @@ func UnmarshalErrorMsg(b []byte) (ErrorMsg, error) {
1402 }1422 }
1403 1423
1404 // MarshalOutput encodes m into the epoch/base/new/echo/viewRevision/1424 // MarshalOutput encodes m into the epoch/base/new/echo/viewRevision/
1405-// size/full/data message layout.1425+// size/full/compression/decoded-length/data message layout. Compression is
1426+// limited to large, full snapshots and is retained only when it saves bytes.
1406 func MarshalOutput(m Output) ([]byte, error) {1427 func MarshalOutput(m Output) ([]byte, error) {
1407 if err := ValidateOutput(m); err != nil {1428 if err := ValidateOutput(m); err != nil {
1408 return nil, err1429 return nil, err
1409 }1430 }
1431+ kind, data, err := compressOutputData(m)
1432+ if err != nil {
1433+ return nil, err
1434+ }
1410 w := payloadWriter{}1435 w := payloadWriter{}
1411 w.putUint64(m.Epoch)1436 w.putUint64(m.Epoch)
1412 w.putUint64(m.Base)1437 w.putUint64(m.Base)
@@ -1416,12 +1441,37 @@ func MarshalOutput(m Output) ([]byte, error) {
1416 w.putUint16(uint16(m.Size.Cols))1441 w.putUint16(uint16(m.Size.Cols))
1417 w.putUint16(uint16(m.Size.Rows))1442 w.putUint16(uint16(m.Size.Rows))
1418 w.putBool(m.Full)1443 w.putBool(m.Full)
1419- w.putLongBytes(m.Data)1444+ w.putUint8(kind)
1445+ w.putUint32(uint32(len(m.Data)))
1446+ w.putLongBytes(data)
1420 return w.b, nil1447 return w.b, nil
1421 }1448 }
1422 1449
1450+func compressOutputData(m Output) (byte, []byte, error) {
1451+ if !m.Full || len(m.Data) < outputCompressionThreshold {
1452+ return outputCompressionNone, m.Data, nil
1453+ }
1454+ var compressed bytes.Buffer
1455+ writer := outputCompressorPool.Get().(*zlib.Writer)
1456+ writer.Reset(&compressed)
1457+ defer func() {
1458+ writer.Reset(io.Discard)
1459+ outputCompressorPool.Put(writer)
1460+ }()
1461+ if _, err := writer.Write(m.Data); err != nil {
1462+ return 0, nil, err
1463+ }
1464+ if err := writer.Close(); err != nil {
1465+ return 0, nil, err
1466+ }
1467+ if compressed.Len()+outputCompressionHeaderLen >= len(m.Data) {
1468+ return outputCompressionNone, m.Data, nil
1469+ }
1470+ return outputCompressionZlib, compressed.Bytes(), nil
1471+}
1472+
1423 // UnmarshalOutput strictly decodes one final Output payload. Header semantics1473 // UnmarshalOutput strictly decodes one final Output payload. Header semantics
1424-// are validated before the length-prefixed data is copied.1474+// are validated before its bounded decoded data allocation.
1425 func UnmarshalOutput(b []byte) (Output, error) {1475 func UnmarshalOutput(b []byte) (Output, error) {
1426 if len(b) > MaxFrameLen-1 {1476 if len(b) > MaxFrameLen-1 {
1427 return Output{}, ErrInvalidOutput1477 return Output{}, ErrInvalidOutput
@@ -1462,15 +1512,16 @@ func UnmarshalOutput(b []byte) (Output, error) {
1462 if err := ValidateOutput(m); err != nil {1512 if err := ValidateOutput(m); err != nil {
1463 return Output{}, err1513 return Output{}, err
1464 }1514 }
1465- if len(r.b) < 4 {1515+ kind, err := r.getUint8()
1466- return Output{}, errShortPayload1516+ if err != nil {
1517+ return Output{}, err
1467 }1518 }
1468- dataLen := binary.BigEndian.Uint32(r.b)1519+ decodedLen, err := r.getUint32()
1469- if uint64(dataLen) > uint64(len(r.b)-4) {1520+ if err != nil {
1470- return Output{}, errShortPayload1521+ return Output{}, err
1471 }1522 }
1472- if uint64(dataLen) < uint64(len(r.b)-4) {1523+ if int64(decodedLen) > int64(MaxFrameLen-1-outputPayloadOverhead) {
1473- return Output{}, errTrailingBytes1524+ return Output{}, ErrInvalidOutput
1474 }1525 }
1475 data, err := r.getLongBytes()1526 data, err := r.getLongBytes()
1476 if err != nil {1527 if err != nil {
@@ -1479,10 +1530,57 @@ func UnmarshalOutput(b []byte) (Output, error) {
1479 if err := r.done(); err != nil {1530 if err := r.done(); err != nil {
1480 return Output{}, err1531 return Output{}, err
1481 }1532 }
1482- m.Data = data1533+ switch kind {
1534+ case outputCompressionNone:
1535+ if uint32(len(data)) != decodedLen {
1536+ return Output{}, ErrInvalidOutput
1537+ }
1538+ m.Data = data
1539+ case outputCompressionZlib:
1540+ if !m.Full {
1541+ return Output{}, ErrInvalidOutput
1542+ }
1543+ m.Data, err = decompressOutputData(data, int(decodedLen))
1544+ if err != nil {
1545+ return Output{}, ErrInvalidOutput
1546+ }
1547+ default:
1548+ return Output{}, errInvalidEnum
1549+ }
1550+ if err := ValidateOutput(m); err != nil {
1551+ return Output{}, err
1552+ }
1483 return m, nil1553 return m, nil
1484 }1554 }
1485 1555
1556+func decompressOutputData(data []byte, decodedLen int) ([]byte, error) {
1557+ source := bytes.NewReader(data)
1558+ reader, err := zlib.NewReader(source)
1559+ if err != nil {
1560+ return nil, err
1561+ }
1562+ decoded := make([]byte, decodedLen)
1563+ if _, err := io.ReadFull(reader, decoded); err != nil {
1564+ _ = reader.Close()
1565+ return nil, err
1566+ }
1567+ var extra [1]byte
1568+ if n, err := reader.Read(extra[:]); n != 0 || err != io.EOF {
1569+ _ = reader.Close()
1570+ if err == nil {
1571+ return nil, errors.New("ports: compressed output exceeds declared length")
1572+ }
1573+ return nil, err
1574+ }
1575+ if err := reader.Close(); err != nil {
1576+ return nil, err
1577+ }
1578+ if source.Len() != 0 {
1579+ return nil, errors.New("ports: compressed output has trailing bytes")
1580+ }
1581+ return decoded, nil
1582+}
1583+
1486 // MarshalAttachTarget encodes a strict server attach-target payload.1584 // MarshalAttachTarget encodes a strict server attach-target payload.
1487 func MarshalAttachTarget(m AttachTarget) []byte {1585 func MarshalAttachTarget(m AttachTarget) []byte {
1488 if err := ValidateAttachTarget(m); err != nil {1586 if err := ValidateAttachTarget(m); err != nil {
M internal/ports/wire_final_test.go +5 -4
@@ -11,8 +11,8 @@ import (
1111 )
1212
1313 func TestFinalProtocolVersionAndNoV21Hello(t *testing.T) {
14- if ProtocolVersion != 32 {
15- t.Fatalf("ProtocolVersion = %d, want 32", ProtocolVersion)
14+ if ProtocolVersion != 33 {
15+ t.Fatalf("ProtocolVersion = %d, want 33", ProtocolVersion)
1616 }
1717 payload := MarshalHello(Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}})
1818 if len(payload) < 2 {
@@ -26,7 +26,7 @@ func TestFinalProtocolVersionAndNoV21Hello(t *testing.T) {
2626
2727 func TestFinalHelloGoldenStrict(t *testing.T) {
2828 msg := Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}}
29- want := append([]byte{0, 32, 2}, make([]byte, 16+8)...)
29+ want := append([]byte{0, 33, 2}, make([]byte, 16+8)...)
3030 want = append(want, 0, 0, 0, 80, 0, 24, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
3131 got := MarshalHello(msg)
3232 if !bytes.Equal(got, want) {
@@ -99,6 +99,7 @@ func TestFinalOutputGoldenStrict(t *testing.T) {
9999 0, 0, 0, 0, 0, 0, 0, 8,
100100 0, 0, 0, 0, 0, 0, 0, 9,
101101 0, 80, 0, 24, 1,
102+ 0, 0, 0, 0, 2,
102103 0, 0, 0, 2, 'o', 'k',
103104 }
104105 got, err := MarshalOutput(msg)
@@ -140,7 +141,7 @@ func TestFinalOutputSemanticValidationBeforeDataAllocation(t *testing.T) {
140141 {name: "size zero", mutate: func(b []byte) { binary.BigEndian.PutUint16(b[40:42], 0) }},
141142 {name: "full flag false for reset", mutate: func(b []byte) { b[44] = 0 }},
142143 {name: "invalid bool", mutate: func(b []byte) { b[44] = 2 }},
143- {name: "impossible data length", mutate: func(b []byte) { binary.BigEndian.PutUint32(b[45:49], ^uint32(0)) }},
144+ {name: "impossible data length", mutate: func(b []byte) { binary.BigEndian.PutUint32(b[50:54], ^uint32(0)) }},
144145 } {
145146 t.Run(tt.name, func(t *testing.T) {
146147 bad := append([]byte(nil), payload...)
@@ -11,8 +11,8 @@ import (
11 )11 )
12 12
13 func TestFinalProtocolVersionAndNoV21Hello(t *testing.T) {13 func TestFinalProtocolVersionAndNoV21Hello(t *testing.T) {
14- if ProtocolVersion != 32 {14+ if ProtocolVersion != 33 {
15- t.Fatalf("ProtocolVersion = %d, want 32", ProtocolVersion)15+ t.Fatalf("ProtocolVersion = %d, want 33", ProtocolVersion)
16 }16 }
17 payload := MarshalHello(Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}})17 payload := MarshalHello(Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}})
18 if len(payload) < 2 {18 if len(payload) < 2 {
@@ -26,7 +26,7 @@ func TestFinalProtocolVersionAndNoV21Hello(t *testing.T) {
26 26
27 func TestFinalHelloGoldenStrict(t *testing.T) {27 func TestFinalHelloGoldenStrict(t *testing.T) {
28 msg := Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}}28 msg := Hello{Version: ProtocolVersion, Intent: IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}}
29- want := append([]byte{0, 32, 2}, make([]byte, 16+8)...)29+ want := append([]byte{0, 33, 2}, make([]byte, 16+8)...)
30 want = append(want, 0, 0, 0, 80, 0, 24, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)30 want = append(want, 0, 0, 0, 80, 0, 24, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
31 got := MarshalHello(msg)31 got := MarshalHello(msg)
32 if !bytes.Equal(got, want) {32 if !bytes.Equal(got, want) {
@@ -99,6 +99,7 @@ func TestFinalOutputGoldenStrict(t *testing.T) {
99 0, 0, 0, 0, 0, 0, 0, 8,99 0, 0, 0, 0, 0, 0, 0, 8,
100 0, 0, 0, 0, 0, 0, 0, 9,100 0, 0, 0, 0, 0, 0, 0, 9,
101 0, 80, 0, 24, 1,101 0, 80, 0, 24, 1,
102+ 0, 0, 0, 0, 2,
102 0, 0, 0, 2, 'o', 'k',103 0, 0, 0, 2, 'o', 'k',
103 }104 }
104 got, err := MarshalOutput(msg)105 got, err := MarshalOutput(msg)
@@ -140,7 +141,7 @@ func TestFinalOutputSemanticValidationBeforeDataAllocation(t *testing.T) {
140 {name: "size zero", mutate: func(b []byte) { binary.BigEndian.PutUint16(b[40:42], 0) }},141 {name: "size zero", mutate: func(b []byte) { binary.BigEndian.PutUint16(b[40:42], 0) }},
141 {name: "full flag false for reset", mutate: func(b []byte) { b[44] = 0 }},142 {name: "full flag false for reset", mutate: func(b []byte) { b[44] = 0 }},
142 {name: "invalid bool", mutate: func(b []byte) { b[44] = 2 }},143 {name: "invalid bool", mutate: func(b []byte) { b[44] = 2 }},
143- {name: "impossible data length", mutate: func(b []byte) { binary.BigEndian.PutUint32(b[45:49], ^uint32(0)) }},144+ {name: "impossible data length", mutate: func(b []byte) { binary.BigEndian.PutUint32(b[50:54], ^uint32(0)) }},
144 } {145 } {
145 t.Run(tt.name, func(t *testing.T) {146 t.Run(tt.name, func(t *testing.T) {
146 bad := append([]byte(nil), payload...)147 bad := append([]byte(nil), payload...)
M internal/ports/wire_test.go +7 -7
@@ -131,7 +131,7 @@ func TestHelloEnvironmentCodec(t *testing.T) {
131131 }
132132 got := MarshalHello(msg)
133133 want := []byte{
134- 0x00, 0x20, 0x00, // version, intent, first client ID byte
134+ 0x00, 0x21, 0x00, // version, intent, first client ID byte
135135 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
136136 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // client ID
137137 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // resume token
@@ -361,7 +361,7 @@ func TestThemeGenerationClearedWireGoldenPreservesProtocolVersion(t *testing.T)
361361 }
362362 want := append([]byte{0x0f, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x00, 0x00}, make([]byte, 48)...)
363363 require.Equal(t, want, MarshalTheme(cleared))
364- require.Equal(t, uint16(32), ProtocolVersion)
364+ require.Equal(t, uint16(33), ProtocolVersion)
365365 }
366366
367367 func TestResizeGoldenAndRoundTrip(t *testing.T) {
@@ -522,7 +522,7 @@ func TestCommandRequestGoldenAndRoundTrip(t *testing.T) {
522522 name: "minimal",
523523 msg: CommandRequest{Version: ProtocolVersion, Slug: "split-right"},
524524 want: []byte{
525- 0x00, 0x20,
525+ 0x00, 0x21,
526526 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // request ID
527527 0x00, // attached
528528 0x00, // self
@@ -545,7 +545,7 @@ func TestCommandRequestGoldenAndRoundTrip(t *testing.T) {
545545 JSON: true,
546546 },
547547 want: []byte{
548- 0x00, 0x20,
548+ 0x00, 0x21,
549549 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // request ID
550550 0x00, // attached
551551 0x01, // self
@@ -671,7 +671,7 @@ func TestErrorMsgGoldenAndRoundTrip(t *testing.T) {
671671
672672 func TestOutputGoldenAndRoundTrip(t *testing.T) {
673673 msg := Output{Epoch: 1, Base: 0, New: 2, Echo: 3, Size: domain.Size{Cols: 1, Rows: 1}, Full: true, Data: []byte("hello\n")}
674- want := []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x01, 0x01, 0x00, 0x00, 0x00, 0x06, 0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x0a}
674+ want := []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x01, 0x01, 0x00, 0x00, 0x00, 0x00, 0x06, 0x00, 0x00, 0x00, 0x06, 0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x0a}
675675
676676 got, err := MarshalOutput(msg)
677677 if err != nil {
@@ -974,7 +974,7 @@ func TestCommandCorrelationGoldenAndStrict(t *testing.T) {
974974 name: "attached",
975975 msg: CommandRequest{Version: ProtocolVersion, RequestID: 0x0102030405060708, Attached: true, Self: true, Slug: "split-right", Args: []string{"--vertical"}},
976976 want: []byte{
977- 0x00, 0x20, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x01, 0x01,
977+ 0x00, 0x21, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x01, 0x01,
978978 0x00, 0x0b, 's', 'p', 'l', 'i', 't', '-', 'r', 'i', 'g', 'h', 't',
979979 0x00, 0x01, 0x00, 0x00, 0x00, 0x0a, '-', '-', 'v', 'e', 'r', 't', 'i', 'c', 'a', 'l',
980980 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
@@ -984,7 +984,7 @@ func TestCommandCorrelationGoldenAndStrict(t *testing.T) {
984984 name: "control",
985985 msg: CommandRequest{Version: ProtocolVersion, Slug: "ls"},
986986 want: []byte{
987- 0x00, 0x20, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
987+ 0x00, 0x21, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
988988 0x00, 0x02, 'l', 's', 0x00, 0x00,
989989 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
990990 },
@@ -131,7 +131,7 @@ func TestHelloEnvironmentCodec(t *testing.T) {
131 }131 }
132 got := MarshalHello(msg)132 got := MarshalHello(msg)
133 want := []byte{133 want := []byte{
134- 0x00, 0x20, 0x00, // version, intent, first client ID byte134+ 0x00, 0x21, 0x00, // version, intent, first client ID byte
135 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,135 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
136 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // client ID136 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // client ID
137 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // resume token137 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // resume token
@@ -361,7 +361,7 @@ func TestThemeGenerationClearedWireGoldenPreservesProtocolVersion(t *testing.T)
361 }361 }
362 want := append([]byte{0x0f, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x00, 0x00}, make([]byte, 48)...)362 want := append([]byte{0x0f, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x00, 0x00}, make([]byte, 48)...)
363 require.Equal(t, want, MarshalTheme(cleared))363 require.Equal(t, want, MarshalTheme(cleared))
364- require.Equal(t, uint16(32), ProtocolVersion)364+ require.Equal(t, uint16(33), ProtocolVersion)
365 }365 }
366 366
367 func TestResizeGoldenAndRoundTrip(t *testing.T) {367 func TestResizeGoldenAndRoundTrip(t *testing.T) {
@@ -522,7 +522,7 @@ func TestCommandRequestGoldenAndRoundTrip(t *testing.T) {
522 name: "minimal",522 name: "minimal",
523 msg: CommandRequest{Version: ProtocolVersion, Slug: "split-right"},523 msg: CommandRequest{Version: ProtocolVersion, Slug: "split-right"},
524 want: []byte{524 want: []byte{
525- 0x00, 0x20,525+ 0x00, 0x21,
526 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // request ID526 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // request ID
527 0x00, // attached527 0x00, // attached
528 0x00, // self528 0x00, // self
@@ -545,7 +545,7 @@ func TestCommandRequestGoldenAndRoundTrip(t *testing.T) {
545 JSON: true,545 JSON: true,
546 },546 },
547 want: []byte{547 want: []byte{
548- 0x00, 0x20,548+ 0x00, 0x21,
549 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // request ID549 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // request ID
550 0x00, // attached550 0x00, // attached
551 0x01, // self551 0x01, // self
@@ -671,7 +671,7 @@ func TestErrorMsgGoldenAndRoundTrip(t *testing.T) {
671 671
672 func TestOutputGoldenAndRoundTrip(t *testing.T) {672 func TestOutputGoldenAndRoundTrip(t *testing.T) {
673 msg := Output{Epoch: 1, Base: 0, New: 2, Echo: 3, Size: domain.Size{Cols: 1, Rows: 1}, Full: true, Data: []byte("hello\n")}673 msg := Output{Epoch: 1, Base: 0, New: 2, Echo: 3, Size: domain.Size{Cols: 1, Rows: 1}, Full: true, Data: []byte("hello\n")}
674- want := []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x01, 0x01, 0x00, 0x00, 0x00, 0x06, 0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x0a}674+ want := []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x01, 0x01, 0x00, 0x00, 0x00, 0x00, 0x06, 0x00, 0x00, 0x00, 0x06, 0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x0a}
675 675
676 got, err := MarshalOutput(msg)676 got, err := MarshalOutput(msg)
677 if err != nil {677 if err != nil {
@@ -974,7 +974,7 @@ func TestCommandCorrelationGoldenAndStrict(t *testing.T) {
974 name: "attached",974 name: "attached",
975 msg: CommandRequest{Version: ProtocolVersion, RequestID: 0x0102030405060708, Attached: true, Self: true, Slug: "split-right", Args: []string{"--vertical"}},975 msg: CommandRequest{Version: ProtocolVersion, RequestID: 0x0102030405060708, Attached: true, Self: true, Slug: "split-right", Args: []string{"--vertical"}},
976 want: []byte{976 want: []byte{
977- 0x00, 0x20, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x01, 0x01,977+ 0x00, 0x21, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x01, 0x01,
978 0x00, 0x0b, 's', 'p', 'l', 'i', 't', '-', 'r', 'i', 'g', 'h', 't',978 0x00, 0x0b, 's', 'p', 'l', 'i', 't', '-', 'r', 'i', 'g', 'h', 't',
979 0x00, 0x01, 0x00, 0x00, 0x00, 0x0a, '-', '-', 'v', 'e', 'r', 't', 'i', 'c', 'a', 'l',979 0x00, 0x01, 0x00, 0x00, 0x00, 0x0a, '-', '-', 'v', 'e', 'r', 't', 'i', 'c', 'a', 'l',
980 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,980 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
@@ -984,7 +984,7 @@ func TestCommandCorrelationGoldenAndStrict(t *testing.T) {
984 name: "control",984 name: "control",
985 msg: CommandRequest{Version: ProtocolVersion, Slug: "ls"},985 msg: CommandRequest{Version: ProtocolVersion, Slug: "ls"},
986 want: []byte{986 want: []byte{
987- 0x00, 0x20, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,987+ 0x00, 0x21, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
988 0x00, 0x02, 'l', 's', 0x00, 0x00,988 0x00, 0x02, 'l', 's', 0x00, 0x00,
989 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,989 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
990 },990 },
M internal/usecase/client/ack_queue_test.go +1 -1
@@ -31,7 +31,7 @@ func TestCumulativeAckBypassesFullNormalSendQueue(t *testing.T) {
3131 defer cancel()
3232 tr := &ackRecordingTransport{sent: make(chan ports.Frame, sendQueueDepth+1)}
3333 errCh := make(chan error, 1)
34- go runSender(ctx, cancel, tr, normal, acks, errCh, slog.Default())
34+ go runSender(ctx, cancel, tr, normal, nil, acks, errCh, slog.Default())
3535
3636 first := <-tr.sent
3737 require.Equal(t, ports.MsgAck, first.Type)
@@ -31,7 +31,7 @@ func TestCumulativeAckBypassesFullNormalSendQueue(t *testing.T) {
31 defer cancel()31 defer cancel()
32 tr := &ackRecordingTransport{sent: make(chan ports.Frame, sendQueueDepth+1)}32 tr := &ackRecordingTransport{sent: make(chan ports.Frame, sendQueueDepth+1)}
33 errCh := make(chan error, 1)33 errCh := make(chan error, 1)
34- go runSender(ctx, cancel, tr, normal, acks, errCh, slog.Default())34+ go runSender(ctx, cancel, tr, normal, nil, acks, errCh, slog.Default())
35 35
36 first := <-tr.sent36 first := <-tr.sent
37 require.Equal(t, ports.MsgAck, first.Type)37 require.Equal(t, ports.MsgAck, first.Type)
M internal/usecase/client/client.go +47 -2
@@ -1197,6 +1197,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
11971197 defer revokeInputClaim()
11981198
11991199 sendCh := make(chan ports.Frame, sendQueueDepth)
1200+ inputGate := newSamePeerInputGate()
12001201 ackQueue := newCumulativeAckQueue()
12011202 sendErrCh := make(chan error, 1)
12021203 senderStarted := false
@@ -1307,6 +1308,8 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
13071308 awaitingReconnectReset := false
13081309 outputState := outputApplyState{}
13091310 outputResetRequested := false
1311+ var samePeerSwitch *samePeerSwitchPending
1312+ var nextSamePeerRequestID uint64
13101313 if reconnect.showing {
13111314 if reconnect.remote {
13121315 // Before the asynchronous sender starts, preserve transport ordering by
@@ -1335,7 +1338,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
13351338 // or unchanged screen may legitimately produce no initial Output frame.
13361339 endHandshake()
13371340 senderStarted = true
1338- go runSender(loopCtx, cancel, transport, sendCh, ackQueue, sendErrCh, log)
1341+ go runSender(loopCtx, cancel, transport, sendCh, inputGate, ackQueue, sendErrCh, log)
13391342 stdinDone := make(chan struct{})
13401343 go func() {
13411344 defer close(stdinDone)
@@ -1569,6 +1572,29 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
15691572 if derr != nil {
15701573 return welcomedResult(fmt.Errorf("vev: decoding attach target: %w", derr))
15711574 }
1575+ if target.Endpoint == "" && target.RemoteTarget == nil && target.ExactTarget != nil && a.runner.ledger != nil {
1576+ if samePeerSwitch != nil {
1577+ return welcomedResult(errors.New("vev: concurrent same-peer switch offer"))
1578+ }
1579+ nextSamePeerRequestID++
1580+ request := a.runner.ledger.samePeerHandoff(request, target)
1581+ payload, err := ports.MarshalSamePeerSwitchRequest(ports.SamePeerSwitchRequest{
1582+ RequestID: nextSamePeerRequestID,
1583+ Target: *target.ExactTarget,
1584+ PreferredTabID: request.PreferredTabID,
1585+ })
1586+ if err != nil {
1587+ return welcomedResult(fmt.Errorf("vev: encoding same-peer switch: %w", err))
1588+ }
1589+ samePeerSwitch = &samePeerSwitchPending{requestID: nextSamePeerRequestID, target: *target.ExactTarget}
1590+ inputGate.setPaused(true)
1591+ select {
1592+ case sendCh <- ports.Frame{Type: ports.MsgSamePeerSwitchRequest, Payload: payload}:
1593+ continue
1594+ case <-loopCtx.Done():
1595+ return loopCanceledResult()
1596+ }
1597+ }
15721598 handoff := welcomedResult(nil)
15731599 handoff.target = &target
15741600 return handoff
@@ -1609,6 +1635,9 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
16091635 if derr != nil {
16101636 return welcomedResult(fmt.Errorf("vev: decoding committed route identity: %w", derr))
16111637 }
1638+ if samePeerSwitch != nil && identity.Target != samePeerSwitch.target {
1639+ return welcomedResult(errors.New("vev: same-peer switch committed an unexpected target"))
1640+ }
16121641 name = identity.Target.SessionName
16131642 committedIdentity = cloneCommittedIdentity(&identity)
16141643 if a.runner.ledger == nil {
@@ -1620,6 +1649,19 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
16201649 if derr := publishRouteSnapshot(); derr != nil {
16211650 return welcomedResult(fmt.Errorf("vev: publishing daemon-local route snapshot: %w", derr))
16221651 }
1652+ if samePeerSwitch != nil {
1653+ samePeerSwitch = nil
1654+ inputGate.setPaused(false)
1655+ }
1656+ case ports.MsgSamePeerSwitchFailure:
1657+ failure, derr := ports.UnmarshalSamePeerSwitchFailure(r.frame.Payload)
1658+ if derr != nil {
1659+ return welcomedResult(fmt.Errorf("vev: decoding same-peer switch failure: %w", derr))
1660+ }
1661+ if samePeerSwitch != nil && failure.RequestID == samePeerSwitch.requestID {
1662+ samePeerSwitch = nil
1663+ inputGate.setPaused(false)
1664+ }
16231665 case ports.MsgRoutePosition:
16241666 position, derr := ports.UnmarshalRoutePosition(r.frame.Payload)
16251667 if derr != nil {
@@ -1721,7 +1763,7 @@ func (q *cumulativeAckQueue) take() (uint64, uint64) {
17211763 return epoch, state
17221764 }
17231765
1724-func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.Transport, in <-chan ports.Frame, acks *cumulativeAckQueue, errCh chan<- error, log *slog.Logger) {
1766+func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.Transport, in <-chan ports.Frame, inputGate *samePeerInputGate, acks *cumulativeAckQueue, errCh chan<- error, log *slog.Logger) {
17251767 defer log.Debug("sender pump exited")
17261768 send := func(f ports.Frame) bool {
17271769 if err := transport.Send(f); err != nil {
@@ -1770,6 +1812,9 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.T
17701812 return
17711813 default:
17721814 }
1815+ if (f.Type == ports.MsgInput || f.Type == ports.MsgImagePush) && !inputGate.wait(ctx) {
1816+ return
1817+ }
17731818 if !send(f) {
17741819 return
17751820 }
@@ -1197,6 +1197,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
1197 defer revokeInputClaim()1197 defer revokeInputClaim()
1198 1198
1199 sendCh := make(chan ports.Frame, sendQueueDepth)1199 sendCh := make(chan ports.Frame, sendQueueDepth)
1200+ inputGate := newSamePeerInputGate()
1200 ackQueue := newCumulativeAckQueue()1201 ackQueue := newCumulativeAckQueue()
1201 sendErrCh := make(chan error, 1)1202 sendErrCh := make(chan error, 1)
1202 senderStarted := false1203 senderStarted := false
@@ -1307,6 +1308,8 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
1307 awaitingReconnectReset := false1308 awaitingReconnectReset := false
1308 outputState := outputApplyState{}1309 outputState := outputApplyState{}
1309 outputResetRequested := false1310 outputResetRequested := false
1311+ var samePeerSwitch *samePeerSwitchPending
1312+ var nextSamePeerRequestID uint64
1310 if reconnect.showing {1313 if reconnect.showing {
1311 if reconnect.remote {1314 if reconnect.remote {
1312 // Before the asynchronous sender starts, preserve transport ordering by1315 // Before the asynchronous sender starts, preserve transport ordering by
@@ -1335,7 +1338,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
1335 // or unchanged screen may legitimately produce no initial Output frame.1338 // or unchanged screen may legitimately produce no initial Output frame.
1336 endHandshake()1339 endHandshake()
1337 senderStarted = true1340 senderStarted = true
1338- go runSender(loopCtx, cancel, transport, sendCh, ackQueue, sendErrCh, log)1341+ go runSender(loopCtx, cancel, transport, sendCh, inputGate, ackQueue, sendErrCh, log)
1339 stdinDone := make(chan struct{})1342 stdinDone := make(chan struct{})
1340 go func() {1343 go func() {
1341 defer close(stdinDone)1344 defer close(stdinDone)
@@ -1569,6 +1572,29 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
1569 if derr != nil {1572 if derr != nil {
1570 return welcomedResult(fmt.Errorf("vev: decoding attach target: %w", derr))1573 return welcomedResult(fmt.Errorf("vev: decoding attach target: %w", derr))
1571 }1574 }
1575+ if target.Endpoint == "" && target.RemoteTarget == nil && target.ExactTarget != nil && a.runner.ledger != nil {
1576+ if samePeerSwitch != nil {
1577+ return welcomedResult(errors.New("vev: concurrent same-peer switch offer"))
1578+ }
1579+ nextSamePeerRequestID++
1580+ request := a.runner.ledger.samePeerHandoff(request, target)
1581+ payload, err := ports.MarshalSamePeerSwitchRequest(ports.SamePeerSwitchRequest{
1582+ RequestID: nextSamePeerRequestID,
1583+ Target: *target.ExactTarget,
1584+ PreferredTabID: request.PreferredTabID,
1585+ })
1586+ if err != nil {
1587+ return welcomedResult(fmt.Errorf("vev: encoding same-peer switch: %w", err))
1588+ }
1589+ samePeerSwitch = &samePeerSwitchPending{requestID: nextSamePeerRequestID, target: *target.ExactTarget}
1590+ inputGate.setPaused(true)
1591+ select {
1592+ case sendCh <- ports.Frame{Type: ports.MsgSamePeerSwitchRequest, Payload: payload}:
1593+ continue
1594+ case <-loopCtx.Done():
1595+ return loopCanceledResult()
1596+ }
1597+ }
1572 handoff := welcomedResult(nil)1598 handoff := welcomedResult(nil)
1573 handoff.target = &target1599 handoff.target = &target
1574 return handoff1600 return handoff
@@ -1609,6 +1635,9 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
1609 if derr != nil {1635 if derr != nil {
1610 return welcomedResult(fmt.Errorf("vev: decoding committed route identity: %w", derr))1636 return welcomedResult(fmt.Errorf("vev: decoding committed route identity: %w", derr))
1611 }1637 }
1638+ if samePeerSwitch != nil && identity.Target != samePeerSwitch.target {
1639+ return welcomedResult(errors.New("vev: same-peer switch committed an unexpected target"))
1640+ }
1612 name = identity.Target.SessionName1641 name = identity.Target.SessionName
1613 committedIdentity = cloneCommittedIdentity(&identity)1642 committedIdentity = cloneCommittedIdentity(&identity)
1614 if a.runner.ledger == nil {1643 if a.runner.ledger == nil {
@@ -1620,6 +1649,19 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
1620 if derr := publishRouteSnapshot(); derr != nil {1649 if derr := publishRouteSnapshot(); derr != nil {
1621 return welcomedResult(fmt.Errorf("vev: publishing daemon-local route snapshot: %w", derr))1650 return welcomedResult(fmt.Errorf("vev: publishing daemon-local route snapshot: %w", derr))
1622 }1651 }
1652+ if samePeerSwitch != nil {
1653+ samePeerSwitch = nil
1654+ inputGate.setPaused(false)
1655+ }
1656+ case ports.MsgSamePeerSwitchFailure:
1657+ failure, derr := ports.UnmarshalSamePeerSwitchFailure(r.frame.Payload)
1658+ if derr != nil {
1659+ return welcomedResult(fmt.Errorf("vev: decoding same-peer switch failure: %w", derr))
1660+ }
1661+ if samePeerSwitch != nil && failure.RequestID == samePeerSwitch.requestID {
1662+ samePeerSwitch = nil
1663+ inputGate.setPaused(false)
1664+ }
1623 case ports.MsgRoutePosition:1665 case ports.MsgRoutePosition:
1624 position, derr := ports.UnmarshalRoutePosition(r.frame.Payload)1666 position, derr := ports.UnmarshalRoutePosition(r.frame.Payload)
1625 if derr != nil {1667 if derr != nil {
@@ -1721,7 +1763,7 @@ func (q *cumulativeAckQueue) take() (uint64, uint64) {
1721 return epoch, state1763 return epoch, state
1722 }1764 }
1723 1765
1724-func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.Transport, in <-chan ports.Frame, acks *cumulativeAckQueue, errCh chan<- error, log *slog.Logger) {1766+func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.Transport, in <-chan ports.Frame, inputGate *samePeerInputGate, acks *cumulativeAckQueue, errCh chan<- error, log *slog.Logger) {
1725 defer log.Debug("sender pump exited")1767 defer log.Debug("sender pump exited")
1726 send := func(f ports.Frame) bool {1768 send := func(f ports.Frame) bool {
1727 if err := transport.Send(f); err != nil {1769 if err := transport.Send(f); err != nil {
@@ -1770,6 +1812,9 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.T
1770 return1812 return
1771 default:1813 default:
1772 }1814 }
1815+ if (f.Type == ports.MsgInput || f.Type == ports.MsgImagePush) && !inputGate.wait(ctx) {
1816+ return
1817+ }
1773 if !send(f) {1818 if !send(f) {
1774 return1819 return
1775 }1820 }
M internal/usecase/client/client_test.go +3 -8
@@ -484,12 +484,10 @@ func TestLocalOnlyHandoffBetweenLocalSessionsKeepsClientRunning(t *testing.T) {
484484 Session: "second", Intent: ports.IntentAttach, ExactTarget: &secondTarget,
485485 EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
486486 }))},
487- }}
488- second := &recordingTransport{recvs: []recvItem{
489- {f: welcome("second", secondLifecycle)},
487+ {f: frameOf(ports.MsgCommittedRouteIdentity, mustMarshalCommittedIdentity(ports.CommittedRouteIdentity{Target: secondTarget}))},
490488 {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
491489 }}
492- dialer := &sequenceDialer{trs: []ports.Transport{first, second}}
490+ dialer := &sequenceDialer{trs: []ports.Transport{first}}
493491
494492 err := runTestClient(context.Background(), testDependencies(dialer, term, realClock{}, nil, nil), client.AttachRequest{
495493 Intent: ports.IntentAttach, SessionName: "first",
@@ -498,10 +496,7 @@ func TestLocalOnlyHandoffBetweenLocalSessionsKeepsClientRunning(t *testing.T) {
498496 })
499497
500498 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)
499+ require.Equal(t, int32(1), dialer.calls.Load(), "same-peer switching must not dial a replacement transport")
505500 }
506501
507502 func TestAttachHelloPreservesCompleteAttachRequest(t *testing.T) {
@@ -484,12 +484,10 @@ func TestLocalOnlyHandoffBetweenLocalSessionsKeepsClientRunning(t *testing.T) {
484 Session: "second", Intent: ports.IntentAttach, ExactTarget: &secondTarget,484 Session: "second", Intent: ports.IntentAttach, ExactTarget: &secondTarget,
485 EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,485 EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
486 }))},486 }))},
487- }}487+ {f: frameOf(ports.MsgCommittedRouteIdentity, mustMarshalCommittedIdentity(ports.CommittedRouteIdentity{Target: secondTarget}))},
488- second := &recordingTransport{recvs: []recvItem{
489- {f: welcome("second", secondLifecycle)},
490 {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},488 {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
491 }}489 }}
492- dialer := &sequenceDialer{trs: []ports.Transport{first, second}}490+ dialer := &sequenceDialer{trs: []ports.Transport{first}}
493 491
494 err := runTestClient(context.Background(), testDependencies(dialer, term, realClock{}, nil, nil), client.AttachRequest{492 err := runTestClient(context.Background(), testDependencies(dialer, term, realClock{}, nil, nil), client.AttachRequest{
495 Intent: ports.IntentAttach, SessionName: "first",493 Intent: ports.IntentAttach, SessionName: "first",
@@ -498,10 +496,7 @@ func TestLocalOnlyHandoffBetweenLocalSessionsKeepsClientRunning(t *testing.T) {
498 })496 })
499 497
500 require.NoError(t, err)498 require.NoError(t, err)
501- hello := helloFromSend(t, second)499+ require.Equal(t, int32(1), dialer.calls.Load(), "same-peer switching must not dial a replacement transport")
502- require.Equal(t, "second", hello.Name)
503- require.Equal(t, ports.EnvironmentPolicyClientOwned, hello.EnvironmentPolicy)
504- require.Nil(t, hello.RemoteTarget)
505 }500 }
506 501
507 func TestAttachHelloPreservesCompleteAttachRequest(t *testing.T) {502 func TestAttachHelloPreservesCompleteAttachRequest(t *testing.T) {
A internal/usecase/client/same_peer_switch.go +58 -0
@@ -0,0 +1,58 @@
1+package client
2+
3+import (
4+ "context"
5+ "sync"
6+
7+ "github.com/bnema/vev/internal/ports"
8+)
9+
10+// samePeerInputGate holds raw input behind a requested in-band switch. It does
11+// not drop or reparse bytes: once the daemon commits, queued input follows the
12+// attachment to the target; a typed pre-commit failure releases it to source.
13+type samePeerInputGate struct {
14+ mu sync.Mutex
15+ paused bool
16+ changed chan struct{}
17+}
18+
19+type samePeerSwitchPending struct {
20+ requestID uint64
21+ target ports.ExactSessionTarget
22+}
23+
24+func newSamePeerInputGate() *samePeerInputGate {
25+ return &samePeerInputGate{changed: make(chan struct{})}
26+}
27+
28+func (g *samePeerInputGate) setPaused(paused bool) {
29+ if g == nil {
30+ return
31+ }
32+ g.mu.Lock()
33+ if g.paused != paused {
34+ g.paused = paused
35+ close(g.changed)
36+ g.changed = make(chan struct{})
37+ }
38+ g.mu.Unlock()
39+}
40+
41+func (g *samePeerInputGate) wait(ctx context.Context) bool {
42+ if g == nil {
43+ return true
44+ }
45+ for {
46+ g.mu.Lock()
47+ paused, changed := g.paused, g.changed
48+ g.mu.Unlock()
49+ if !paused {
50+ return true
51+ }
52+ select {
53+ case <-ctx.Done():
54+ return false
55+ case <-changed:
56+ }
57+ }
58+}
@@ -0,0 +1,58 @@
1+package client
2+
3+import (
4+ "context"
5+ "sync"
6+
7+ "github.com/bnema/vev/internal/ports"
8+)
9+
10+// samePeerInputGate holds raw input behind a requested in-band switch. It does
11+// not drop or reparse bytes: once the daemon commits, queued input follows the
12+// attachment to the target; a typed pre-commit failure releases it to source.
13+type samePeerInputGate struct {
14+ mu sync.Mutex
15+ paused bool
16+ changed chan struct{}
17+}
18+
19+type samePeerSwitchPending struct {
20+ requestID uint64
21+ target ports.ExactSessionTarget
22+}
23+
24+func newSamePeerInputGate() *samePeerInputGate {
25+ return &samePeerInputGate{changed: make(chan struct{})}
26+}
27+
28+func (g *samePeerInputGate) setPaused(paused bool) {
29+ if g == nil {
30+ return
31+ }
32+ g.mu.Lock()
33+ if g.paused != paused {
34+ g.paused = paused
35+ close(g.changed)
36+ g.changed = make(chan struct{})
37+ }
38+ g.mu.Unlock()
39+}
40+
41+func (g *samePeerInputGate) wait(ctx context.Context) bool {
42+ if g == nil {
43+ return true
44+ }
45+ for {
46+ g.mu.Lock()
47+ paused, changed := g.paused, g.changed
48+ g.mu.Unlock()
49+ if !paused {
50+ return true
51+ }
52+ select {
53+ case <-ctx.Done():
54+ return false
55+ case <-changed:
56+ }
57+ }
58+}
A internal/usecase/client/same_peer_switch_test.go +29 -0
@@ -0,0 +1,29 @@
1+package client
2+
3+import (
4+ "context"
5+ "testing"
6+)
7+
8+func TestSamePeerInputGateBlocksUntilReleased(t *testing.T) {
9+ gate := newSamePeerInputGate()
10+ gate.setPaused(true)
11+ ctx, cancel := context.WithCancel(context.Background())
12+ defer cancel()
13+ ready := make(chan bool, 1)
14+ go func() { ready <- gate.wait(ctx) }()
15+ gate.setPaused(false)
16+ if ok := <-ready; !ok {
17+ t.Fatal("input gate did not release")
18+ }
19+}
20+
21+func TestSamePeerInputGateCancels(t *testing.T) {
22+ gate := newSamePeerInputGate()
23+ gate.setPaused(true)
24+ ctx, cancel := context.WithCancel(context.Background())
25+ cancel()
26+ if gate.wait(ctx) {
27+ t.Fatal("cancelled input gate wait succeeded")
28+ }
29+}
@@ -0,0 +1,29 @@
1+package client
2+
3+import (
4+ "context"
5+ "testing"
6+)
7+
8+func TestSamePeerInputGateBlocksUntilReleased(t *testing.T) {
9+ gate := newSamePeerInputGate()
10+ gate.setPaused(true)
11+ ctx, cancel := context.WithCancel(context.Background())
12+ defer cancel()
13+ ready := make(chan bool, 1)
14+ go func() { ready <- gate.wait(ctx) }()
15+ gate.setPaused(false)
16+ if ok := <-ready; !ok {
17+ t.Fatal("input gate did not release")
18+ }
19+}
20+
21+func TestSamePeerInputGateCancels(t *testing.T) {
22+ gate := newSamePeerInputGate()
23+ gate.setPaused(true)
24+ ctx, cancel := context.WithCancel(context.Background())
25+ cancel()
26+ if gate.wait(ctx) {
27+ t.Fatal("cancelled input gate wait succeeded")
28+ }
29+}
M internal/usecase/daemon/attachment_transition_publish.go +1 -0
@@ -187,6 +187,7 @@ func (d *Daemon) applyTargetStateLocked(publication *attachmentPublication) bool
187187 func publishAttachmentOwnershipLocked(publication *attachmentPublication) {
188188 req := publication.req
189189 if publication.source != req.target {
190+ req.next.clearSamePeerOffer()
190191 unregisterAttachmentSessionLocked(publication.source, req.next)
191192 }
192193 registered := registerAttachmentSessionLocked(req.target, req.next)
@@ -187,6 +187,7 @@ func (d *Daemon) applyTargetStateLocked(publication *attachmentPublication) bool
187 func publishAttachmentOwnershipLocked(publication *attachmentPublication) {187 func publishAttachmentOwnershipLocked(publication *attachmentPublication) {
188 req := publication.req188 req := publication.req
189 if publication.source != req.target {189 if publication.source != req.target {
190+ req.next.clearSamePeerOffer()
190 unregisterAttachmentSessionLocked(publication.source, req.next)191 unregisterAttachmentSessionLocked(publication.source, req.next)
191 }192 }
192 registered := registerAttachmentSessionLocked(req.target, req.next)193 registered := registerAttachmentSessionLocked(req.target, req.next)
M internal/usecase/daemon/client.go +2 -0
@@ -98,6 +98,8 @@ type attachedClient struct {
9898 routeMu sync.RWMutex
9999 routeSnapshot ports.RecentRouteSnapshot
100100 pendingRouteIdentity bool
101+ samePeerOfferMu sync.Mutex
102+ samePeerOffer *ports.ExactSessionTarget
101103 routeAttentionSubscription ports.RouteAttentionSubscription
102104 // routeCreatedSession marks a session created by this attachment's route.
103105 // A handshake that never commits Welcome must tear down that exact empty
@@ -98,6 +98,8 @@ type attachedClient struct {
98 routeMu sync.RWMutex98 routeMu sync.RWMutex
99 routeSnapshot ports.RecentRouteSnapshot99 routeSnapshot ports.RecentRouteSnapshot
100 pendingRouteIdentity bool100 pendingRouteIdentity bool
101+ samePeerOfferMu sync.Mutex
102+ samePeerOffer *ports.ExactSessionTarget
101 routeAttentionSubscription ports.RouteAttentionSubscription103 routeAttentionSubscription ports.RouteAttentionSubscription
102 // routeCreatedSession marks a session created by this attachment's route.104 // routeCreatedSession marks a session created by this attachment's route.
103 // A handshake that never commits Welcome must tear down that exact empty105 // A handshake that never commits Welcome must tear down that exact empty
M internal/usecase/daemon/client_frame_routing.go +7 -0
@@ -126,6 +126,13 @@ func (d *Daemon) handleAttachmentClientFrame(token attachmentConnectionToken, f
126126 if _, derr := ports.UnmarshalOutputResetRequest(f.Payload); derr == nil {
127127 d.resetOutput(token)
128128 }
129+ case ports.MsgSamePeerSwitchRequest:
130+ request, derr := ports.UnmarshalSamePeerSwitchRequest(f.Payload)
131+ if derr != nil {
132+ d.log.Warn("malformed same-peer switch request", "err", derr)
133+ break
134+ }
135+ d.switchSamePeerForAttachment(token, request)
129136 case ports.MsgRecentRouteSnapshot:
130137 snapshot, derr := ports.UnmarshalRecentRouteSnapshot(f.Payload)
131138 if derr != nil {
@@ -126,6 +126,13 @@ func (d *Daemon) handleAttachmentClientFrame(token attachmentConnectionToken, f
126 if _, derr := ports.UnmarshalOutputResetRequest(f.Payload); derr == nil {126 if _, derr := ports.UnmarshalOutputResetRequest(f.Payload); derr == nil {
127 d.resetOutput(token)127 d.resetOutput(token)
128 }128 }
129+ case ports.MsgSamePeerSwitchRequest:
130+ request, derr := ports.UnmarshalSamePeerSwitchRequest(f.Payload)
131+ if derr != nil {
132+ d.log.Warn("malformed same-peer switch request", "err", derr)
133+ break
134+ }
135+ d.switchSamePeerForAttachment(token, request)
129 case ports.MsgRecentRouteSnapshot:136 case ports.MsgRecentRouteSnapshot:
130 snapshot, derr := ports.UnmarshalRecentRouteSnapshot(f.Payload)137 snapshot, derr := ports.UnmarshalRecentRouteSnapshot(f.Payload)
131 if derr != nil {138 if derr != nil {
M internal/usecase/daemon/client_transport_errors.go +1 -0
@@ -65,6 +65,7 @@ func (d *Daemon) finishClientGone(sess *session, ac *attachedClient, failed port
6565 if sess == nil || ac == nil {
6666 return
6767 }
68+ ac.clearSamePeerOffer()
6869 if d.afterClientGoneDetach != nil {
6970 d.afterClientGoneDetach()
7071 }
@@ -65,6 +65,7 @@ func (d *Daemon) finishClientGone(sess *session, ac *attachedClient, failed port
65 if sess == nil || ac == nil {65 if sess == nil || ac == nil {
66 return66 return
67 }67 }
68+ ac.clearSamePeerOffer()
68 if d.afterClientGoneDetach != nil {69 if d.afterClientGoneDetach != nil {
69 d.afterClientGoneDetach()70 d.afterClientGoneDetach()
70 }71 }
M internal/usecase/daemon/picker.go +22 -10
@@ -899,9 +899,10 @@ func (d *Daemon) switchToTargetForAttachment(token attachmentConnectionToken, ta
899899 return d.sendLocalAttachTargetForAttachment(token, target, guard, action)
900900 }
901901
902-// sendLocalAttachTargetForAttachment makes local picker navigation follow the
903-// same client-owned handoff path as remote navigation. The client reconnects
904-// to this daemon and restores its per-route tab from the route ledger.
902+// sendLocalAttachTargetForAttachment offers an endpoint-empty, exact local
903+// target on the current authenticated connection. A current client confirms it
904+// with MsgSamePeerSwitchRequest; an interrupted or older client retains the
905+// existing close-and-redial fallback without any daemon-origin inference.
905906 func (d *Daemon) sendLocalAttachTargetForAttachment(token attachmentConnectionToken, target picker.Target, guard sessionHandoffGuard, action string) error {
906907 d.mu.Lock()
907908 var targetSess *session
@@ -930,9 +931,9 @@ func (d *Daemon) sendLocalAttachTargetForAttachment(token attachmentConnectionTo
930931 if !matches || !token.attachmentCurrent() {
931932 return errAttachmentTransition
932933 }
933- if target.TabID != "" {
934+ if targetSess != nil && target.TabID != "" {
934935 // An explicit tab row is a direct user choice, not route memory; retain
935- // the daemon-side transition so it opens exactly that tab.
936+ // the established daemon-side transition so it opens exactly that tab.
936937 return d.switchToTargetGuardedForAttachment(token.sess, token.ac, target, guard, &token, action)
937938 }
938939
@@ -942,16 +943,27 @@ func (d *Daemon) sendLocalAttachTargetForAttachment(token attachmentConnectionTo
942943 ExactTarget: exactTarget,
943944 EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
944945 })
945- if payload == nil {
946+ if payload == nil || exactTarget == nil {
946947 return errAttachmentTransition
947948 }
949+ samePeerEligible := action == "picker-select" && targetSess != nil && target.TabIndex == 0
950+ if samePeerEligible {
951+ token.ac.offerSamePeerTarget(*exactTarget)
952+ }
948953 if err := token.sendControl(ports.Frame{Type: ports.MsgAttachTarget, Payload: payload}); err != nil {
949- return domain.UserErr(domain.NoticeSessionUnavailable, "couldn't attach to local session", err)
954+ if samePeerEligible {
955+ token.ac.clearSamePeerOffer()
956+ }
957+ return domain.UserErr(domain.NoticeSessionUnavailable, "couldn't offer local session switch", err)
950958 }
951- if guard.closePicker {
952- d.closePicker(token.ac)
959+ if !samePeerEligible {
960+ // Only ordinary active picker session rows are eligible. Stopped,
961+ // index-only, and non-picker navigation retain close-and-dial handoff.
962+ if guard.closePicker {
963+ d.closePicker(token.ac)
964+ }
965+ d.clientGoneForAttachment(token, false)
953966 }
954- d.clientGoneForAttachment(token, false)
955967 return nil
956968 }
957969
@@ -899,9 +899,10 @@ func (d *Daemon) switchToTargetForAttachment(token attachmentConnectionToken, ta
899 return d.sendLocalAttachTargetForAttachment(token, target, guard, action)899 return d.sendLocalAttachTargetForAttachment(token, target, guard, action)
900 }900 }
901 901
902-// sendLocalAttachTargetForAttachment makes local picker navigation follow the902+// sendLocalAttachTargetForAttachment offers an endpoint-empty, exact local
903-// same client-owned handoff path as remote navigation. The client reconnects903+// target on the current authenticated connection. A current client confirms it
904-// to this daemon and restores its per-route tab from the route ledger.904+// with MsgSamePeerSwitchRequest; an interrupted or older client retains the
905+// existing close-and-redial fallback without any daemon-origin inference.
905 func (d *Daemon) sendLocalAttachTargetForAttachment(token attachmentConnectionToken, target picker.Target, guard sessionHandoffGuard, action string) error {906 func (d *Daemon) sendLocalAttachTargetForAttachment(token attachmentConnectionToken, target picker.Target, guard sessionHandoffGuard, action string) error {
906 d.mu.Lock()907 d.mu.Lock()
907 var targetSess *session908 var targetSess *session
@@ -930,9 +931,9 @@ func (d *Daemon) sendLocalAttachTargetForAttachment(token attachmentConnectionTo
930 if !matches || !token.attachmentCurrent() {931 if !matches || !token.attachmentCurrent() {
931 return errAttachmentTransition932 return errAttachmentTransition
932 }933 }
933- if target.TabID != "" {934+ if targetSess != nil && target.TabID != "" {
934 // An explicit tab row is a direct user choice, not route memory; retain935 // An explicit tab row is a direct user choice, not route memory; retain
935- // the daemon-side transition so it opens exactly that tab.936+ // the established daemon-side transition so it opens exactly that tab.
936 return d.switchToTargetGuardedForAttachment(token.sess, token.ac, target, guard, &token, action)937 return d.switchToTargetGuardedForAttachment(token.sess, token.ac, target, guard, &token, action)
937 }938 }
938 939
@@ -942,16 +943,27 @@ func (d *Daemon) sendLocalAttachTargetForAttachment(token attachmentConnectionTo
942 ExactTarget: exactTarget,943 ExactTarget: exactTarget,
943 EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,944 EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
944 })945 })
945- if payload == nil {946+ if payload == nil || exactTarget == nil {
946 return errAttachmentTransition947 return errAttachmentTransition
947 }948 }
949+ samePeerEligible := action == "picker-select" && targetSess != nil && target.TabIndex == 0
950+ if samePeerEligible {
951+ token.ac.offerSamePeerTarget(*exactTarget)
952+ }
948 if err := token.sendControl(ports.Frame{Type: ports.MsgAttachTarget, Payload: payload}); err != nil {953 if err := token.sendControl(ports.Frame{Type: ports.MsgAttachTarget, Payload: payload}); err != nil {
949- return domain.UserErr(domain.NoticeSessionUnavailable, "couldn't attach to local session", err)954+ if samePeerEligible {
955+ token.ac.clearSamePeerOffer()
956+ }
957+ return domain.UserErr(domain.NoticeSessionUnavailable, "couldn't offer local session switch", err)
950 }958 }
951- if guard.closePicker {959+ if !samePeerEligible {
952- d.closePicker(token.ac)960+ // Only ordinary active picker session rows are eligible. Stopped,
961+ // index-only, and non-picker navigation retain close-and-dial handoff.
962+ if guard.closePicker {
963+ d.closePicker(token.ac)
964+ }
965+ d.clientGoneForAttachment(token, false)
953 }966 }
954- d.clientGoneForAttachment(token, false)
955 return nil967 return nil
956 }968 }
957 969
M internal/usecase/daemon/remote_attach_test.go +2 -2
@@ -17,7 +17,7 @@ func remoteLifecycleForTest() domain.SessionLifecycleID {
1717 return id
1818 }
1919
20-func TestLocalPickerHandoffCarriesExactLifecycle(t *testing.T) {
20+func TestLocalPickerOfferCarriesExactLifecycle(t *testing.T) {
2121 d := newRemotePickerDaemon(nil)
2222 source, ac, sends := addRemoteRefreshPickerOwner(t, d, "source")
2323 target, _, _ := addRemoteRefreshPickerOwner(t, d, "target")
@@ -35,7 +35,7 @@ func TestLocalPickerHandoffCarriesExactLifecycle(t *testing.T) {
3535 require.Empty(t, got.Endpoint)
3636 require.Equal(t, ports.IntentAttach, got.Intent)
3737 require.Equal(t, &ports.ExactSessionTarget{LifecycleID: target.incarnation, SessionName: target.name}, got.ExactTarget)
38- require.Nil(t, ac.currentAttachmentSession())
38+ require.Same(t, source, ac.currentAttachmentSession(), "the source remains attached until the client confirms the switch")
3939 }
4040
4141 func TestRemotePickerRichHandoffCarriesLifecycleTabAndPolicy(t *testing.T) {
@@ -17,7 +17,7 @@ func remoteLifecycleForTest() domain.SessionLifecycleID {
17 return id17 return id
18 }18 }
19 19
20-func TestLocalPickerHandoffCarriesExactLifecycle(t *testing.T) {20+func TestLocalPickerOfferCarriesExactLifecycle(t *testing.T) {
21 d := newRemotePickerDaemon(nil)21 d := newRemotePickerDaemon(nil)
22 source, ac, sends := addRemoteRefreshPickerOwner(t, d, "source")22 source, ac, sends := addRemoteRefreshPickerOwner(t, d, "source")
23 target, _, _ := addRemoteRefreshPickerOwner(t, d, "target")23 target, _, _ := addRemoteRefreshPickerOwner(t, d, "target")
@@ -35,7 +35,7 @@ func TestLocalPickerHandoffCarriesExactLifecycle(t *testing.T) {
35 require.Empty(t, got.Endpoint)35 require.Empty(t, got.Endpoint)
36 require.Equal(t, ports.IntentAttach, got.Intent)36 require.Equal(t, ports.IntentAttach, got.Intent)
37 require.Equal(t, &ports.ExactSessionTarget{LifecycleID: target.incarnation, SessionName: target.name}, got.ExactTarget)37 require.Equal(t, &ports.ExactSessionTarget{LifecycleID: target.incarnation, SessionName: target.name}, got.ExactTarget)
38- require.Nil(t, ac.currentAttachmentSession())38+ require.Same(t, source, ac.currentAttachmentSession(), "the source remains attached until the client confirms the switch")
39 }39 }
40 40
41 func TestRemotePickerRichHandoffCarriesLifecycleTabAndPolicy(t *testing.T) {41 func TestRemotePickerRichHandoffCarriesLifecycleTabAndPolicy(t *testing.T) {
A internal/usecase/daemon/same_peer_switch.go +120 -0
@@ -0,0 +1,120 @@
1+package daemon
2+
3+import (
4+ "github.com/bnema/vev/internal/domain"
5+ "github.com/bnema/vev/internal/ports"
6+)
7+
8+func (ac *attachedClient) offerSamePeerTarget(target ports.ExactSessionTarget) {
9+ ac.samePeerOfferMu.Lock()
10+ ac.samePeerOffer = &target
11+ ac.samePeerOfferMu.Unlock()
12+}
13+
14+func (ac *attachedClient) clearSamePeerOffer() {
15+ ac.samePeerOfferMu.Lock()
16+ ac.samePeerOffer = nil
17+ ac.samePeerOfferMu.Unlock()
18+}
19+
20+// consumeSamePeerOffer linearizes the client confirmation with the daemon's
21+// endpoint-empty offer. A client cannot select an arbitrary local session by
22+// manufacturing this frame.
23+func (ac *attachedClient) consumeSamePeerOffer(target ports.ExactSessionTarget) bool {
24+ ac.samePeerOfferMu.Lock()
25+ defer ac.samePeerOfferMu.Unlock()
26+ if ac.samePeerOffer == nil || *ac.samePeerOffer != target {
27+ return false
28+ }
29+ ac.samePeerOffer = nil
30+ return true
31+}
32+
33+// switchSamePeerForAttachment commits one client-confirmed endpoint-empty
34+// target. The request never names an endpoint: its authority is the current
35+// authenticated attachment plus the target's exact lifecycle identity.
36+func (d *Daemon) switchSamePeerForAttachment(token attachmentConnectionToken, request ports.SamePeerSwitchRequest) {
37+ if err := request.Validate(); err != nil || token.sess == nil || token.ac == nil {
38+ return
39+ }
40+ if !token.ac.consumeSamePeerOffer(request.Target) {
41+ d.sendSamePeerSwitchFailure(token, request.RequestID, ports.SamePeerSwitchStaleTarget)
42+ return
43+ }
44+
45+ target, targetTabIndex, ok := d.samePeerTarget(request)
46+ if !ok || target == token.sess {
47+ d.sendSamePeerSwitchFailure(token, request.RequestID, ports.SamePeerSwitchStaleTarget)
48+ return
49+ }
50+
51+ transition, err := d.transitionAttachment(attachmentTransitionRequest{
52+ source: token.sess,
53+ target: target,
54+ next: token.ac,
55+ expectedTransport: token.transport,
56+ sourceToken: &token,
57+ action: "same-peer-switch",
58+ expectedTargetLifecycle: &attachmentLifecycleFence{
59+ name: request.Target.SessionName,
60+ incarnation: request.Target.LifecycleID,
61+ checkIncarnation: true,
62+ },
63+ activateTargetTab: true,
64+ targetTabIndex: targetTabIndex,
65+ ready: true,
66+ })
67+ if err != nil {
68+ d.sendSamePeerSwitchFailure(token, request.RequestID, ports.SamePeerSwitchStaleTarget)
69+ return
70+ }
71+
72+ if fresh, admitted := token.ac.beginAttachmentEffect(transition.published); admitted {
73+ d.closePicker(token.ac)
74+ fresh.End()
75+ }
76+ d.touchMRU(target)
77+ d.deferAttachmentTransitionCleanups(transition)
78+ d.firstPaintForTransition(transition.published)
79+}
80+
81+// samePeerTarget resolves the exact target and the client-owned tab cursor
82+// under the daemon/session locks. A missing cursor deliberately falls back to
83+// the target session's current default rather than leaking the source view.
84+func (d *Daemon) samePeerTarget(request ports.SamePeerSwitchRequest) (*session, int, bool) {
85+ d.mu.Lock()
86+ defer d.mu.Unlock()
87+
88+ target := d.findByNameLocked(request.Target.SessionName)
89+ if target == nil {
90+ return nil, 0, false
91+ }
92+ target.mu.Lock()
93+ defer target.mu.Unlock()
94+ if target.incarnation != request.Target.LifecycleID || len(target.tabs) == 0 {
95+ return nil, 0, false
96+ }
97+ tabIndex := 0
98+ if request.PreferredTabID == "" {
99+ return target, tabIndex, true
100+ }
101+ for index, tab := range target.tabs {
102+ if tab != nil && domain.TabStableID(tab.stableID) == request.PreferredTabID {
103+ return target, index, true
104+ }
105+ }
106+ return target, tabIndex, true
107+}
108+
109+func (d *Daemon) sendSamePeerSwitchFailure(token attachmentConnectionToken, requestID uint64, code ports.SamePeerSwitchFailureCode) {
110+ if !token.attachmentCurrent() {
111+ return
112+ }
113+ payload, err := ports.MarshalSamePeerSwitchFailure(ports.SamePeerSwitchFailure{RequestID: requestID, Code: code})
114+ if err != nil {
115+ return
116+ }
117+ if err := token.sendControl(ports.Frame{Type: ports.MsgSamePeerSwitchFailure, Payload: payload}); err != nil {
118+ d.detachOnAttachmentSendError(token, token.transport.transport)
119+ }
120+}
@@ -0,0 +1,120 @@
1+package daemon
2+
3+import (
4+ "github.com/bnema/vev/internal/domain"
5+ "github.com/bnema/vev/internal/ports"
6+)
7+
8+func (ac *attachedClient) offerSamePeerTarget(target ports.ExactSessionTarget) {
9+ ac.samePeerOfferMu.Lock()
10+ ac.samePeerOffer = &target
11+ ac.samePeerOfferMu.Unlock()
12+}
13+
14+func (ac *attachedClient) clearSamePeerOffer() {
15+ ac.samePeerOfferMu.Lock()
16+ ac.samePeerOffer = nil
17+ ac.samePeerOfferMu.Unlock()
18+}
19+
20+// consumeSamePeerOffer linearizes the client confirmation with the daemon's
21+// endpoint-empty offer. A client cannot select an arbitrary local session by
22+// manufacturing this frame.
23+func (ac *attachedClient) consumeSamePeerOffer(target ports.ExactSessionTarget) bool {
24+ ac.samePeerOfferMu.Lock()
25+ defer ac.samePeerOfferMu.Unlock()
26+ if ac.samePeerOffer == nil || *ac.samePeerOffer != target {
27+ return false
28+ }
29+ ac.samePeerOffer = nil
30+ return true
31+}
32+
33+// switchSamePeerForAttachment commits one client-confirmed endpoint-empty
34+// target. The request never names an endpoint: its authority is the current
35+// authenticated attachment plus the target's exact lifecycle identity.
36+func (d *Daemon) switchSamePeerForAttachment(token attachmentConnectionToken, request ports.SamePeerSwitchRequest) {
37+ if err := request.Validate(); err != nil || token.sess == nil || token.ac == nil {
38+ return
39+ }
40+ if !token.ac.consumeSamePeerOffer(request.Target) {
41+ d.sendSamePeerSwitchFailure(token, request.RequestID, ports.SamePeerSwitchStaleTarget)
42+ return
43+ }
44+
45+ target, targetTabIndex, ok := d.samePeerTarget(request)
46+ if !ok || target == token.sess {
47+ d.sendSamePeerSwitchFailure(token, request.RequestID, ports.SamePeerSwitchStaleTarget)
48+ return
49+ }
50+
51+ transition, err := d.transitionAttachment(attachmentTransitionRequest{
52+ source: token.sess,
53+ target: target,
54+ next: token.ac,
55+ expectedTransport: token.transport,
56+ sourceToken: &token,
57+ action: "same-peer-switch",
58+ expectedTargetLifecycle: &attachmentLifecycleFence{
59+ name: request.Target.SessionName,
60+ incarnation: request.Target.LifecycleID,
61+ checkIncarnation: true,
62+ },
63+ activateTargetTab: true,
64+ targetTabIndex: targetTabIndex,
65+ ready: true,
66+ })
67+ if err != nil {
68+ d.sendSamePeerSwitchFailure(token, request.RequestID, ports.SamePeerSwitchStaleTarget)
69+ return
70+ }
71+
72+ if fresh, admitted := token.ac.beginAttachmentEffect(transition.published); admitted {
73+ d.closePicker(token.ac)
74+ fresh.End()
75+ }
76+ d.touchMRU(target)
77+ d.deferAttachmentTransitionCleanups(transition)
78+ d.firstPaintForTransition(transition.published)
79+}
80+
81+// samePeerTarget resolves the exact target and the client-owned tab cursor
82+// under the daemon/session locks. A missing cursor deliberately falls back to
83+// the target session's current default rather than leaking the source view.
84+func (d *Daemon) samePeerTarget(request ports.SamePeerSwitchRequest) (*session, int, bool) {
85+ d.mu.Lock()
86+ defer d.mu.Unlock()
87+
88+ target := d.findByNameLocked(request.Target.SessionName)
89+ if target == nil {
90+ return nil, 0, false
91+ }
92+ target.mu.Lock()
93+ defer target.mu.Unlock()
94+ if target.incarnation != request.Target.LifecycleID || len(target.tabs) == 0 {
95+ return nil, 0, false
96+ }
97+ tabIndex := 0
98+ if request.PreferredTabID == "" {
99+ return target, tabIndex, true
100+ }
101+ for index, tab := range target.tabs {
102+ if tab != nil && domain.TabStableID(tab.stableID) == request.PreferredTabID {
103+ return target, index, true
104+ }
105+ }
106+ return target, tabIndex, true
107+}
108+
109+func (d *Daemon) sendSamePeerSwitchFailure(token attachmentConnectionToken, requestID uint64, code ports.SamePeerSwitchFailureCode) {
110+ if !token.attachmentCurrent() {
111+ return
112+ }
113+ payload, err := ports.MarshalSamePeerSwitchFailure(ports.SamePeerSwitchFailure{RequestID: requestID, Code: code})
114+ if err != nil {
115+ return
116+ }
117+ if err := token.sendControl(ports.Frame{Type: ports.MsgSamePeerSwitchFailure, Payload: payload}); err != nil {
118+ d.detachOnAttachmentSendError(token, token.transport.transport)
119+ }
120+}
A internal/usecase/daemon/same_peer_switch_test.go +77 -0
@@ -0,0 +1,77 @@
1+package daemon
2+
3+import (
4+ "testing"
5+
6+ "github.com/stretchr/testify/require"
7+
8+ "github.com/bnema/vev/internal/domain"
9+ "github.com/bnema/vev/internal/ports"
10+)
11+
12+func TestSamePeerSwitchTransitionsExactTargetAndPreferredTab(t *testing.T) {
13+ d, source, ac, sends, releases := newManualTabSession(t, 1)
14+ defer releaseAll(releases)
15+
16+ lifecycle := domain.SessionLifecycleID{7}
17+ target := &session{
18+ sessionCore: sessionCore{id: "target", name: "target", incarnation: lifecycle, attachments: make(map[*attachedClient]struct{})},
19+ ctx: source.ctx,
20+ cancel: func() {},
21+ tabs: []*tab{
22+ newTab(nil, domain.Size{Cols: 80, Rows: 23}),
23+ newTab(nil, domain.Size{Cols: 80, Rows: 23}),
24+ },
25+ }
26+ for _, tab := range target.tabs {
27+ publishTiledPaneOwners(target, tab)
28+ }
29+ d.mu.Lock()
30+ d.sessions[target.id] = target
31+ d.mu.Unlock()
32+ ac.setRouteSnapshot(ports.RecentRouteSnapshot{Generation: 1})
33+
34+ token := source.attachmentToken(ac, ac.transport())
35+ effect, admitted := ac.beginAttachmentEffect(token)
36+ require.True(t, admitted)
37+ token.effect = effect
38+ defer effect.End()
39+ requestTarget := ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: "target"}
40+ ac.offerSamePeerTarget(requestTarget)
41+
42+ d.switchSamePeerForAttachment(token, ports.SamePeerSwitchRequest{
43+ RequestID: 1,
44+ Target: requestTarget,
45+ PreferredTabID: domain.TabStableID(target.tabs[1].stableID),
46+ })
47+
48+ require.Same(t, target, ac.currentAttachmentSession())
49+ require.Equal(t, domain.TabStableID(target.tabs[1].stableID), ac.viewSnapshot().tabID)
50+ identityFrame := awaitFrame(t, sends, ports.MsgCommittedRouteIdentity)
51+ identity, err := ports.UnmarshalCommittedRouteIdentity(identityFrame.Payload)
52+ require.NoError(t, err)
53+ require.Equal(t, ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: "target"}, identity.Target)
54+}
55+
56+func TestSamePeerSwitchRejectsStaleTargetWithoutMutation(t *testing.T) {
57+ d, source, ac, sends, releases := newManualTabSession(t, 1)
58+ defer releaseAll(releases)
59+ ac.setRouteSnapshot(ports.RecentRouteSnapshot{Generation: 1})
60+
61+ token := source.attachmentToken(ac, ac.transport())
62+ effect, admitted := ac.beginAttachmentEffect(token)
63+ require.True(t, admitted)
64+ token.effect = effect
65+ defer effect.End()
66+
67+ d.switchSamePeerForAttachment(token, ports.SamePeerSwitchRequest{
68+ RequestID: 1,
69+ Target: ports.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{9}, SessionName: "missing"},
70+ })
71+
72+ require.Same(t, source, ac.currentAttachmentSession())
73+ failureFrame := awaitFrame(t, sends, ports.MsgSamePeerSwitchFailure)
74+ failure, err := ports.UnmarshalSamePeerSwitchFailure(failureFrame.Payload)
75+ require.NoError(t, err)
76+ require.Equal(t, ports.SamePeerSwitchFailure{RequestID: 1, Code: ports.SamePeerSwitchStaleTarget}, failure)
77+}
@@ -0,0 +1,77 @@
1+package daemon
2+
3+import (
4+ "testing"
5+
6+ "github.com/stretchr/testify/require"
7+
8+ "github.com/bnema/vev/internal/domain"
9+ "github.com/bnema/vev/internal/ports"
10+)
11+
12+func TestSamePeerSwitchTransitionsExactTargetAndPreferredTab(t *testing.T) {
13+ d, source, ac, sends, releases := newManualTabSession(t, 1)
14+ defer releaseAll(releases)
15+
16+ lifecycle := domain.SessionLifecycleID{7}
17+ target := &session{
18+ sessionCore: sessionCore{id: "target", name: "target", incarnation: lifecycle, attachments: make(map[*attachedClient]struct{})},
19+ ctx: source.ctx,
20+ cancel: func() {},
21+ tabs: []*tab{
22+ newTab(nil, domain.Size{Cols: 80, Rows: 23}),
23+ newTab(nil, domain.Size{Cols: 80, Rows: 23}),
24+ },
25+ }
26+ for _, tab := range target.tabs {
27+ publishTiledPaneOwners(target, tab)
28+ }
29+ d.mu.Lock()
30+ d.sessions[target.id] = target
31+ d.mu.Unlock()
32+ ac.setRouteSnapshot(ports.RecentRouteSnapshot{Generation: 1})
33+
34+ token := source.attachmentToken(ac, ac.transport())
35+ effect, admitted := ac.beginAttachmentEffect(token)
36+ require.True(t, admitted)
37+ token.effect = effect
38+ defer effect.End()
39+ requestTarget := ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: "target"}
40+ ac.offerSamePeerTarget(requestTarget)
41+
42+ d.switchSamePeerForAttachment(token, ports.SamePeerSwitchRequest{
43+ RequestID: 1,
44+ Target: requestTarget,
45+ PreferredTabID: domain.TabStableID(target.tabs[1].stableID),
46+ })
47+
48+ require.Same(t, target, ac.currentAttachmentSession())
49+ require.Equal(t, domain.TabStableID(target.tabs[1].stableID), ac.viewSnapshot().tabID)
50+ identityFrame := awaitFrame(t, sends, ports.MsgCommittedRouteIdentity)
51+ identity, err := ports.UnmarshalCommittedRouteIdentity(identityFrame.Payload)
52+ require.NoError(t, err)
53+ require.Equal(t, ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: "target"}, identity.Target)
54+}
55+
56+func TestSamePeerSwitchRejectsStaleTargetWithoutMutation(t *testing.T) {
57+ d, source, ac, sends, releases := newManualTabSession(t, 1)
58+ defer releaseAll(releases)
59+ ac.setRouteSnapshot(ports.RecentRouteSnapshot{Generation: 1})
60+
61+ token := source.attachmentToken(ac, ac.transport())
62+ effect, admitted := ac.beginAttachmentEffect(token)
63+ require.True(t, admitted)
64+ token.effect = effect
65+ defer effect.End()
66+
67+ d.switchSamePeerForAttachment(token, ports.SamePeerSwitchRequest{
68+ RequestID: 1,
69+ Target: ports.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{9}, SessionName: "missing"},
70+ })
71+
72+ require.Same(t, source, ac.currentAttachmentSession())
73+ failureFrame := awaitFrame(t, sends, ports.MsgSamePeerSwitchFailure)
74+ failure, err := ports.UnmarshalSamePeerSwitchFailure(failureFrame.Payload)
75+ require.NoError(t, err)
76+ require.Equal(t, ports.SamePeerSwitchFailure{RequestID: 1, Code: ports.SamePeerSwitchStaleTarget}, failure)
77+}

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