9d20550 fix(client): harden typed session handling · vev

M internal/adapters/sessionwire/client_test.go +22 -7


@@ -35,7 +35,7 @@ func TestClientConnectionEncodesEveryClientMessage(t *testing.T) {
35	35		{name: "command", message: protocol.CommandRequest{Version: protocol.Version, RequestID: 1, Slug: "list-sessions"}, typeID: wire.MsgCommand},
36	36		{name: "reset", message: protocol.OutputResetRequest{}, typeID: wire.MsgOutputResetRequest},
37	37		{name: "preview", message: protocol.RemotePreviewRequest{Version: protocol.RemotePreviewSchemaVersion, Target: target, Width: 1, Height: 1}, typeID: wire.MsgRemotePreviewRequest},
38		-	{name: "attention", message: protocol.RouteAttentionSubscription{}, typeID: wire.MsgRouteAttentionSubscription},
	38	+	{name: "attention", message: protocol.RouteAttentionSubscription{Targets: []protocol.RouteAttentionTarget{}}, typeID: wire.MsgRouteAttentionSubscription},
39	39		{name: "same peer", message: protocol.SamePeerSwitchRequest{RequestID: 1, Target: exact}, typeID: wire.MsgSamePeerSwitchRequest},
40	40		{name: "parked", message: protocol.ParkedRouteRequest{RequestID: 1, LeaseID: lease, Action: protocol.ParkedRoutePrepare}, typeID: wire.MsgParkedRouteRequest},
41	41		{name: "snapshot", message: protocol.RecentRouteSnapshot{}, typeID: wire.MsgRecentRouteSnapshot},

@@ -46,6 +46,9 @@ func TestClientConnectionEncodesEveryClientMessage(t *testing.T) {
46	46		raw := &scriptedTransport{}
47	47		require.NoError(t, NewClientConnection(raw).SendClient(tt.message))
48	48		require.Equal(t, tt.typeID, raw.sent.Type)
	49	+	got, err := decodeClient(raw.sent)
	50	+	require.NoError(t, err)
	51	+	require.Equal(t, tt.message, got)
49	52		})
50	53		}
51	54		}

@@ -86,15 +89,27 @@ func TestClientConnectionDecodesEveryServerMessage(t *testing.T) {
86	89		}
87	90
88	91		func TestClientConnectionClassifiesFailuresAndPreservesCapabilities(t *testing.T) {
89		-	connection := NewClientConnection(&scriptedTransport{recv: wire.Frame{Type: wire.MsgInput}})
90		-	_, err := connection.ReceiveServer()
91		-	var failure *protocol.DecodeFailure
92		-	require.ErrorAs(t, err, &failure)
93		-	require.Equal(t, protocol.DecodeWrongDirection, failure.Category)
	92	+	validError := wire.MarshalErrorMsg(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "error"})
	93	+	for _, tt := range []struct {
	94	+	name string
	95	+	frame wire.Frame
	96	+	category protocol.DecodeCategory
	97	+	}{
	98	+	{name: "wrong direction", frame: wire.Frame{Type: wire.MsgInput}, category: protocol.DecodeWrongDirection},
	99	+	{name: "truncated payload", frame: wire.Frame{Type: wire.MsgError, Payload: validError[:1]}, category: protocol.DecodeMalformed},
	100	+	{name: "trailing garbage", frame: wire.Frame{Type: wire.MsgError, Payload: append(append([]byte(nil), validError...), 0xff)}, category: protocol.DecodeMalformed},
	101	+	} {
	102	+	t.Run(tt.name, func(t *testing.T) {
	103	+	_, err := NewClientConnection(&scriptedTransport{recv: tt.frame}).ReceiveServer()
	104	+	var failure *protocol.DecodeFailure
	105	+	require.ErrorAs(t, err, &failure)
	106	+	require.Equal(t, tt.category, failure.Category)
	107	+	})
	108	+	}
94	109
95	110		events := make(chan ports.LinkEvent, 1)
96	111		raw := &capableTransport{events: events}
97		-	connection = NewClientConnection(raw)
	112	+	connection := NewClientConnection(raw)
98	113		require.Equal(t, uint8(1), connection.Capabilities().PreferredOutputWindow)
99	114		require.True(t, connection.Capabilities().LinkState)
100	115		require.Equal(t, ports.LinkStateDegraded, connection.LinkState())

 		{name: "command", message: protocol.CommandRequest{Version: protocol.Version, RequestID: 1, Slug: "list-sessions"}, typeID: wire.MsgCommand},
 		{name: "reset", message: protocol.OutputResetRequest{}, typeID: wire.MsgOutputResetRequest},
 		{name: "preview", message: protocol.RemotePreviewRequest{Version: protocol.RemotePreviewSchemaVersion, Target: target, Width: 1, Height: 1}, typeID: wire.MsgRemotePreviewRequest},
-		{name: "attention", message: protocol.RouteAttentionSubscription{}, typeID: wire.MsgRouteAttentionSubscription},
+		{name: "attention", message: protocol.RouteAttentionSubscription{Targets: []protocol.RouteAttentionTarget{}}, typeID: wire.MsgRouteAttentionSubscription},
 		{name: "same peer", message: protocol.SamePeerSwitchRequest{RequestID: 1, Target: exact}, typeID: wire.MsgSamePeerSwitchRequest},
 		{name: "parked", message: protocol.ParkedRouteRequest{RequestID: 1, LeaseID: lease, Action: protocol.ParkedRoutePrepare}, typeID: wire.MsgParkedRouteRequest},
 		{name: "snapshot", message: protocol.RecentRouteSnapshot{}, typeID: wire.MsgRecentRouteSnapshot},
 			raw := &scriptedTransport{}
 			require.NoError(t, NewClientConnection(raw).SendClient(tt.message))
 			require.Equal(t, tt.typeID, raw.sent.Type)
+			got, err := decodeClient(raw.sent)
+			require.NoError(t, err)
+			require.Equal(t, tt.message, got)
 		})
 	}
 }
 }
 func TestClientConnectionClassifiesFailuresAndPreservesCapabilities(t *testing.T) {
-	connection := NewClientConnection(&scriptedTransport{recv: wire.Frame{Type: wire.MsgInput}})
+	validError := wire.MarshalErrorMsg(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "error"})
-	_, err := connection.ReceiveServer()
+	for _, tt := range []struct {
-	var failure *protocol.DecodeFailure
+		name     string
-	require.ErrorAs(t, err, &failure)
+		frame    wire.Frame
-	require.Equal(t, protocol.DecodeWrongDirection, failure.Category)
+		category protocol.DecodeCategory
+	}{
+		{name: "wrong direction", frame: wire.Frame{Type: wire.MsgInput}, category: protocol.DecodeWrongDirection},
+		{name: "truncated payload", frame: wire.Frame{Type: wire.MsgError, Payload: validError[:1]}, category: protocol.DecodeMalformed},
+		{name: "trailing garbage", frame: wire.Frame{Type: wire.MsgError, Payload: append(append([]byte(nil), validError...), 0xff)}, category: protocol.DecodeMalformed},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := NewClientConnection(&scriptedTransport{recv: tt.frame}).ReceiveServer()
+			var failure *protocol.DecodeFailure
+			require.ErrorAs(t, err, &failure)
+			require.Equal(t, tt.category, failure.Category)
+		})
+	}
 	events := make(chan ports.LinkEvent, 1)
 	raw := &capableTransport{events: events}
-	connection = NewClientConnection(raw)
+	connection := NewClientConnection(raw)
 	require.Equal(t, uint8(1), connection.Capabilities().PreferredOutputWindow)
 	require.True(t, connection.Capabilities().LinkState)
 	require.Equal(t, ports.LinkStateDegraded, connection.LinkState())

M internal/app/run_test.go +6 -6


@@ -812,9 +812,9 @@ func (d namedDialer) Dial(context.Context) (ports.Transport, error) {
812	812		return nil, fmt.Errorf("not used: %s", d.name)
813	813		}
814	814
815		-	func requireNamedClientDialer(t *testing.T, dialer ports.ClientDialer, name string) {
	815	+	func requireNamedClientDialer(t *testing.T, ctx context.Context, dialer ports.ClientDialer, name string) {
816	816		t.Helper()
817		-	_, err := dialer.Dial(context.Background())
	817	+	_, err := dialer.Dial(ctx)
818	818		require.EqualError(t, err, "not used: "+name)
819	819		}
820	820

@@ -846,8 +846,8 @@ func TestRunAttachWithDepsSelectsRemoteTransport(t *testing.T) {
846	846		remoteDialerFactory: factory,
847	847		selectedRemoteTransport: tt.selectedTransport,
848	848		clipboard: clip,
849		-	runClient: func(_ context.Context, deps client.Dependencies, request client.AttachRequest) error {
850		-	requireNamedClientDialer(t, deps.Dialer, "remote")
	849	+	runClient: func(ctx context.Context, deps client.Dependencies, request client.AttachRequest) error {
	850	+	requireNamedClientDialer(t, ctx, deps.Dialer, "remote")
851	851		gotDialer = "remote"
852	852		gotRemote = deps.Remote
853	853		if !request.Remote \|\| request.EnvironmentPolicy != protocol.EnvironmentPolicyClientOwned {

@@ -1069,8 +1069,8 @@ func TestRunAttachWithDepsBuildsLocalDialer(t *testing.T) {
1069	1069		localDialer: func() ports.Dialer { return namedDialer{name: "local"} },
1070	1070		remoteDialerFactory: factory,
1071	1071		clipboard: &fakeClipboardReader{}, // must NOT reach runClient for a local attach
1072		-	runClient: func(_ context.Context, deps client.Dependencies, request client.AttachRequest) error {
1073		-	requireNamedClientDialer(t, deps.Dialer, "local")
	1072	+	runClient: func(ctx context.Context, deps client.Dependencies, request client.AttachRequest) error {
	1073	+	requireNamedClientDialer(t, ctx, deps.Dialer, "local")
1074	1074		gotDialer = "local"
1075	1075		gotRemote = deps.Remote
1076	1076		if request.Remote {


@@ -812,9 +812,9 @@ func (d namedDialer) Dial(context.Context) (ports.Transport, error) {
812		return nil, fmt.Errorf("not used: %s", d.name)	812		return nil, fmt.Errorf("not used: %s", d.name)
813		}	813		}
814			814
815	-	func requireNamedClientDialer(t *testing.T, dialer ports.ClientDialer, name string) {	815	+	func requireNamedClientDialer(t *testing.T, ctx context.Context, dialer ports.ClientDialer, name string) {
816		t.Helper()	816		t.Helper()
817	-	_, err := dialer.Dial(context.Background())	817	+	_, err := dialer.Dial(ctx)
818		require.EqualError(t, err, "not used: "+name)	818		require.EqualError(t, err, "not used: "+name)
819		}	819		}
820			820

@@ -846,8 +846,8 @@ func TestRunAttachWithDepsSelectsRemoteTransport(t *testing.T) {
846		remoteDialerFactory: factory,	846		remoteDialerFactory: factory,
847		selectedRemoteTransport: tt.selectedTransport,	847		selectedRemoteTransport: tt.selectedTransport,
848		clipboard: clip,	848		clipboard: clip,
849	-	runClient: func(_ context.Context, deps client.Dependencies, request client.AttachRequest) error {	849	+	runClient: func(ctx context.Context, deps client.Dependencies, request client.AttachRequest) error {
850	-	requireNamedClientDialer(t, deps.Dialer, "remote")	850	+	requireNamedClientDialer(t, ctx, deps.Dialer, "remote")
851		gotDialer = "remote"	851		gotDialer = "remote"
852		gotRemote = deps.Remote	852		gotRemote = deps.Remote
853		if !request.Remote \|\| request.EnvironmentPolicy != protocol.EnvironmentPolicyClientOwned {	853		if !request.Remote \|\| request.EnvironmentPolicy != protocol.EnvironmentPolicyClientOwned {

@@ -1069,8 +1069,8 @@ func TestRunAttachWithDepsBuildsLocalDialer(t *testing.T) {
1069		localDialer: func() ports.Dialer { return namedDialer{name: "local"} },	1069		localDialer: func() ports.Dialer { return namedDialer{name: "local"} },
1070		remoteDialerFactory: factory,	1070		remoteDialerFactory: factory,
1071		clipboard: &fakeClipboardReader{}, // must NOT reach runClient for a local attach	1071		clipboard: &fakeClipboardReader{}, // must NOT reach runClient for a local attach
1072	-	runClient: func(_ context.Context, deps client.Dependencies, request client.AttachRequest) error {	1072	+	runClient: func(ctx context.Context, deps client.Dependencies, request client.AttachRequest) error {
1073	-	requireNamedClientDialer(t, deps.Dialer, "local")	1073	+	requireNamedClientDialer(t, ctx, deps.Dialer, "local")
1074		gotDialer = "local"	1074		gotDialer = "local"
1075		gotRemote = deps.Remote	1075		gotRemote = deps.Remote
1076		if request.Remote {	1076		if request.Remote {

M internal/app/transport_observability_test.go +2 -2


@@ -236,11 +236,11 @@ func TestRunAttachPropagatesOneObserverToRemoteTransportFactory(t *testing.T) {
236	236		t.Cleanup(func() { newRemoteDialerFactoryWithRuntimeObserver = originalFactory })
237	237
238	238		originalRunClient := runClientWithDeps
239		-	runClientWithDeps = func(_ context.Context, deps client.Dependencies, _ client.AttachRequest) error {
	239	+	runClientWithDeps = func(ctx context.Context, deps client.Dependencies, _ client.AttachRequest) error {
240	240		if deps.RuntimeObserver != observer {
241	241		t.Fatalf("client transport observer = %v, want process observer %v", deps.RuntimeObserver, observer)
242	242		}
243		-	requireNamedClientDialer(t, deps.Dialer, "remote")
	243	+	requireNamedClientDialer(t, ctx, deps.Dialer, "remote")
244	244		return nil
245	245		}
246	246		t.Cleanup(func() { runClientWithDeps = originalRunClient })


@@ -236,11 +236,11 @@ func TestRunAttachPropagatesOneObserverToRemoteTransportFactory(t *testing.T) {
236		t.Cleanup(func() { newRemoteDialerFactoryWithRuntimeObserver = originalFactory })	236		t.Cleanup(func() { newRemoteDialerFactoryWithRuntimeObserver = originalFactory })
237			237
238		originalRunClient := runClientWithDeps	238		originalRunClient := runClientWithDeps
239	-	runClientWithDeps = func(_ context.Context, deps client.Dependencies, _ client.AttachRequest) error {	239	+	runClientWithDeps = func(ctx context.Context, deps client.Dependencies, _ client.AttachRequest) error {
240		if deps.RuntimeObserver != observer {	240		if deps.RuntimeObserver != observer {
241		t.Fatalf("client transport observer = %v, want process observer %v", deps.RuntimeObserver, observer)	241		t.Fatalf("client transport observer = %v, want process observer %v", deps.RuntimeObserver, observer)
242		}	242		}
243	-	requireNamedClientDialer(t, deps.Dialer, "remote")	243	+	requireNamedClientDialer(t, ctx, deps.Dialer, "remote")
244		return nil	244		return nil
245		}	245		}
246		t.Cleanup(func() { runClientWithDeps = originalRunClient })	246		t.Cleanup(func() { runClientWithDeps = originalRunClient })

M internal/usecase/client/client.go +17 -6


@@ -1487,7 +1487,11 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
1487	1487		// Before the asynchronous sender starts, preserve transport ordering by
1488	1488		// requesting the reset synchronously after the initial Theme publication.
1489	1489		resetErr := sendHandshake(func() error {
1490		-	return transport.SendClient(protocol.Resize{Size: geometry.Size, PixelWidth: geometry.PixelWidth, PixelHeight: geometry.PixelHeight})
	1490	+	message := protocol.Resize{Size: geometry.Size, PixelWidth: geometry.PixelWidth, PixelHeight: geometry.PixelHeight}
	1491	+	if err := protocol.ValidateGeometry(message.Geometry()); err != nil {
	1492	+	return fmt.Errorf("validating terminal size for reconnect reconciliation: %w", err)
	1493	+	}
	1494	+	return transport.SendClient(message)
1491	1495		})
1492	1496		if resetErr != nil {
1493	1497		resetErr = handshakeContextError(ctx, handshakeTimedOut, resetErr)

@@ -2111,6 +2115,13 @@ func runRecv(ctx context.Context, transport ports.ClientConnection, out chan<- r
2111	2115		if err != nil {
2112	2116		var failure *protocol.DecodeFailure
2113	2117		if errors.As(err, &failure) && (failure.Category == protocol.DecodeUnknownType \|\| failure.Category == protocol.DecodeWrongDirection) {
	2118	+	log.Debug("ignoring rejected server message",
	2119	+	"category", failure.Category,
	2120	+	"kind", failure.Kind,
	2121	+	"type", failure.Type,
	2122	+	"version", failure.Version,
	2123	+	"request", failure.RequestID,
	2124	+	)
2114	2125		continue
2115	2126		}
2116	2127		}

@@ -2128,9 +2139,6 @@ func runRecv(ctx context.Context, transport ports.ClientConnection, out chan<- r
2128	2139		}
2129	2140		}
2130	2141
2131		-	// runSender is the sole caller of SendClient: serialising the input and resize
2132		-	// pumps through one goroutine keeps the connection's single-writer contract.
2133		-	// A send failure cancels the loop and is surfaced once.
2134	2142		type cumulativeAckQueue struct {
2135	2143		mu sync.Mutex
2136	2144		latestEpoch uint64

@@ -2165,8 +2173,11 @@ func (q *cumulativeAckQueue) take() (uint64, uint64) {
2165	2173		return epoch, state
2166	2174		}
2167	2175
2168		-	// runSender preserves normal-frame order while allowing switch control and
2169		-	// output ACKs to make progress when raw input is held for a same-peer switch.
	2176	+	// runSender is the sole caller of SendClient: serialising the input and resize
	2177	+	// pumps through one goroutine keeps the connection's single-writer contract.
	2178	+	// It preserves normal-frame order while allowing switch control and output ACKs
	2179	+	// to make progress when raw input is held for a same-peer switch. A send failure
	2180	+	// cancels the loop and is surfaced once.
2170	2181		func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.ClientConnection, control <-chan protocol.ClientMessage, barriers <-chan chan struct{}, in <-chan protocol.ClientMessage, inputGate samePeerInputGate, acks cumulativeAckQueue, errCh chan<- error, log *slog.Logger) {
2171	2182		defer log.Debug("sender pump exited")
2172	2183		send := func(message protocol.ClientMessage) bool {

 			// Before the asynchronous sender starts, preserve transport ordering by
 			// requesting the reset synchronously after the initial Theme publication.
 			resetErr := sendHandshake(func() error {
-				return transport.SendClient(protocol.Resize{Size: geometry.Size, PixelWidth: geometry.PixelWidth, PixelHeight: geometry.PixelHeight})
+				message := protocol.Resize{Size: geometry.Size, PixelWidth: geometry.PixelWidth, PixelHeight: geometry.PixelHeight}
+				if err := protocol.ValidateGeometry(message.Geometry()); err != nil {
+					return fmt.Errorf("validating terminal size for reconnect reconciliation: %w", err)
+				}
+				return transport.SendClient(message)
 			})
 			if resetErr != nil {
 				resetErr = handshakeContextError(ctx, handshakeTimedOut, resetErr)
 		if err != nil {
 			var failure *protocol.DecodeFailure
 			if errors.As(err, &failure) && (failure.Category == protocol.DecodeUnknownType || failure.Category == protocol.DecodeWrongDirection) {
+				log.Debug("ignoring rejected server message",
+					"category", failure.Category,
+					"kind", failure.Kind,
+					"type", failure.Type,
+					"version", failure.Version,
+					"request", failure.RequestID,
+				)
 				continue
 			}
 		}
 	}
 }
-// runSender is the sole caller of SendClient: serialising the input and resize
-// pumps through one goroutine keeps the connection's single-writer contract.
-// A send failure cancels the loop and is surfaced once.
 type cumulativeAckQueue struct {
 	mu          sync.Mutex
 	latestEpoch uint64
 	return epoch, state
 }
-// runSender preserves normal-frame order while allowing switch control and
+// runSender is the sole caller of SendClient: serialising the input and resize
-// output ACKs to make progress when raw input is held for a same-peer switch.
+// pumps through one goroutine keeps the connection's single-writer contract.
+// It preserves normal-frame order while allowing switch control and output ACKs
+// to make progress when raw input is held for a same-peer switch. A send failure
+// cancels the loop and is surfaced once.
 func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.ClientConnection, control <-chan protocol.ClientMessage, barriers <-chan chan struct{}, in <-chan protocol.ClientMessage, inputGate *samePeerInputGate, acks *cumulativeAckQueue, errCh chan<- error, log *slog.Logger) {
 	defer log.Debug("sender pump exited")
 	send := func(message protocol.ClientMessage) bool {

M internal/usecase/client/handshake.go +0 -4


@@ -83,10 +83,6 @@ func watchHandshakeTransport(ctx context.Context, transport ports.ClientConnecti
83	83		}
84	84		}
85	85
86		-	func boundedHandshakeOperation(ctx context.Context, transport ports.ClientConnection, operation func() error) error {
87		-	return boundedHandshakeOperationWithTransition(ctx, transport, operation, nil)
88		-	}
89		-
90	86		func boundedHandshakeOperationWithTransition(ctx context.Context, transport ports.ClientConnection, operation func() error, transition *transitionUI) error {
91	87		if err := ctx.Err(); err != nil {
92	88		_ = transport.Close()


@@ -83,10 +83,6 @@ func watchHandshakeTransport(ctx context.Context, transport ports.ClientConnecti
83		}	83	}
84		}	84	}
85			85
86	-	func boundedHandshakeOperation(ctx context.Context, transport ports.ClientConnection, operation func() error) error {
87	-	return boundedHandshakeOperationWithTransition(ctx, transport, operation, nil)
88	-	}
89	-
90		func boundedHandshakeOperationWithTransition(ctx context.Context, transport ports.ClientConnection, operation func() error, transition *transitionUI) error {	86	func boundedHandshakeOperationWithTransition(ctx context.Context, transport ports.ClientConnection, operation func() error, transition *transitionUI) error {
91		if err := ctx.Err(); err != nil {	87	if err := ctx.Err(); err != nil {
92		_ = transport.Close()	88	_ = transport.Close()

M internal/usecase/client/typed_connection_external_test.go +9 -4


@@ -89,12 +89,17 @@ func (d *mockClientDialer) Dial(ctx context.Context) (ports.ClientConnection, er
89	89		}
90	90
91	91		func externalCapabilities(raw ports.Transport) protocol.ConnectionCapabilities {
92		-	_, d := raw.(ports.DatagramTransport)
93		-	window := uint8(8)
94		-	if d {
	92	+	_, dgram := raw.(ports.DatagramTransport)
	93	+	_, link := raw.(ports.LinkStateReporter)
	94	+	window := uint8(protocol.MaxOutputWindow)
	95	+	if dgram {
95	96		window = 1
96	97		}
97		-	return protocol.ConnectionCapabilities{OutputDataLimit: protocol.MaxOutputDataLen, PreferredOutputWindow: window}
	98	+	return protocol.ConnectionCapabilities{
	99	+	OutputDataLimit: protocol.MaxOutputDataLen,
	100	+	PreferredOutputWindow: window,
	101	+	LinkState: link,
	102	+	}
98	103		}
99	104		func externalClientFrame(m protocol.ClientMessage) (wire.Frame, error) {
100	105		switch x := m.(type) {

 }
 func externalCapabilities(raw ports.Transport) protocol.ConnectionCapabilities {
-	_, d := raw.(ports.DatagramTransport)
+	_, dgram := raw.(ports.DatagramTransport)
-	window := uint8(8)
+	_, link := raw.(ports.LinkStateReporter)
-	if d {
+	window := uint8(protocol.MaxOutputWindow)
+	if dgram {
 		window = 1
 	}
-	return protocol.ConnectionCapabilities{OutputDataLimit: protocol.MaxOutputDataLen, PreferredOutputWindow: window}
+	return protocol.ConnectionCapabilities{
+		OutputDataLimit:       protocol.MaxOutputDataLen,
+		PreferredOutputWindow: window,
+		LinkState:             link,
+	}
 }
 func externalClientFrame(m protocol.ClientMessage) (wire.Frame, error) {
 	switch x := m.(type) {

M internal/usecase/client/typed_connection_test.go +13 -4

@@ -1,6 +1,7 @@
1	1		package client
2	2
3	3		import (
	4	+	"bytes"
4	5		"context"
5	6		"errors"
6	7		"io"

@@ -181,7 +182,7 @@ func testServerMessage(frame wire.Frame) (protocol.ServerMessage, error) {
181	182		func testClientCapabilities(raw ports.Transport) protocol.ConnectionCapabilities {
182	183		_, dgram := raw.(ports.DatagramTransport)
183	184		_, link := raw.(ports.LinkStateReporter)
184		-	window := uint8(8)
	185	+	window := uint8(protocol.MaxOutputWindow)
185	186		if dgram {
186	187		window = 1
187	188		}

@@ -200,14 +201,17 @@ func testClientLinkEvents(raw ports.Transport) <-chan ports.LinkEvent {
200	201		return nil
201	202		}
202	203
203		-	func TestRunRecvIgnoresUnknownTypedBoundaryFailures(t *testing.T) {
	204	+	func TestRunRecvLogsIgnoredTypedBoundaryFailures(t *testing.T) {
204	205		connection := portsmocks.NewMockClientConnection(t)
205		-	connection.EXPECT().ReceiveServer().Return(nil, &protocol.DecodeFailure{Category: protocol.DecodeUnknownType}).Once()
	206	+	connection.EXPECT().ReceiveServer().Return(nil, &protocol.DecodeFailure{Category: protocol.DecodeUnknownType, Type: 255}).Once()
	207	+	connection.EXPECT().ReceiveServer().Return(nil, &protocol.DecodeFailure{Category: protocol.DecodeWrongDirection, Type: uint8(wire.MsgInput)}).Once()
206	208		connection.EXPECT().ReceiveServer().Return(protocol.Pong{}, nil).Once()
207	209		connection.EXPECT().ReceiveServer().Return(nil, io.EOF).Once()
208	210		results := make(chan recvResult, 2)
209	211		failed := make(chan struct{})
210		-	runRecv(context.Background(), connection, results, failed, slog.New(slog.DiscardHandler))
	212	+	var logs bytes.Buffer
	213	+	log := slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug}))
	214	+	runRecv(context.Background(), connection, results, failed, log)
211	215		first := <-results
212	216		require.Equal(t, protocol.Pong{}, first.message)
213	217		require.NoError(t, first.err)

@@ -218,6 +222,11 @@ func TestRunRecvIgnoresUnknownTypedBoundaryFailures(t *testing.T) {
218	222		default:
219	223		t.Fatal("terminal receive failure was not reported")
220	224		}
	225	+	require.Contains(t, logs.String(), "ignoring rejected server message")
	226	+	require.Contains(t, logs.String(), "category=2")
	227	+	require.Contains(t, logs.String(), "category=3")
	228	+	require.Contains(t, logs.String(), "type=255")
	229	+	require.Contains(t, logs.String(), "type=2")
221	230		}
222	231
223	232		func (t *ackRecordingTransport) SendClient(m protocol.ClientMessage) error {

 package client
 import (
+	"bytes"
 	"context"
 	"errors"
 	"io"
 func testClientCapabilities(raw ports.Transport) protocol.ConnectionCapabilities {
 	_, dgram := raw.(ports.DatagramTransport)
 	_, link := raw.(ports.LinkStateReporter)
-	window := uint8(8)
+	window := uint8(protocol.MaxOutputWindow)
 	if dgram {
 		window = 1
 	}
 	return nil
 }
-func TestRunRecvIgnoresUnknownTypedBoundaryFailures(t *testing.T) {
+func TestRunRecvLogsIgnoredTypedBoundaryFailures(t *testing.T) {
 	connection := portsmocks.NewMockClientConnection(t)
-	connection.EXPECT().ReceiveServer().Return(nil, &protocol.DecodeFailure{Category: protocol.DecodeUnknownType}).Once()
+	connection.EXPECT().ReceiveServer().Return(nil, &protocol.DecodeFailure{Category: protocol.DecodeUnknownType, Type: 255}).Once()
+	connection.EXPECT().ReceiveServer().Return(nil, &protocol.DecodeFailure{Category: protocol.DecodeWrongDirection, Type: uint8(wire.MsgInput)}).Once()
 	connection.EXPECT().ReceiveServer().Return(protocol.Pong{}, nil).Once()
 	connection.EXPECT().ReceiveServer().Return(nil, io.EOF).Once()
 	results := make(chan recvResult, 2)
 	failed := make(chan struct{})
-	runRecv(context.Background(), connection, results, failed, slog.New(slog.DiscardHandler))
+	var logs bytes.Buffer
+	log := slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug}))
+	runRecv(context.Background(), connection, results, failed, log)
 	first := <-results
 	require.Equal(t, protocol.Pong{}, first.message)
 	require.NoError(t, first.err)
 	default:
 		t.Fatal("terminal receive failure was not reported")
 	}
+	require.Contains(t, logs.String(), "ignoring rejected server message")
+	require.Contains(t, logs.String(), "category=2")
+	require.Contains(t, logs.String(), "category=3")
+	require.Contains(t, logs.String(), "type=255")
+	require.Contains(t, logs.String(), "type=2")
 }
 func (t *ackRecordingTransport) SendClient(m protocol.ClientMessage) error {