diff --git a/transport/client_test.go b/transport/client_test.go
index 5b8afd95..80aa339e 100644
--- a/transport/client_test.go
+++ b/transport/client_test.go
@@ -21,8 +21,8 @@ import (
 	"bytes"
 	"crypto/tls"
 	"errors"
+	"io"
 	"net"
-	"net/http"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -250,13 +250,29 @@ func TestTCPClient(t *testing.T) {
 			return nil, err
 		}
 
-		go func() { _ = http.Serve(listener, nil) }()
+		// Read and discard everything the client sends. The peer must not
+		// interpret the bytes as HTTP: once the codec batch path is enabled,
+		// it receives deflate/snappy frames, and an HTTP server closes the
+		// connection mid-test, making the write accounting assertions flaky.
+		go func() {
+			for {
+				conn, err := listener.Accept()
+				if err != nil {
+					return
+				}
+				go func() {
+					defer func() { _ = conn.Close() }()
+					_, _ = io.Copy(io.Discard, conn)
+				}()
+			}
+		}()
 		return listener, nil
 	}
 
 	listener, err := listenLocalServer()
 	assert.Nil(t, err)
 	assert.NotNil(t, listener)
+	t.Cleanup(func() { _ = listener.Close() })
 
 	addr := listener.Addr().(*net.TCPAddr)
 	t.Logf("server addr: %v", addr)
@@ -285,7 +301,9 @@ func TestTCPClient(t *testing.T) {
 	assert.Nil(t, err)
 	active := ss.GetActive()
 	assert.NotNil(t, active)
-	ss.SetCompressType(CompressNone)
+	// the session read loop already started IO, so reconfiguring the codec
+	// mid-stream must be rejected (see SetCompressType)
+	assert.Panics(t, func() { ss.SetCompressType(CompressNone) })
 	conn := ss.(*session).Connection.(*gettyTCPConn)
 	assert.True(t, conn.compress == CompressNone)
 	beforeWriteBytes := conn.writeBytes
@@ -313,7 +331,7 @@ func TestTCPClient(t *testing.T) {
 	beforeWriteBytes.Add(10)
 	assert.Equal(t, beforeWritePkgNum, conn.writePkgNum)
 	assert.Equal(t, beforeWriteBytes, conn.writeBytes)
-	ss.SetCompressType(CompressSnappy)
+	assert.Panics(t, func() { ss.SetCompressType(CompressSnappy) })
 	var anotherPkgs [][]byte
 	anotherPkgs = append(anotherPkgs, []byte("hello"), []byte("hello"))
 	l, err = ss.WriteBytesArray(anotherPkgs...)
@@ -323,7 +341,7 @@ func TestTCPClient(t *testing.T) {
 	beforeWriteBytes.Add(10)
 	assert.Equal(t, beforeWritePkgNum, conn.writePkgNum)
 	assert.Equal(t, beforeWriteBytes, conn.writeBytes)
-	assert.True(t, conn.compress == CompressSnappy)
+	assert.True(t, conn.compress == CompressNone)
 
 	batchSize := 128 * 1023
 	source := make([]byte, batchSize)
@@ -485,7 +503,9 @@ func TestNewWSClient(t *testing.T) {
 
 	assert.Equal(t, 1, msgHandler.SessionNumber())
 	ss := msgHandler.array[0]
-	ss.SetCompressType(CompressNone)
+	// the session read loop already started IO, so reconfiguring compression
+	// mid-stream must be rejected, same contract as the TCP connection
+	assert.Panics(t, func() { ss.SetCompressType(CompressNone) })
 	conn := ss.(*session).Connection.(*gettyWSConn)
 	assert.True(t, conn.compress == CompressNone)
 	err := conn.handlePing("hello")
diff --git a/transport/connection.go b/transport/connection.go
index 4fd89023..2d161d35 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -20,9 +20,11 @@ package getty
 import (
 	"compress/flate"
 	"crypto/tls"
+	"errors"
 	"fmt"
 	"io"
 	"net"
+	"os"
 	"sync"
 	"time"
 )
@@ -46,9 +48,46 @@ var (
 	connID     uatomic.Uint32
 )
 
+// ErrCodecStreamBroken is returned once a codec(compressed) stream cannot be
+// trusted any more: a read/write timed out or failed in the middle of a codec
+// block, so the decoder/encoder state and the bytes already on the wire are
+// desynchronized. flate/snappy latch such errors forever, hence the connection
+// is unusable and the session must be closed and rebuilt. It deliberately does
+// not unwrap to a net.Error, so session.handleTCPPackage treats it as fatal
+// instead of as a benign read timeout that can be retried.
+var ErrCodecStreamBroken = errors.New("getty: codec stream is broken, connection must be closed")
+
+// CodecStallTimeout bounds how long a codec(compressed) stream may stay silent
+// AFTER it has delivered part of a codec block: that is the only situation
+// where the peer provably died mid-write and the decoder state is
+// unrecoverable. A stream that is merely idle - no bytes at all since the last
+// fully decoded block - is healthy and is never subject to this timeout, no
+// matter how long the silence lasts.
+//
+// The mechanics live in codecPollingReader: the connection read timeout (see
+// SetReadTimeout, 1s by default) is used as a poll interval underneath the
+// codec, and poll timeouts are absorbed there instead of reaching the
+// flate/snappy reader (which would latch any error forever). Only a
+// mid-block stall of at least CodecStallTimeout is surfaced to the codec,
+// which is what declares the connection broken.
+//
+// The write side (codecPollingWriter) applies the same bound to a peer that
+// stopped reading: wTimeout is the poll interval, and a write attempt that
+// makes no progress for CodecStallTimeout breaks the stream, while a slow but
+// draining peer is waited on indefinitely.
+//
+// Set it to 0 to disable stall detection (a peer that dies mid-block then
+// blocks the read until the session is closed). It is copied per connection
+// when SetCompressType installs the codec, so set it before creating
+// connections.
+var CodecStallTimeout = 5 * time.Minute
+
 // Connection wrap some connection params and operations
 type Connection interface {
 	ID() uint32
+	// SetCompressType sets the compress type. It must be called before any
+	// recv/Send on the connection, typically in NewSessionCallback; TCP and
+	// websocket connections panic if it is called after IO started.
 	SetCompressType(CompressType)
 	LocalAddr() string
 	RemoteAddr() string
@@ -81,12 +120,18 @@ type Connection interface {
 // ///////////////////////////////////////
 
 type gettyConn struct {
-	id            uint32
-	compress      CompressType
+	id       uint32
+	compress CompressType
+	// codecEnabled reports whether reader/writer have been replaced by a codec.
+	// compress == CompressNone is not a substitute: CompressNone is
+	// flate.NoCompression(0), so SetCompressType(CompressNone) still installs a
+	// flate codec that frames the stream into deflate blocks. Such a stream is
+	// stateful and must not be treated like a raw connection.
+	codecEnabled  bool
 	readBytes     uatomic.Uint32   // read bytes
 	writeBytes    uatomic.Uint32   // write bytes
-	readPkgNum    uatomic.Uint32   // send pkg number
-	writePkgNum   uatomic.Uint32   // recv pkg number
+	readPkgNum    uatomic.Uint32   // recv pkg number
+	writePkgNum   uatomic.Uint32   // send pkg number
 	active        uatomic.Int64    // last active, in milliseconds
 	rTimeout      uatomic.Duration // network current limiting
 	wTimeout      uatomic.Duration
@@ -175,9 +220,31 @@ func (c *gettyConn) SetWriteTimeout(wTimeout time.Duration) {
 
 type gettyTCPConn struct {
 	gettyConn
-	reader io.Reader
-	writer io.Writer
-	conn   net.Conn
+	// lock guards the codec fields below; streamStarted is set by the first
+	// recv/Send and freezes the codec configuration (see SetCompressType).
+	lock          sync.Mutex
+	streamStarted bool
+	reader        io.Reader
+	writer        io.Writer
+	conn          net.Conn // immutable after construction, closed via closeOnce
+	closeOnce     sync.Once
+	// codecBroken is set once a codec read/write failed in the middle of the
+	// stream. Written by the read goroutine and read by writer goroutines, so
+	// it has to be atomic. Once set, recv/Send fail fast: touching the codec
+	// again would only produce more garbage on the wire.
+	codecBroken uatomic.Bool
+	// closing is set by CloseConn before it flushes the codec writer: the
+	// flush is best-effort, so codecPollingWriter surfaces the first poll
+	// timeout instead of waiting out a whole codecStallTimeout on a peer that
+	// stopped reading (sessionClosing() covers the session-driven close, this
+	// covers direct CloseConn use).
+	closing uatomic.Bool
+	// codecStallTimeout is the per-conn copy of CodecStallTimeout, taken when
+	// the codec is installed. 0 disables mid-block stall detection.
+	codecStallTimeout time.Duration
+	// pollReader is the deadline-owning reader installed between the codec and
+	// the raw conn by SetCompressType; nil on a raw connection.
+	pollReader *codecPollingReader
 }
 
 // create gettyTCPConn
@@ -209,6 +276,13 @@ func newGettyTCPConn(conn net.Conn) *gettyTCPConn {
 	}
 }
 
+// buffersWriter is implemented by writers that can take a whole batch at once.
+// The compress writers implement it so a batch becomes one compressed block
+// instead of one block per packet.
+type buffersWriter interface {
+	WriteBuffers(buffers [][]byte) (int64, error)
+}
+
 // for zip compress
 type writeFlusher struct {
 	flusher *flate.Writer
@@ -233,6 +307,34 @@ func (t *writeFlusher) Write(p []byte) (int, error) {
 	return n, nil
 }
 
+// WriteBuffers writes the whole batch into the compressor and flushes once at
+// the end. Calling Write per buffer would flush every time and degrade into N
+// separate deflate blocks: N syscalls, and no way to exploit patterns that
+// repeat across the packets.
+func (t *writeFlusher) WriteBuffers(buffers [][]byte) (int64, error) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+
+	var total int64
+	for _, b := range buffers {
+		n, err := t.flusher.Write(b)
+		total += int64(n)
+		if err != nil {
+			return total, perrors.WithStack(err)
+		}
+	}
+	return total, perrors.WithStack(t.flusher.Flush())
+}
+
+// Close terminates the flate stream (final block marker), so the peer's
+// decoder sees a clean EOF instead of io.ErrUnexpectedEOF. Mirrors
+// snappyWriteFlusher.Close; called by CloseConn on a healthy codec only.
+func (t *writeFlusher) Close() error {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	return perrors.WithStack(t.flusher.Close())
+}
+
 // for snappy compress. #102: snappy.NewBufferedWriter buffers writes and only
 // emits data on Flush, so small packets would sit in the buffer forever if not
 // flushed after every Write. This wrapper flushes on every Write, mirroring the
@@ -259,40 +361,373 @@ func (s *snappyWriteFlusher) Write(p []byte) (int, error) {
 	return n, nil
 }
 
+// WriteBuffers mirrors writeFlusher.WriteBuffers: write the whole batch, flush
+// once at the end, so a batch leaves as a single snappy block.
+func (s *snappyWriteFlusher) WriteBuffers(buffers [][]byte) (int64, error) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	var total int64
+	for _, b := range buffers {
+		n, err := s.writer.Write(b)
+		total += int64(n)
+		if err != nil {
+			return total, perrors.WithStack(err)
+		}
+	}
+	return total, perrors.WithStack(s.writer.Flush())
+}
+
 func (s *snappyWriteFlusher) Close() error {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 	return perrors.WithStack(s.writer.Close())
 }
 
-// SetCompressType set compress type(tcp: zip/snappy, websocket:zip)
+// SetCompressType set compress type(tcp: zip/snappy, websocket:zip).
+// It must be called before the first recv/Send on the connection, typically
+// inside NewSessionCallback; there is no codec renegotiation protocol, so
+// switching the stream format after IO started would desynchronize the peer's
+// decoder. Calling it on a started stream panics.
 func (t *gettyTCPConn) SetCompressType(c CompressType) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	if t.streamStarted {
+		log.Errorf("SetCompressType(%d) called after IO started on connection{local:%s, peer:%s}", c, t.local, t.peer)
+		panic("SetCompressType must be called before any recv/Send on the connection, e.g. in NewSessionCallback")
+	}
+	// The codec never reads the raw conn directly: codecPollingReader sits in
+	// between, owns the read deadlines and absorbs idle poll timeouts, so the
+	// error-latching flate/snappy readers only ever see a genuine mid-block
+	// stall (or a real IO error).
+	poller := &codecPollingReader{t: t, buf: make([]byte, maxReadBufLen)}
+	pollWriter := &codecPollingWriter{t: t}
 	switch c {
 	case CompressNone, CompressZip, CompressBestSpeed, CompressBestCompression, CompressHuffman:
-		ioReader := io.Reader(t.conn)
-		t.reader = flate.NewReader(ioReader)
+		t.reader = flate.NewReader(poller)
 
-		ioWriter := io.Writer(t.conn)
-		w, err := flate.NewWriter(ioWriter, int(c))
+		w, err := flate.NewWriter(pollWriter, int(c))
 		if err != nil {
 			panic(fmt.Sprintf("flate.NewReader(flate.DefaultCompress) = err(%s)", err))
 		}
 		t.writer = &writeFlusher{flusher: w}
 
 	case CompressSnappy:
-		ioReader := io.Reader(t.conn)
-		t.reader = snappy.NewReader(ioReader)
-		ioWriter := io.Writer(t.conn)
+		t.reader = snappy.NewReader(poller)
 		// #102: wrap the buffered snappy writer so every Write is flushed,
 		// otherwise small packets never leave the internal buffer.
-		t.writer = newSnappyWriteFlusher(snappy.NewBufferedWriter(ioWriter))
+		t.writer = newSnappyWriteFlusher(snappy.NewBufferedWriter(pollWriter))
 
 	default:
 		panic(fmt.Sprintf("illegal comparess type %d", c))
 	}
+	t.pollReader = poller
+	// Both branches replaced reader/writer with a codec (CompressNone included,
+	// see the codecEnabled comment), so the conn is no longer raw: all IO must
+	// go through t.reader/t.writer, and a read/write error can no longer be
+	// retried on this stream.
+	t.codecEnabled = true
+	t.codecStallTimeout = CodecStallTimeout
 	t.compress = c
 }
 
+// codecPollingReader is the io.Reader the codec (flate/snappy) reads from
+// instead of the raw conn. It exists because those decoders latch the first
+// error their source ever returns, so a plain read timeout must never reach
+// them - yet without timeouts a dead peer blocks the read goroutine forever.
+//
+// The reader polls the conn with rTimeout deadlines (the same poll interval a
+// raw connection uses) and classifies each timeout:
+//   - no bytes seen since the last fully decoded block: the stream is idle,
+//     which is healthy; the timeout is absorbed and the poll continues.
+//   - partial codec data seen (progress) and silence for at least
+//     codecStallTimeout: the peer died mid-block, the decoder state is
+//     unrecoverable; the timeout is surfaced, latched by the codec and turned
+//     into ErrCodecStreamBroken by codecReadError.
+//
+// The reader also implements io.ByteReader and carries the ONLY read-ahead
+// buffer under the codec: flate uses a Reader+ByteReader source directly
+// instead of wrapping it in a hidden bufio.Reader (snappy never reads ahead),
+// so bytes read ahead of the current codec block always live in r.buf where
+// this classification can see them. Progress therefore means "bytes were
+// delivered to the decoder since the last decode boundary" - a partial block
+// that spans two recv calls is still detected as a stall, because its
+// remainder is delivered from r.buf after the boundary.
+//
+// All fields are only touched by the single read goroutine; session.stop()
+// interacts with it solely by arming a conn deadline (the wakeup is seen here
+// as a timeout and passed through once sessionClosing()/codecBroken is set).
+type codecPollingReader struct {
+	t *gettyTCPConn
+	// buf/next/end: read-ahead buffer, next..end is the undelivered remainder.
+	buf  []byte
+	next int
+	end  int
+	// progress is true once bytes were delivered to the decoder after the last
+	// decode boundary (see boundary), i.e. the decoder may be holding part of
+	// a codec block.
+	progress bool
+	// lastDelivery is when the decoder last received bytes.
+	lastDelivery time.Time
+}
+
+// The ByteReader half of this assertion is load-bearing: without it flate
+// silently wraps the poller in its own bufio.Reader and read-ahead residue
+// becomes invisible again (the cross-recv stall blind spot returns).
+var _ flate.Reader = (*codecPollingReader)(nil)
+
+// boundary is called by recv after the codec returned decoded output: the
+// stream is at a block boundary again, so subsequent silence is idleness, not
+// a stall.
+//
+// "At a block boundary" is exact for flushed packets: flate only surfaces the
+// decoded output after it consumed the sync-flush empty stored block, and
+// snappy only after a whole chunk. The one approximation is flate's 32KB
+// window flush, which surfaces output mid-block; that is still covered
+// because the block's remaining bytes sit undelivered in r.buf and re-mark
+// progress on the next Read - EXCEPT when the peer died byte-exactly at such
+// a flush point with nothing left in r.buf. That coincidence degrades to a
+// hang indistinguishable from a peer dying between packets, which no
+// conn-level detection can catch; application heartbeats own that case.
+func (r *codecPollingReader) boundary() {
+	r.progress = false
+}
+
+func (r *codecPollingReader) markDelivery() {
+	r.progress = true
+	r.lastDelivery = time.Now()
+}
+
+func (r *codecPollingReader) Read(p []byte) (int, error) {
+	if r.next == r.end {
+		if err := r.fill(); err != nil {
+			return 0, err
+		}
+	}
+	n := copy(p, r.buf[r.next:r.end])
+	r.next += n
+	r.markDelivery()
+	return n, nil
+}
+
+// ReadByte makes this a flate.Reader source: flate then consumes exactly the
+// bytes it needs through here rather than over-reading via its own bufio.
+func (r *codecPollingReader) ReadByte() (byte, error) {
+	if r.next == r.end {
+		if err := r.fill(); err != nil {
+			return 0, err
+		}
+	}
+	b := r.buf[r.next]
+	r.next++
+	r.markDelivery()
+	return b, nil
+}
+
+// fill polls the conn until data arrives or a non-absorbable condition is hit.
+func (r *codecPollingReader) fill() error {
+	t := r.t
+	for {
+		if timeout := t.rTimeout.Load(); timeout > 0 {
+			// mid-block the poll must wake up no later than the stall deadline,
+			// otherwise an rTimeout larger than codecStallTimeout would delay
+			// stall detection by up to a whole poll interval.
+			if stall := t.codecStallTimeout; r.progress && stall > 0 {
+				if remaining := stall - time.Since(r.lastDelivery); remaining < timeout {
+					timeout = max(remaining, time.Millisecond)
+				}
+			}
+			// Set Deadline every time, since golang has fixed the performance issue
+			// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
+			currentTime := time.Now()
+			if err := t.conn.SetReadDeadline(currentTime.Add(timeout)); err != nil {
+				return err
+			}
+			t.rLastDeadline.Store(currentTime)
+		}
+
+		n, err := t.conn.Read(r.buf)
+		if n > 0 {
+			// data first: an error that fired while data was pending (even a
+			// real one) reoccurs on the next conn.Read.
+			r.next, r.end = 0, n
+			return nil
+		}
+		if err == nil {
+			continue
+		}
+		if !isTimeoutError(err) {
+			// a real error keeps its identity - io.EOF above all.
+			return err
+		}
+		if t.codecBroken.Load() || t.sessionClosing() {
+			// shutdown wakeup (session.stop arms a deadline for exactly this),
+			// codecReadError passes it through as a plain timeout.
+			return err
+		}
+		if stall := t.codecStallTimeout; r.progress && stall > 0 && time.Since(r.lastDelivery) >= stall {
+			// mid-block stall: surface the timeout, the codec latches it and
+			// codecReadError declares the stream broken.
+			return err
+		}
+		// idle poll timeout: absorb and keep waiting.
+	}
+}
+
+// codecPollingWriter is the write-side twin of codecPollingReader: the codec
+// writers (flate/snappy) latch the first error just like the readers, and a
+// hard wTimeout over a whole compressed burst would break connections whose
+// peer is merely slow. So write deadlines are owned here, wTimeout acts as a
+// poll interval, and a timeout is absorbed as long as the last attempt made
+// progress recently; only zero progress for codecStallTimeout (a peer that
+// genuinely stopped reading) or a shutdown surfaces the timeout, which the
+// codec latches and codecIOError turns into ErrCodecStreamBroken.
+//
+// A partial conn.Write on timeout is resumed from the exact position, which is
+// only safe because this layer owns the byte position - callers above the
+// codec could never retry a partial write without desynchronizing the stream.
+type codecPollingWriter struct {
+	t *gettyTCPConn
+}
+
+func (w *codecPollingWriter) Write(p []byte) (int, error) {
+	t := w.t
+	written := 0
+	lastProgress := time.Now()
+	for written < len(p) {
+		if timeout := t.wTimeout.Load(); timeout > 0 {
+			// the poll must wake up no later than the stall deadline, otherwise
+			// a wTimeout larger than codecStallTimeout would delay stall
+			// detection by up to a whole poll interval.
+			if stall := t.codecStallTimeout; stall > 0 {
+				if remaining := stall - time.Since(lastProgress); remaining < timeout {
+					timeout = max(remaining, time.Millisecond)
+				}
+			}
+			// Set Deadline every time, since golang has fixed the performance issue
+			// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
+			currentTime := time.Now()
+			if err := t.conn.SetWriteDeadline(currentTime.Add(timeout)); err != nil {
+				return written, err
+			}
+			t.wLastDeadline.Store(currentTime)
+		}
+
+		n, err := t.conn.Write(p[written:])
+		written += n
+		if n > 0 {
+			lastProgress = time.Now()
+		}
+		if err == nil {
+			continue
+		}
+		if !isTimeoutError(err) {
+			return written, err
+		}
+		if t.codecBroken.Load() || t.closing.Load() || t.sessionClosing() {
+			// on shutdown the write is abandoned mid-block; the connection is
+			// being torn down anyway, and latching codecBroken here also stops
+			// CloseConn from flushing the half-written snappy stream.
+			return written, err
+		}
+		if stall := t.codecStallTimeout; stall > 0 && time.Since(lastProgress) >= stall {
+			// the peer accepted nothing for a whole stall window: it stopped
+			// reading for good, surface the timeout and break the stream.
+			return written, err
+		}
+		// slow but progressing peer: absorb the timeout and keep writing.
+	}
+	return written, nil
+}
+
+// readDeadlineTimeout returns the deadline recv arms before a read. Only a raw
+// conn needs one (session.handleTCPPackage retries its timeouts as a poll); on
+// a codec stream the deadlines are owned by codecPollingReader underneath the
+// decoder, so recv itself must not arm any.
+func (t *gettyTCPConn) readDeadlineTimeout() time.Duration {
+	if t.codecEnabled {
+		return 0
+	}
+	return t.rTimeout.Load()
+}
+
+// beginRecv/beginSend mark the stream as started and snapshot the codec state,
+// so the blocking IO below runs without the lock and cannot race with
+// SetCompressType.
+func (t *gettyTCPConn) beginRecv() (io.Reader, *codecPollingReader, time.Duration) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	t.streamStarted = true
+	return t.reader, t.pollReader, t.readDeadlineTimeout()
+}
+
+func (t *gettyTCPConn) beginSend() (io.Writer, bool) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	t.streamStarted = true
+	return t.writer, t.codecEnabled
+}
+
+// codecIOError maps a timeout on a codec stream to the fatal
+// ErrCodecStreamBroken and latches the connection as broken.
+//
+// A timed out read leaves half a block in the decoder and misaligns every byte
+// after it; a timed out write leaves half a block on the wire and misaligns the
+// peer's decoder. Either way the stream cannot be reused, so the timeout is
+// reported as ErrCodecStreamBroken instead of as a net.Error: that is what makes
+// session.handleTCPPackage treat it as fatal and close the session (a client
+// then reconnects) rather than retry the read.
+//
+// The socket is closed as well, so the codec connection is terminated rather
+// than left half-alive: it unblocks any concurrent Read/Write and lets the
+// session tear down even when the stalled side is the writer. CloseConn is not
+// used here because closing the snappy writer can block on the very peer that
+// stalled; closing the raw conn is enough, and CloseConn skips the broken
+// writer when it eventually runs.
+//
+// Any other failure keeps its own identity - io.EOF above all, which the session
+// read loop matches on to detect a clean peer shutdown. Those need no latch:
+// flate/snappy record the failure internally and emit nothing more, every
+// further Read/Write just returns it again.
+func (t *gettyTCPConn) codecIOError(err error) error {
+	if err == nil || !t.codecEnabled || !isTimeoutError(err) {
+		return perrors.WithStack(err)
+	}
+
+	t.codecBroken.Store(true)
+	_ = t.conn.Close()
+	return perrors.Wrapf(ErrCodecStreamBroken, "codec stream stalled: %v", err)
+}
+
+// codecReadError is codecIOError for the read path, where a timeout raised while
+// the session is closing must be passed through: session.stop() deliberately
+// arms a read deadline to unblock this goroutine, and that timeout is a shutdown
+// signal, not a stalled peer. Latching it would make every normal close of a
+// compressed session report an error to the listener. The write path has no such
+// exemption - a timed out write desynchronizes the peer whether we are closing
+// or not.
+func (t *gettyTCPConn) codecReadError(err error) error {
+	if t.codecEnabled && isTimeoutError(err) && t.sessionClosing() {
+		return perrors.WithStack(err)
+	}
+	return t.codecIOError(err)
+}
+
+func (t *gettyTCPConn) sessionClosing() bool {
+	// t.ss is set by newSession before any IO goroutine starts; it is nil only
+	// for a connection used without a session (unit tests).
+	ss := t.ss
+	return ss != nil && ss.IsClosed()
+}
+
+func isTimeoutError(err error) bool {
+	var netErr net.Error
+	if errors.As(err, &netErr) && netErr.Timeout() {
+		return true
+	}
+	return errors.Is(err, os.ErrDeadlineExceeded)
+}
+
 // tcp connection read
 func (t *gettyTCPConn) recv(p []byte) (int, error) {
 	var (
@@ -301,21 +736,33 @@ func (t *gettyTCPConn) recv(p []byte) (int, error) {
 		length      int
 	)
 
-	// set read timeout deadline
-	if t.compress == CompressNone && t.rTimeout.Load() > 0 {
+	if t.codecBroken.Load() {
+		return 0, perrors.WithStack(ErrCodecStreamBroken)
+	}
+
+	reader, poller, timeout := t.beginRecv()
+
+	// set read timeout deadline (raw conn only; a codec stream's deadlines are
+	// owned by codecPollingReader, see readDeadlineTimeout)
+	if timeout > 0 {
 		// Set Deadline every time, since golang has fixed the performance issue
 		// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
 		currentTime = time.Now()
-		if err = t.conn.SetReadDeadline(currentTime.Add(t.rTimeout.Load())); err != nil {
+		if err = t.conn.SetReadDeadline(currentTime.Add(timeout)); err != nil {
 			// just a timeout error
 			return 0, perrors.WithStack(err)
 		}
 		t.rLastDeadline.Store(currentTime)
 	}
 
-	length, err = t.reader.Read(p)
+	length, err = reader.Read(p)
+	if poller != nil && length > 0 && err == nil {
+		// decoded output was produced: the stream is at a block boundary,
+		// silence from here on is idleness rather than a mid-block stall.
+		poller.boundary()
+	}
 	t.readBytes.Add(uint32(length))
-	return length, perrors.WithStack(err)
+	return length, t.codecReadError(err)
 }
 
 // tcp connection write
@@ -329,7 +776,18 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 		lg          int64
 	)
 
-	if t.compress == CompressNone && t.wTimeout.Load() > 0 {
+	if t.codecBroken.Load() {
+		return 0, perrors.WithStack(ErrCodecStreamBroken)
+	}
+
+	writer, codecEnabled := t.beginSend()
+
+	// A raw conn arms the write deadline here, so SetWriteTimeout - and the
+	// per-call WritePkg(pkg, timeout) - actually bounds a write to a stalled
+	// peer. On a codec stream the deadlines are owned by codecPollingWriter
+	// underneath the codec (wTimeout is its poll interval there), which
+	// distinguishes a slow-but-progressing peer from one that stopped reading.
+	if !codecEnabled && t.wTimeout.Load() > 0 {
 		// Set Deadline every time, since golang has fixed the performance issue
 		// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
 		currentTime = time.Now()
@@ -340,17 +798,19 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 	}
 
 	if buffers, ok := pkg.([][]byte); ok {
-		// #102: when compression is enabled the [][]byte path must go through
-		// t.writer (the compress writer), otherwise it writes raw frames
-		// directly to t.conn and the peer receives a corrupt mix of
-		// compressed and uncompressed data.
-		if t.compress == CompressNone {
+		// #102: when a codec is installed the [][]byte path must go through
+		// t.writer (the codec writer), otherwise it writes raw frames directly
+		// to t.conn and the peer receives a corrupt mix of coded and raw data.
+		if !codecEnabled {
+			// only a raw conn here, so writev the whole batch in one syscall.
 			netBuf := net.Buffers(buffers)
 			lg, err = netBuf.WriteTo(t.conn)
+		} else if bw, ok := writer.(buffersWriter); ok {
+			lg, err = bw.WriteBuffers(buffers)
 		} else {
 			for _, b := range buffers {
 				var n int
-				n, err = t.writer.Write(b)
+				n, err = writer.Write(b)
 				if err != nil {
 					break
 				}
@@ -361,20 +821,20 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 			t.writeBytes.Add((uint32)(lg))
 			t.writePkgNum.Add((uint32)(len(buffers)))
 		}
-		log.Debugf("localAddr: %s, remoteAddr:%s, now:%s, length:%d, err:%s",
-			t.conn.LocalAddr(), t.conn.RemoteAddr(), currentTime, length, err)
-		return int(lg), perrors.WithStack(err)
+		log.Debugf("localAddr: %s, remoteAddr:%s, length:%d, err:%v",
+			t.conn.LocalAddr(), t.conn.RemoteAddr(), lg, err)
+		return int(lg), t.codecIOError(err)
 	}
 
 	if p, ok = pkg.([]byte); ok {
-		length, err = t.writer.Write(p)
+		length, err = writer.Write(p)
 		if err == nil {
 			t.writeBytes.Add((uint32)(len(p)))
 			t.writePkgNum.Add(1)
 		}
-		log.Debugf("localAddr: %s, remoteAddr:%s, now:%s, length:%d, err:%v",
-			t.conn.LocalAddr(), t.conn.RemoteAddr(), currentTime, length, err)
-		return length, perrors.WithStack(err)
+		log.Debugf("localAddr: %s, remoteAddr:%s, length:%d, err:%v",
+			t.conn.LocalAddr(), t.conn.RemoteAddr(), length, err)
+		return length, t.codecIOError(err)
 	}
 
 	return 0, perrors.Errorf("illegal @pkg{%#v} type", pkg)
@@ -382,15 +842,29 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 
 // close tcp connection
 func (t *gettyTCPConn) CloseConn(waitSec int) {
-	// if tcpConn, ok := t.conn.(*net.TCPConn); ok {
-	// tcpConn.SetLinger(0)
-	// }
-
-	if t.conn != nil {
-		// #102: snappy writer is now wrapped in *snappyWriteFlusher.
-		if writer, ok := t.writer.(*snappyWriteFlusher); ok {
-			if err := writer.Close(); err != nil {
-				log.Errorf("snappy.Writer.Close() = error:%+v", err)
+	t.closeOnce.Do(func() {
+		// best-effort teardown from here on: codecPollingWriter surfaces the
+		// first poll timeout instead of waiting out a stall window, so a peer
+		// that stopped reading cannot pin CloseConn for codecStallTimeout.
+		t.closing.Store(true)
+		t.lock.Lock()
+		writer := t.writer
+		t.lock.Unlock()
+		// #102: the codec writers are wrapped in flushers with a Close that
+		// terminates the stream, giving the peer's decoder a clean EOF.
+		// A broken codec must not be flushed: the stream is already
+		// desynchronized, and Close would only push more garbage into a socket
+		// that may still be stalled.
+		if !t.codecBroken.Load() {
+			switch w := writer.(type) {
+			case *snappyWriteFlusher:
+				if err := w.Close(); err != nil {
+					log.Errorf("snappy.Writer.Close() = error:%+v", err)
+				}
+			case *writeFlusher:
+				if err := w.Close(); err != nil {
+					log.Errorf("flate.Writer.Close() = error:%+v", err)
+				}
 			}
 		}
 		// #103: do not hard-assert *tls.Conn; use safe type assertions so a
@@ -403,8 +877,7 @@ func (t *gettyTCPConn) CloseConn(waitSec int) {
 		} else {
 			_ = t.conn.Close()
 		}
-		t.conn = nil
-	}
+	})
 }
 
 // ///////////////////////////////////////
@@ -422,8 +895,8 @@ func (c UDPContext) String() string {
 
 type gettyUDPConn struct {
 	gettyConn
-	compressType CompressType
-	conn         *net.UDPConn // for server
+	conn      *net.UDPConn // for server; immutable after construction, closed via closeOnce
+	closeOnce sync.Once
 }
 
 // create gettyUDPConn
@@ -455,10 +928,19 @@ func newGettyUDPConn(conn *net.UDPConn) *gettyUDPConn {
 	}
 }
 
+// SetCompressType records the requested type but UDP send/recv never
+// compress: each datagram is an independent packet with no stream to run a
+// codec over, and this implementation never had one. Accepting the call
+// silently made callers believe compression was on, so any type other than
+// CompressNone is now loudly reported as unsupported (still accepted, not a
+// panic, to keep existing callers running).
 func (u *gettyUDPConn) SetCompressType(c CompressType) {
 	switch c {
 	case CompressNone, CompressZip, CompressBestSpeed, CompressBestCompression, CompressHuffman, CompressSnappy:
-		u.compressType = c
+		if c != CompressNone {
+			log.Warnf("UDP connection{local:%s, peer:%s} does not support compression, SetCompressType(%d) has no effect on the wire", u.local, u.peer, c)
+		}
+		u.compress = c
 
 	default:
 		panic(fmt.Sprintf("illegal comparess type %d", c))
@@ -532,10 +1014,9 @@ func (u *gettyUDPConn) Send(udpCtx any) (int, error) {
 
 // close udp connection
 func (u *gettyUDPConn) CloseConn(_ int) {
-	if u.conn != nil {
+	u.closeOnce.Do(func() {
 		_ = u.conn.Close()
-		u.conn = nil
-	}
+	})
 }
 
 // ///////////////////////////////////////
@@ -546,7 +1027,12 @@ type gettyWSConn struct {
 	gettyConn
 	writeLock sync.Mutex
 	readLock  sync.Mutex
-	conn      *websocket.Conn
+	// streamStarted is set by the first read/write and freezes the compression
+	// configuration, mirroring the gettyTCPConn contract: gorilla's
+	// EnableWriteCompression/SetCompressionLevel are plain field writes that
+	// must not race with in-flight writers.
+	streamStarted uatomic.Bool
+	conn          *websocket.Conn
 }
 
 // create websocket connection
@@ -581,18 +1067,29 @@ func newGettyWSConn(conn *websocket.Conn) *gettyWSConn {
 	return gettyWSConn
 }
 
-// SetCompressType set compress type
+// SetCompressType set compress type. Like the TCP variant it must be called
+// before the first recv/Send, typically in NewSessionCallback, and panics on a
+// started stream: gorilla's compression setters are plain field writes that
+// would race with in-flight writers.
 func (w *gettyWSConn) SetCompressType(c CompressType) {
 	switch c {
 	case CompressNone, CompressZip, CompressBestSpeed, CompressBestCompression, CompressHuffman:
-		w.conn.EnableWriteCompression(true)
-		if err := w.conn.SetCompressionLevel(int(c)); err != nil {
-			log.Warnf("failed to set compression level: %+v", err)
-		}
-
 	default:
 		panic(fmt.Sprintf("illegal comparess type %d", c))
 	}
+	// writeLock excludes in-flight writers; the started check under it makes
+	// the panic race-free (writers mark streamStarted while holding the same
+	// lock, readers mark it under readLock before touching the conn).
+	w.writeLock.Lock()
+	defer w.writeLock.Unlock()
+	if w.streamStarted.Load() {
+		log.Errorf("SetCompressType(%d) called after IO started on connection{local:%s, peer:%s}", c, w.local, w.peer)
+		panic("SetCompressType must be called before any recv/Send on the connection, e.g. in NewSessionCallback")
+	}
+	w.conn.EnableWriteCompression(true)
+	if err := w.conn.SetCompressionLevel(int(c)); err != nil {
+		log.Warnf("failed to set compression level: %+v", err)
+	}
 	w.compress = c
 }
 
@@ -669,11 +1166,14 @@ func (w *gettyWSConn) Send(pkg any) (int, error) {
 	if err := w.updateWriteDeadline(); err != nil {
 		log.Warnf("failed to update write deadline: %+v", err)
 	}
-	if err = w.threadSafeWriteMessage(websocket.BinaryMessage, p); err == nil {
-		w.writeBytes.Add((uint32)(len(p)))
-		w.writePkgNum.Add(1)
+	// a failed WriteMessage delivers nothing (gorilla discards the frame), so
+	// report 0 written bytes: callers treat the count as the success count.
+	if err = w.threadSafeWriteMessage(websocket.BinaryMessage, p); err != nil {
+		return 0, perrors.WithStack(err)
 	}
-	return len(p), perrors.WithStack(err)
+	w.writeBytes.Add((uint32)(len(p)))
+	w.writePkgNum.Add(1)
+	return len(p), nil
 }
 
 func (w *gettyWSConn) writePing() error {
@@ -713,6 +1213,7 @@ func (w *gettyWSConn) CloseConn(waitSec int) {
 func (w *gettyWSConn) threadSafeWriteMessage(messageType int, data []byte) error {
 	w.writeLock.Lock()
 	defer w.writeLock.Unlock()
+	w.streamStarted.Store(true)
 	if err := w.conn.WriteMessage(messageType, data); err != nil {
 		return err
 	}
@@ -723,6 +1224,7 @@ func (w *gettyWSConn) threadSafeWriteMessage(messageType int, data []byte) error
 func (w *gettyWSConn) threadSafeReadMessage() (int, []byte, error) {
 	w.readLock.Lock()
 	defer w.readLock.Unlock()
+	w.streamStarted.Store(true)
 	messageType, readBytes, err := w.conn.ReadMessage()
 	if err != nil {
 		return messageType, nil, err
diff --git a/transport/connection_test.go b/transport/connection_test.go
index ca601f3e..5dc56820 100644
--- a/transport/connection_test.go
+++ b/transport/connection_test.go
@@ -18,8 +18,10 @@
 package getty
 
 import (
+	"bytes"
 	"compress/flate"
 	"errors"
+	"fmt"
 	"io"
 	"net"
 	"sync"
@@ -29,6 +31,8 @@ import (
 
 import (
 	"github.com/golang/snappy"
+
+	perrors "github.com/pkg/errors"
 )
 
 type blockingSnappyWriter struct {
@@ -167,3 +171,874 @@ func TestSnappyWriteFlusherCloseWaitsForWrite(t *testing.T) {
 		t.Fatalf("Close failed: %v", err)
 	}
 }
+
+func newTCPConnPair(t *testing.T) (*gettyTCPConn, *gettyTCPConn) {
+	t.Helper()
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = listener.Close() }()
+
+	type acceptResult struct {
+		conn net.Conn
+		err  error
+	}
+	accepted := make(chan acceptResult, 1)
+	go func() {
+		conn, err := listener.Accept()
+		accepted <- acceptResult{conn: conn, err: err}
+	}()
+
+	clientRaw, err := net.Dial("tcp", listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	result := <-accepted
+	if result.err != nil {
+		_ = clientRaw.Close()
+		t.Fatal(result.err)
+	}
+
+	client := newGettyTCPConn(clientRaw)
+	server := newGettyTCPConn(result.conn)
+	t.Cleanup(func() {
+		client.CloseConn(0)
+		server.CloseConn(0)
+	})
+	return client, server
+}
+
+// startReceiver continuously decodes from conn until wantLen bytes have been
+// read. A tiny buffer forces reads to cross codec block boundaries.
+func startReceiver(conn *gettyTCPConn, wantLen int, bufSize int) (<-chan []byte, <-chan error) {
+	gotCh := make(chan []byte, 1)
+	errCh := make(chan error, 1)
+	go func() {
+		var got []byte
+		buf := make([]byte, bufSize)
+		for len(got) < wantLen {
+			n, err := conn.recv(buf)
+			if err != nil {
+				errCh <- err
+				return
+			}
+			if n == 0 {
+				continue
+			}
+			got = append(got, buf[:n]...)
+		}
+		gotCh <- got
+	}()
+	return gotCh, errCh
+}
+
+// TestSendMixedSingleAndBatchOverCompressNoneCodec pins the #102/#107 wire
+// format: SetCompressType(CompressNone) installs a real flate codec, so every
+// Send path, including [][]byte, must go through the codec writer. Otherwise
+// one connection carries a corrupt mix of coded []byte sends and raw
+// [][]byte sends and the peer's decoder fails.
+func TestSendMixedSingleAndBatchOverCompressNoneCodec(t *testing.T) {
+	client, server := newTCPConnPair(t)
+	client.SetCompressType(CompressNone)
+	server.SetCompressType(CompressNone)
+
+	single := []byte("single-packet-")
+	batch := [][]byte{
+		[]byte("batch-part-0-"),
+		[]byte("batch-part-1-"),
+		[]byte("batch-part-2"),
+	}
+	tail := []byte("-tail")
+	expected := bytes.Join([][]byte{
+		single,
+		batch[0],
+		batch[1],
+		batch[2],
+		tail,
+	}, nil)
+
+	gotCh, errCh := startReceiver(server, len(expected), 3)
+	for _, pkg := range []any{single, batch, tail} {
+		if _, err := client.Send(pkg); err != nil {
+			t.Fatalf("Send(%T) failed: %v", pkg, err)
+		}
+	}
+
+	select {
+	case got := <-gotCh:
+		if !bytes.Equal(got, expected) {
+			t.Fatalf("decoded stream mismatch:\n got: %q\nwant: %q", got, expected)
+		}
+	case err := <-errCh:
+		t.Fatalf("peer failed to decode codec stream: %v", err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("timed out waiting for the decoded stream")
+	}
+}
+
+// TestSendBatchRawWithoutCompressType keeps the raw writev path covered: when
+// SetCompressType was never called, a [][]byte send must reach the peer
+// untouched (net.Buffers.WriteTo(t.conn), a single writev on *net.TCPConn).
+func TestSendBatchRawWithoutCompressType(t *testing.T) {
+	client, server := newTCPConnPair(t)
+	if client.codecEnabled {
+		t.Fatal("new connection unexpectedly has a codec installed")
+	}
+
+	batch := [][]byte{
+		[]byte("raw-part-0-"),
+		[]byte("raw-part-1-"),
+		[]byte("raw-part-2"),
+	}
+	expected := bytes.Join(batch, nil)
+
+	gotCh, errCh := startReceiver(server, len(expected), 3)
+	if _, err := client.Send(batch); err != nil {
+		t.Fatalf("Send([][]byte) failed: %v", err)
+	}
+
+	select {
+	case got := <-gotCh:
+		if !bytes.Equal(got, expected) {
+			t.Fatalf("raw stream mismatch:\n got: %q\nwant: %q", got, expected)
+		}
+	case err := <-errCh:
+		t.Fatalf("peer failed to read raw stream: %v", err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("timed out waiting for the raw stream")
+	}
+}
+
+// codecStallCases covers both codec families: they take different branches in
+// SetCompressType and both latch IO errors, so the stall handling has to hold
+// for each.
+var codecStallCases = []struct {
+	name     string
+	compress CompressType
+}{
+	{name: "flate", compress: CompressZip},
+	{name: "snappy", compress: CompressSnappy},
+}
+
+// halfCodecBlock returns the first half of a valid, flushed codec block: a prefix
+// the decoder cannot finish decoding, i.e. exactly what a peer that dies
+// mid-write leaves behind.
+func halfCodecBlock(t *testing.T, c CompressType, payload string) []byte {
+	t.Helper()
+
+	var (
+		buf    bytes.Buffer
+		writer interface {
+			io.Writer
+			Flush() error
+		}
+	)
+	if c == CompressSnappy {
+		writer = snappy.NewBufferedWriter(&buf)
+	} else {
+		flateWriter, err := flate.NewWriter(&buf, int(c))
+		if err != nil {
+			t.Fatal(err)
+		}
+		writer = flateWriter
+	}
+	if _, err := writer.Write([]byte(payload)); err != nil {
+		t.Fatal(err)
+	}
+	if err := writer.Flush(); err != nil {
+		t.Fatal(err)
+	}
+	block := buf.Bytes()
+	if len(block) < 4 {
+		t.Fatalf("codec block too small to truncate: %d bytes", len(block))
+	}
+	return block[:len(block)/2]
+}
+
+// assertCodecStreamBroken checks the error a stalled codec stream must produce.
+// The last two assertions are the actual encoding of the fix: session.handleTCPPackage
+// classifies read errors by perrors.Cause(), retrying net.Error timeouts and
+// treating io.EOF as a clean peer shutdown. A stalled codec stream must fall
+// into neither bucket, otherwise the session keeps reading from a decoder that
+// has already latched the error.
+func assertCodecStreamBroken(t *testing.T, err error) {
+	t.Helper()
+
+	if err == nil {
+		t.Fatal("stalled codec stream returned no error")
+	}
+	if !errors.Is(err, ErrCodecStreamBroken) {
+		t.Fatalf("error = %v, want ErrCodecStreamBroken", err)
+	}
+	cause := perrors.Cause(err)
+	if netErr, ok := cause.(net.Error); ok && netErr.Timeout() {
+		t.Fatalf("cause %v is a retryable net.Error timeout; the session would keep using the dead codec", cause)
+	}
+	if cause == io.EOF {
+		t.Fatalf("cause %v would be treated as a clean peer shutdown", cause)
+	}
+}
+
+// TestCodecRecvStalledPeerBreaksStream covers the P1: a peer that sends half a
+// codec block and then goes silent must not block the reader forever. The read
+// deadline has to reach the socket even though a codec is installed, and the
+// resulting timeout must terminate the connection instead of being retried on a
+// decoder that has already latched the error.
+func TestCodecRecvStalledPeerBreaksStream(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			client, server := newTCPConnPair(t)
+			client.SetReadTimeout(20 * time.Millisecond)
+			client.SetCompressType(test.compress)
+			client.codecStallTimeout = 200 * time.Millisecond
+
+			if _, err := server.conn.Write(halfCodecBlock(t, test.compress, "stalled-peer-payload")); err != nil {
+				t.Fatalf("peer write failed: %v", err)
+			}
+			// the peer now stalls: no more bytes, and the connection stays open.
+
+			// recv runs in a goroutine so a regression (no deadline on the codec
+			// stream) fails the test in seconds instead of hanging until the test
+			// binary panics.
+			recvErr := make(chan error, 1)
+			go func() {
+				buf := make([]byte, 64)
+				for callsLeft := 1000; callsLeft > 0; callsLeft-- {
+					if _, err := client.recv(buf); err != nil {
+						recvErr <- err
+						return
+					}
+				}
+				recvErr <- nil
+			}()
+
+			var err error
+			select {
+			case err = <-recvErr:
+			case <-time.After(3 * time.Second):
+				t.Fatal("recv never returned: the read deadline did not reach the socket")
+			}
+
+			assertCodecStreamBroken(t, err)
+			if !client.codecBroken.Load() {
+				t.Fatal("connection was not latched as broken")
+			}
+			// the codec connection must be terminated, not just marked unusable:
+			// the peer has to observe the close instead of a silent open socket.
+			if err := client.conn.SetReadDeadline(time.Now()); err == nil {
+				t.Fatal("underlying conn is still open after a stalled read")
+			}
+			_ = server.conn.SetReadDeadline(time.Now().Add(time.Second))
+			if _, err := server.conn.Read(make([]byte, 8)); err == nil {
+				t.Fatal("peer read still succeeded after the codec connection was terminated")
+			}
+			// a broken stream must fail fast instead of touching the codec again
+			if _, err := client.recv(make([]byte, 64)); !errors.Is(err, ErrCodecStreamBroken) {
+				t.Fatalf("recv after break = %v, want ErrCodecStreamBroken", err)
+			}
+			if _, err := client.Send([]byte("nope")); !errors.Is(err, ErrCodecStreamBroken) {
+				t.Fatalf("Send after break = %v, want ErrCodecStreamBroken", err)
+			}
+		})
+	}
+}
+
+// TestCodecSendStalledPeerBreaksStream covers the write half of the P1: a peer
+// that stops reading must not block Send forever. net.Pipe is unbuffered, so a
+// write blocks until the peer reads - exactly the stalled-reader condition.
+func TestCodecSendStalledPeerBreaksStream(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			clientRaw, peerRaw := net.Pipe()
+			t.Cleanup(func() {
+				_ = clientRaw.Close()
+				_ = peerRaw.Close()
+			})
+
+			client := newGettyTCPConn(clientRaw)
+			client.SetWriteTimeout(50 * time.Millisecond)
+			client.SetCompressType(test.compress)
+			// zero write progress for this long means the peer stopped reading
+			client.codecStallTimeout = 200 * time.Millisecond
+
+			sendErr := make(chan error, 1)
+			go func() {
+				_, err := client.Send([]byte("peer never reads this"))
+				sendErr <- err
+			}()
+
+			var err error
+			select {
+			case err = <-sendErr:
+			case <-time.After(3 * time.Second):
+				t.Fatal("Send never returned: the write deadline did not reach the socket")
+			}
+			assertCodecStreamBroken(t, err)
+			if !client.codecBroken.Load() {
+				t.Fatal("connection was not latched as broken")
+			}
+			// the codec connection must be terminated: the stalled peer has to
+			// observe the close instead of a still-open pipe.
+			if err := client.conn.SetWriteDeadline(time.Now()); err == nil {
+				t.Fatal("underlying conn is still open after a stalled write")
+			}
+			if _, err := peerRaw.Write([]byte("ping")); err == nil {
+				t.Fatal("peer write still succeeded after the codec connection was terminated")
+			}
+
+			// the second Send must fail fast rather than block on the dead codec
+			start := time.Now()
+			if _, err = client.Send([]byte("still nope")); !errors.Is(err, ErrCodecStreamBroken) {
+				t.Fatalf("Send after break = %v, want ErrCodecStreamBroken", err)
+			}
+			if elapsed := time.Since(start); elapsed > 100*time.Millisecond {
+				t.Fatalf("Send after break blocked for %s, want an immediate failure", elapsed)
+			}
+		})
+	}
+}
+
+// TestCodecRecvIdleThenResume is the regression guard for the deadline value: a
+// codec stream must survive an idle period of many read timeouts. The read
+// timeout is a poll interval (session.handleTCPPackage retries it), so arming it
+// on a codec stream would kill every idle compressed connection - flate/snappy
+// latch the timeout and never decode again.
+func TestCodecRecvIdleThenResume(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			client, server := newTCPConnPair(t)
+			client.SetReadTimeout(20 * time.Millisecond)
+			server.SetWriteTimeout(time.Second)
+			client.SetCompressType(test.compress)
+			server.SetCompressType(test.compress)
+			client.codecStallTimeout = 2 * time.Second
+
+			const idle = 200 * time.Millisecond // 10 read timeouts
+			payload := []byte("packet-after-idle")
+			writeErr := make(chan error, 1)
+			go func() {
+				time.Sleep(idle)
+				_, err := server.Send(payload)
+				writeErr <- err
+			}()
+
+			got := make([]byte, 0, len(payload))
+			buf := make([]byte, 64)
+			for len(got) < len(payload) {
+				n, err := client.recv(buf)
+				if err != nil {
+					t.Fatalf("recv after %s of idling failed: %v", idle, err)
+				}
+				got = append(got, buf[:n]...)
+			}
+			if err := <-writeErr; err != nil {
+				t.Fatalf("peer Send failed: %v", err)
+			}
+			if !bytes.Equal(got, payload) {
+				t.Fatalf("decoded %q, want %q", got, payload)
+			}
+			if client.codecBroken.Load() {
+				t.Fatal("an idle codec stream was wrongly latched as broken")
+			}
+		})
+	}
+}
+
+// TestCodecRecvIdleBeyondStallTimeoutStaysHealthy pins the stall/idle
+// distinction: CodecStallTimeout only applies to a stream that stalled in the
+// middle of a codec block. A connection that never received a byte, or that is
+// idle between fully decoded packets, must survive silence far beyond the
+// stall timeout and resume normally - previously any silence longer than the
+// timeout was misclassified as a broken stream and the socket was closed.
+func TestCodecRecvIdleBeyondStallTimeoutStaysHealthy(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			client, server := newTCPConnPair(t)
+			client.SetReadTimeout(20 * time.Millisecond)
+			client.SetCompressType(test.compress)
+			client.codecStallTimeout = 100 * time.Millisecond
+			server.SetWriteTimeout(time.Second)
+			server.SetCompressType(test.compress)
+
+			payload := []byte("packet-after-long-idle")
+			expected := bytes.Repeat(payload, 2)
+			gotCh, errCh := startReceiver(client, len(expected), 8)
+
+			// silence with zero bytes ever received, 4x the stall timeout
+			const idle = 400 * time.Millisecond
+			select {
+			case err := <-errCh:
+				t.Fatalf("connection that never received a byte was killed after idling: %v", err)
+			case <-time.After(idle):
+			}
+			if client.codecBroken.Load() {
+				t.Fatal("never-used codec stream was latched as broken by pure idleness")
+			}
+			if _, err := server.Send(payload); err != nil {
+				t.Fatalf("peer Send after idle failed: %v", err)
+			}
+
+			// idle again between two fully decoded packets, then resume
+			select {
+			case err := <-errCh:
+				t.Fatalf("connection idling between packets was killed: %v", err)
+			case <-time.After(idle):
+			}
+			if client.codecBroken.Load() {
+				t.Fatal("codec stream idling between packets was latched as broken")
+			}
+			if _, err := server.Send(payload); err != nil {
+				t.Fatalf("peer Send after second idle failed: %v", err)
+			}
+
+			select {
+			case got := <-gotCh:
+				if !bytes.Equal(got, expected) {
+					t.Fatalf("decoded stream mismatch:\n got: %q\nwant: %q", got, expected)
+				}
+			case err := <-errCh:
+				t.Fatalf("peer failed to decode after idle periods: %v", err)
+			case <-time.After(10 * time.Second):
+				t.Fatal("timed out waiting for the decoded stream")
+			}
+			if client.codecBroken.Load() {
+				t.Fatal("healthy idle connection ended up latched as broken")
+			}
+		})
+	}
+}
+
+// TestCodecRecvStallAcrossRecvBoundaryBreaksStream pins the exactness of the
+// stall detection: the peer flushes one complete packet plus the first half of
+// the next block in a single burst, then dies. The remainder is delivered to
+// the decoder from the poller's own read-ahead buffer after the first packet
+// decoded, so the following silence must still be classified as a mid-block
+// stall - with a hidden bufio between poller and decoder this case was
+// indistinguishable from idleness and hung until session close.
+func TestCodecRecvStallAcrossRecvBoundaryBreaksStream(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			client, server := newTCPConnPair(t)
+			client.SetReadTimeout(20 * time.Millisecond)
+			client.SetCompressType(test.compress)
+			client.codecStallTimeout = 200 * time.Millisecond
+
+			// one codec stream: full flushed block for payload1, then payload2's
+			// block truncated in half - exactly what a peer that dies mid-write
+			// leaves after a healthy packet.
+			payload1 := []byte("complete-first-packet")
+			var (
+				buf    bytes.Buffer
+				writer interface {
+					io.Writer
+					Flush() error
+				}
+			)
+			if test.compress == CompressSnappy {
+				writer = snappy.NewBufferedWriter(&buf)
+			} else {
+				flateWriter, err := flate.NewWriter(&buf, int(test.compress))
+				if err != nil {
+					t.Fatal(err)
+				}
+				writer = flateWriter
+			}
+			if _, err := writer.Write(payload1); err != nil {
+				t.Fatal(err)
+			}
+			if err := writer.Flush(); err != nil {
+				t.Fatal(err)
+			}
+			firstLen := buf.Len()
+			if _, err := writer.Write([]byte("second-packet-that-never-finishes")); err != nil {
+				t.Fatal(err)
+			}
+			if err := writer.Flush(); err != nil {
+				t.Fatal(err)
+			}
+			second := buf.Bytes()[firstLen:]
+			if len(second) < 4 {
+				t.Fatalf("second codec block too small to truncate: %d bytes", len(second))
+			}
+			burst := buf.Bytes()[:firstLen+len(second)/2]
+
+			if _, err := server.conn.Write(burst); err != nil {
+				t.Fatalf("peer write failed: %v", err)
+			}
+			// the peer now dies: no more bytes, connection stays open.
+
+			recvErr := make(chan error, 1)
+			go func() {
+				var got []byte
+				buf := make([]byte, 64)
+				for {
+					n, err := client.recv(buf)
+					if err != nil {
+						recvErr <- err
+						return
+					}
+					got = append(got, buf[:n]...)
+					if len(got) > len(payload1) {
+						recvErr <- fmt.Errorf("decoded beyond the first packet: %q", got)
+						return
+					}
+				}
+			}()
+
+			var err error
+			select {
+			case err = <-recvErr:
+			case <-time.After(3 * time.Second):
+				t.Fatal("recv never returned: the cross-recv stall was classified as idleness")
+			}
+			assertCodecStreamBroken(t, err)
+			if !client.codecBroken.Load() {
+				t.Fatal("connection was not latched as broken")
+			}
+		})
+	}
+}
+
+// TestCodecSendSlowPeerBeyondWriteTimeoutSurvives is the write-side twin of the
+// idle/stall distinction: a peer that drains slowly but steadily must not be
+// killed just because one compressed burst takes longer than wTimeout - only
+// zero progress for codecStallTimeout may break the stream.
+func TestCodecSendSlowPeerBeyondWriteTimeoutSurvives(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			clientRaw, peerRaw := net.Pipe()
+			t.Cleanup(func() {
+				_ = clientRaw.Close()
+				_ = peerRaw.Close()
+			})
+
+			client := newGettyTCPConn(clientRaw)
+			client.SetWriteTimeout(20 * time.Millisecond)
+			client.SetCompressType(test.compress)
+			client.codecStallTimeout = 500 * time.Millisecond
+
+			// the peer drains a few bytes at a time, far slower than wTimeout
+			// allows for the whole burst, but never stops for a stall window.
+			peerDone := make(chan struct{})
+			go func() {
+				defer close(peerDone)
+				buf := make([]byte, 8)
+				for {
+					if _, err := peerRaw.Read(buf); err != nil {
+						return
+					}
+					time.Sleep(30 * time.Millisecond)
+				}
+			}()
+
+			payload := bytes.Repeat([]byte("slow-but-alive-"), 20) // 300 bytes
+			sendErr := make(chan error, 1)
+			go func() {
+				_, err := client.Send(payload)
+				sendErr <- err
+			}()
+
+			select {
+			case err := <-sendErr:
+				if err != nil {
+					t.Fatalf("Send to a slow but draining peer failed: %v", err)
+				}
+			case <-time.After(10 * time.Second):
+				t.Fatal("Send never completed against a slow but draining peer")
+			}
+			if client.codecBroken.Load() {
+				t.Fatal("slow but draining peer was latched as a broken stream")
+			}
+			_ = clientRaw.Close()
+			<-peerDone
+		})
+	}
+}
+
+// TestRawConnRecvTimeoutStaysRetryable pins the other half of the contract: on a
+// raw connection a read timeout is still a benign, retryable net.Error - it is
+// the poll that lets session.handleTCPPackage notice a closed session.
+func TestRawConnRecvTimeoutStaysRetryable(t *testing.T) {
+	client, _ := newTCPConnPair(t)
+	client.SetReadTimeout(50 * time.Millisecond)
+
+	_, err := client.recv(make([]byte, 64))
+	if err == nil {
+		t.Fatal("recv on a silent peer returned no error")
+	}
+	if errors.Is(err, ErrCodecStreamBroken) {
+		t.Fatalf("raw conn read timeout reported as a broken codec stream: %v", err)
+	}
+	netErr, ok := perrors.Cause(err).(net.Error)
+	if !ok || !netErr.Timeout() {
+		t.Fatalf("cause = %v, want a net.Error timeout", perrors.Cause(err))
+	}
+	if client.codecBroken.Load() {
+		t.Fatal("raw conn was latched as broken")
+	}
+}
+
+// TestCodecRecvPeerCloseKeepsErrorIdentity pins the deliberate scope of the
+// latch: only a timeout means "stalled mid-stream". A clean peer close must keep
+// its own EOF-family error, which session.handleTCPPackage matches on to skip
+// reconnecting, and must not latch the connection - otherwise a concurrent
+// WritePkg would report ErrCodecStreamBroken instead of the real cause.
+func TestCodecRecvPeerCloseKeepsErrorIdentity(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			client, server := newTCPConnPair(t)
+			client.SetReadTimeout(time.Second)
+			client.SetCompressType(test.compress)
+			server.CloseConn(0)
+
+			_, err := client.recv(make([]byte, 64))
+			if err == nil {
+				t.Fatal("recv on a closed peer returned no error")
+			}
+			if errors.Is(err, ErrCodecStreamBroken) {
+				t.Fatalf("clean peer close reported as a stalled codec stream: %v", err)
+			}
+			if client.codecBroken.Load() {
+				t.Fatal("clean peer close latched the connection as broken")
+			}
+		})
+	}
+}
+
+// TestSetCompressTypeAfterIORejected pins the codec configuration contract:
+// once a connection has sent or received, SetCompressType must panic instead
+// of replacing the codec under in-flight IO and desynchronizing the peer.
+func TestSetCompressTypeAfterIORejected(t *testing.T) {
+	client, server := newTCPConnPair(t)
+
+	payload := []byte("started")
+	gotCh, errCh := startReceiver(server, len(payload), 8)
+	if _, err := client.Send(payload); err != nil {
+		t.Fatalf("Send failed: %v", err)
+	}
+	select {
+	case <-gotCh:
+	case err := <-errCh:
+		t.Fatalf("peer failed to read: %v", err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("timed out waiting for the payload")
+	}
+
+	for name, conn := range map[string]*gettyTCPConn{"sender": client, "receiver": server} {
+		func() {
+			defer func() {
+				if recover() == nil {
+					t.Fatalf("SetCompressType on a started %s did not panic", name)
+				}
+			}()
+			conn.SetCompressType(CompressSnappy)
+		}()
+	}
+}
+
+// TestSetCompressTypeConcurrentWithSend is the race regression for the PR#107
+// review: SetCompressType(CompressSnappy) racing with Send([]byte) on a started
+// stream must be rejected, stay race-free under `go test -race` and leave the
+// raw stream intact.
+func TestSetCompressTypeConcurrentWithSend(t *testing.T) {
+	client, server := newTCPConnPair(t)
+
+	payload := []byte("payload-")
+	const sends = 100
+	expected := bytes.Repeat(payload, sends+1)
+	gotCh, errCh := startReceiver(server, len(expected), 16)
+
+	// start the stream, so the SetCompressType below is guaranteed to be late
+	if _, err := client.Send(payload); err != nil {
+		t.Fatalf("Send failed: %v", err)
+	}
+
+	start := make(chan struct{})
+	panicked := make(chan bool, 1)
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		<-start
+		for i := 0; i < sends; i++ {
+			if _, err := client.Send(payload); err != nil {
+				t.Errorf("Send failed: %v", err)
+				return
+			}
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		defer func() { panicked <- recover() != nil }()
+		<-start
+		client.SetCompressType(CompressSnappy)
+	}()
+	close(start)
+	wg.Wait()
+
+	if !<-panicked {
+		t.Fatal("late SetCompressType was not rejected")
+	}
+	if client.codecEnabled {
+		t.Fatal("rejected SetCompressType still installed a codec")
+	}
+	select {
+	case got := <-gotCh:
+		if !bytes.Equal(got, expected) {
+			t.Fatalf("stream corrupted:\n got: %q\nwant: %q", got, expected)
+		}
+	case err := <-errCh:
+		t.Fatalf("peer failed to read: %v", err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("timed out waiting for the stream")
+	}
+}
+
+// TestUDPSetCompressTypeHasNoWireEffect pins the UDP contract: compression is
+// not supported, the call records the type (and warns) instead of silently
+// pretending or panicking under existing callers.
+func TestUDPSetCompressTypeHasNoWireEffect(t *testing.T) {
+	raw, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
+	if err != nil {
+		t.Fatal(err)
+	}
+	conn := newGettyUDPConn(raw)
+	t.Cleanup(func() { conn.CloseConn(0) })
+
+	conn.SetCompressType(CompressSnappy)
+	if conn.compress != CompressSnappy {
+		t.Fatalf("compress = %d, want %d recorded", conn.compress, CompressSnappy)
+	}
+}
+
+// TestCloseConnCodecFlushIsBounded pins the closing flag: CloseConn flushes the
+// codec writer best-effort, so a peer that stopped reading must cost at most
+// one poll timeout, never a whole codecStallTimeout.
+func TestCloseConnCodecFlushIsBounded(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			clientRaw, peerRaw := net.Pipe()
+			t.Cleanup(func() {
+				_ = clientRaw.Close()
+				_ = peerRaw.Close()
+			})
+
+			client := newGettyTCPConn(clientRaw)
+			client.SetWriteTimeout(20 * time.Millisecond)
+			client.SetCompressType(test.compress)
+			client.codecStallTimeout = 5 * time.Second
+
+			done := make(chan struct{})
+			go func() {
+				client.CloseConn(0)
+				close(done)
+			}()
+			select {
+			case <-done:
+			case <-time.After(2 * time.Second):
+				t.Fatal("CloseConn blocked on the codec flush instead of giving up after one poll timeout")
+			}
+		})
+	}
+}
+
+// TestCloseConnTerminatesFlateStreamCleanly pins the write-side symmetry with
+// snappy: CloseConn closes the flate writer too, so the peer's decoder reads
+// the payload and then a clean EOF instead of io.ErrUnexpectedEOF.
+func TestCloseConnTerminatesFlateStreamCleanly(t *testing.T) {
+	client, server := newTCPConnPair(t)
+	client.SetWriteTimeout(time.Second)
+	client.SetCompressType(CompressZip)
+
+	payload := []byte("last-packet-before-close")
+	if _, err := client.Send(payload); err != nil {
+		t.Fatalf("Send failed: %v", err)
+	}
+	client.CloseConn(1)
+
+	if err := server.conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil {
+		t.Fatal(err)
+	}
+	got, err := io.ReadAll(flate.NewReader(server.conn))
+	if err != nil {
+		t.Fatalf("peer decoder did not see a clean end of stream: %v", err)
+	}
+	if !bytes.Equal(got, payload) {
+		t.Fatalf("decoded %q, want %q", got, payload)
+	}
+}
+
+// TestCodecRecvStallDetectionOutrunsLongPollInterval pins the poll clamp: when
+// the stream is mid-block, the poll deadline is capped at the remaining stall
+// window, so an rTimeout larger than codecStallTimeout (here 10s vs 200ms) must
+// not delay stall detection until the next full poll interval.
+func TestCodecRecvStallDetectionOutrunsLongPollInterval(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			client, server := newTCPConnPair(t)
+			client.SetReadTimeout(10 * time.Second)
+			client.SetCompressType(test.compress)
+			client.codecStallTimeout = 200 * time.Millisecond
+
+			if _, err := server.conn.Write(halfCodecBlock(t, test.compress, "stalled-peer-payload")); err != nil {
+				t.Fatalf("peer write failed: %v", err)
+			}
+
+			recvErr := make(chan error, 1)
+			go func() {
+				buf := make([]byte, 64)
+				for callsLeft := 1000; callsLeft > 0; callsLeft-- {
+					if _, err := client.recv(buf); err != nil {
+						recvErr <- err
+						return
+					}
+				}
+				recvErr <- nil
+			}()
+
+			var err error
+			select {
+			case err = <-recvErr:
+			case <-time.After(3 * time.Second):
+				t.Fatal("stall not detected within 3s: the poll interval was not clamped to the stall deadline")
+			}
+			assertCodecStreamBroken(t, err)
+		})
+	}
+}
+
+// TestCodecSendStallDetectionOutrunsLongPollInterval is the write-side twin of
+// the poll clamp test: a peer that never reads must be detected within the
+// stall window even when wTimeout is far larger than codecStallTimeout.
+func TestCodecSendStallDetectionOutrunsLongPollInterval(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			clientRaw, peerRaw := net.Pipe()
+			t.Cleanup(func() {
+				_ = clientRaw.Close()
+				_ = peerRaw.Close()
+			})
+
+			client := newGettyTCPConn(clientRaw)
+			client.SetWriteTimeout(10 * time.Second)
+			client.SetCompressType(test.compress)
+			client.codecStallTimeout = 200 * time.Millisecond
+
+			sendErr := make(chan error, 1)
+			go func() {
+				_, err := client.Send([]byte("peer never reads this"))
+				sendErr <- err
+			}()
+
+			var err error
+			select {
+			case err = <-sendErr:
+			case <-time.After(3 * time.Second):
+				t.Fatal("stall not detected within 3s: the poll interval was not clamped to the stall deadline")
+			}
+			assertCodecStreamBroken(t, err)
+		})
+	}
+}
diff --git a/transport/connection_ws_test.go b/transport/connection_ws_test.go
new file mode 100644
index 00000000..e129dc68
--- /dev/null
+++ b/transport/connection_ws_test.go
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package getty
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+import (
+	"github.com/gorilla/websocket"
+)
+
+func newWSClientConn(t *testing.T) *gettyWSConn {
+	t.Helper()
+
+	upgrader := websocket.Upgrader{}
+	serverConnCh := make(chan *websocket.Conn, 1)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		serverConnCh <- conn
+	}))
+	t.Cleanup(srv.Close)
+
+	clientWS, resp, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(srv.URL, "http"), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if resp != nil && resp.Body != nil {
+		_ = resp.Body.Close()
+	}
+	serverWS := <-serverConnCh
+	t.Cleanup(func() {
+		_ = clientWS.Close()
+		_ = serverWS.Close()
+	})
+	return newGettyWSConn(clientWS)
+}
+
+// TestWSSendReturnsZeroOnError pins the write count contract: a failed
+// WriteMessage delivers nothing, so Send must report 0 written bytes -
+// returning len(p) alongside the error inflates the caller's success count
+// (session.WritePkg uses it as successCount).
+func TestWSSendReturnsZeroOnError(t *testing.T) {
+	client := newWSClientConn(t)
+
+	payload := []byte("ws-payload")
+	if n, err := client.Send(payload); err != nil || n != len(payload) {
+		t.Fatalf("healthy Send = (%d, %v), want (%d, nil)", n, err, len(payload))
+	}
+	if got := client.writePkgNum.Load(); got != 1 {
+		t.Fatalf("writePkgNum = %d after one successful Send, want 1", got)
+	}
+
+	_ = client.conn.UnderlyingConn().Close()
+	n, err := client.Send(payload)
+	if err == nil {
+		t.Fatal("Send on a closed websocket returned no error")
+	}
+	if n != 0 {
+		t.Fatalf("failed Send reported %d written bytes, want 0", n)
+	}
+	if got := client.writePkgNum.Load(); got != 1 {
+		t.Fatalf("failed Send bumped writePkgNum to %d, want it unchanged at 1", got)
+	}
+}