diff --git a/context.go b/context.go
index df9b3aa..07213a7 100644
--- a/context.go
+++ b/context.go
@@ -60,6 +60,12 @@ var (
 	// ContextKeyKeepAlive is a context key for use with Contexts in this package.
 	// The associated value will be of type *SessionKeepAlive.
 	ContextKeyKeepAlive = &contextKey{"keep-alive"}
+
+	// contextKeyOpenChannels is an internal context key for the
+	// per-connection *openChannelSet used by connection keep-alive
+	// bookkeeping. Unexported because the value type is unexported and
+	// external consumers should not depend on this detail.
+	contextKeyOpenChannels = &contextKey{"open-channels"}
 )
 
 // Context is a package specific context interface. It exposes connection
diff --git a/keepalive.go b/keepalive.go
index 77a7321..e675d8d 100644
--- a/keepalive.go
+++ b/keepalive.go
@@ -71,6 +71,21 @@ func (ska *SessionKeepAlive) Reset() {
 	}
 }
 
+// notePeerActivity records inbound traffic from the peer. It bumps
+// lastReceived (clearing the dead-peer deadline used by TimeIsUp) and
+// resets the ticker so the next probe fires `interval` after the most
+// recent activity. Matches OpenSSH sshd, which clears keep_alive_timeouts
+// on every successfully-received packet and defers the next probe on
+// inbound traffic. Internal — driven by the package's request loops.
+func (ska *SessionKeepAlive) notePeerActivity() {
+	ska.m.Lock()
+	defer ska.m.Unlock()
+	if ska.ticker != nil && !ska.closed {
+		ska.lastReceived = time.Now()
+		ska.ticker.Reset(ska.clientAliveInterval)
+	}
+}
+
 // Ticks returns the channel that fires on each keep-alive interval.
 func (ska *SessionKeepAlive) Ticks() <-chan time.Time {
 	return ska.tickerCh
diff --git a/server.go b/server.go
index 6f3d43f..a95acc3 100644
--- a/server.go
+++ b/server.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
 	"net"
 	"sync"
 	"time"
@@ -15,6 +16,128 @@ import (
 // and ListenAndServeTLS methods after a call to Shutdown or Close.
 var ErrServerClosed = errors.New("ssh: Server closed")
 
+// openChannelSet tracks accepted channels for a connection so that
+// connectionKeepAlive can mirror OpenSSH's client_alive_check() behavior:
+// when at least one channel is open, send the keepalive as a channel
+// request on that channel; otherwise fall back to a global request.
+type openChannelSet struct {
+	mu    sync.Mutex
+	chans []gossh.Channel
+}
+
+func (s *openChannelSet) add(c gossh.Channel) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.chans = append(s.chans, c)
+}
+
+func (s *openChannelSet) any() gossh.Channel {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if len(s.chans) == 0 {
+		return nil
+	}
+	return s.chans[0]
+}
+
+func (s *openChannelSet) remove(c gossh.Channel) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for i, ch := range s.chans {
+		if ch == c {
+			s.chans = append(s.chans[:i], s.chans[i+1:]...)
+			return
+		}
+	}
+}
+
+// trackingNewChannel wraps a gossh.NewChannel so that successful Accept
+// calls register the underlying channel with an openChannelSet. The
+// ChannelHandler API receives a gossh.NewChannel and typically calls
+// Accept() inside the handler, so wrapping at HandleConn dispatch time is
+// the only place to observe acceptance for arbitrary external handlers
+// without modifying them.
+//
+// The per-channel request stream returned by Accept is also wrapped so
+// that close of the underlying gossh stream (channel teardown) drops the
+// channel from the openChannelSet, and so each inbound request bumps the
+// keep-alive activity marker.
+type trackingNewChannel struct {
+	gossh.NewChannel
+	set              *openChannelSet
+	onAccept         func(gossh.Channel)
+	onClose          func(gossh.Channel)
+	notePeerActivity func()
+	ctx              Context
+}
+
+// trackingChanReqBuffer matches gossh's per-channel request channel
+// buffer (see chanSize in golang.org/x/crypto/ssh/handshake.go). Keeping
+// the size in sync avoids changing back-pressure semantics for handlers
+// that did not previously block on a full request channel.
+const trackingChanReqBuffer = 16
+
+func (t *trackingNewChannel) Accept() (gossh.Channel, <-chan *gossh.Request, error) {
+	ch, reqs, err := t.NewChannel.Accept()
+	if err != nil {
+		return ch, reqs, err
+	}
+	if t.onAccept != nil {
+		t.onAccept(ch)
+	}
+	wrapped := make(chan *gossh.Request, trackingChanReqBuffer)
+	go func() {
+		defer close(wrapped)
+		defer func() {
+			if t.onClose != nil {
+				t.onClose(ch)
+			}
+		}()
+		ctxDone := t.ctx.Done()
+		for r := range reqs {
+			if t.notePeerActivity != nil {
+				t.notePeerActivity()
+			}
+			select {
+			case wrapped <- r:
+			case <-ctxDone:
+				// Handler stopped draining and the connection is
+				// going away. Negatively reply to any want-reply
+				// request we were holding, then drain upstream so
+				// gossh's per-channel request goroutine doesn't
+				// leak, replying false to any subsequent
+				// want-reply requests as we go.
+				if r.WantReply {
+					_ = r.Reply(false, nil)
+				}
+				for r2 := range reqs {
+					if r2.WantReply {
+						_ = r2.Reply(false, nil)
+					}
+				}
+				return
+			}
+		}
+	}()
+	return ch, wrapped, nil
+}
+
+// NewChannelUnwrapper is implemented by NewChannel implementations that
+// wrap another NewChannel for internal bookkeeping. Channel handlers
+// that need access to the underlying gossh.NewChannel (e.g., to type-
+// assert against a custom implementation) can call Unwrap to recover
+// it. This is needed because the library wraps every incoming
+// NewChannel to track per-channel keep-alive activity; the wrapper is
+// otherwise transparent.
+type NewChannelUnwrapper interface {
+	Unwrap() gossh.NewChannel
+}
+
+// Unwrap returns the underlying gossh.NewChannel. See NewChannelUnwrapper.
+func (t *trackingNewChannel) Unwrap() gossh.NewChannel {
+	return t.NewChannel
+}
+
 // SubsystemHandler is a callback for handling SSH subsystem requests.
 type SubsystemHandler func(s Session)
 
@@ -63,6 +186,7 @@ type Server struct {
 	// succeed, never both.
 	ConnectionFailedCallback   ConnectionFailedCallback   // callback to report connection failures
 	ConnectionCompleteCallback ConnectionCompleteCallback // callback to report connection completion
+	ConnectionClosingCallback  ConnectionClosingCallback  // see ConnectionClosingCallback type doc
 
 	IdleTimeout time.Duration // connection timeout when no activity, none if empty
 	MaxTimeout  time.Duration // absolute connection timeout, none if empty
@@ -70,6 +194,12 @@ type Server struct {
 	// ChannelHandlers allow overriding the built-in session handlers or provide
 	// extensions to the protocol, such as tcpip forwarding. By default only the
 	// "session" handler is enabled.
+	//
+	// The gossh.NewChannel value passed to handlers may be wrapped by this
+	// package for keep-alive bookkeeping (tracking open channels and noting
+	// inbound activity). Handlers that need access to the unwrapped
+	// underlying value can type-assert to NewChannelUnwrapper and call
+	// Unwrap.
 	ChannelHandlers map[string]ChannelHandler
 
 	// RequestHandlers allow overriding the server-level request handlers or
@@ -323,6 +453,29 @@ func (srv *Server) HandleConn(newConn net.Conn) {
 	applyConnMetadata(ctx, sshConn)
 	// To prevent race conditions, we need to configure the keep-alive before goroutines kick off
 	applyKeepAlive(ctx, srv.ClientAliveInterval, srv.ClientAliveCountMax)
+	openChans := &openChannelSet{}
+	ctx.SetValue(contextKeyOpenChannels, openChans)
+
+	// Connection-level keep-alive: runs for the lifetime of the transport,
+	// independent of whether any session is active. This is what detects a
+	// dead transport between sessions (e.g., an idle ControlMaster whose
+	// outer ssh has been -O exit'd but whose ProxyCommand chain hasn't
+	// propagated EOF). On timeout we close sshConn so HandleConn's
+	// `for ch := range chans` loop unblocks and the closing/complete
+	// callbacks can fire.
+	keepAliveDone := make(chan struct{})
+	go srv.connectionKeepAlive(ctx, sshConn, keepAliveDone)
+	// LIFO: close(keepAliveDone) runs first to signal connectionKeepAlive
+	// to return, then KeepAlive().Close() stops the ticker deterministically.
+	// Without the explicit Close, the ticker keeps firing on a dropped
+	// channel until ctx (held by spawned handler goroutines and the
+	// per-channel forwarder) becomes unreachable and GC reclaims it.
+	// All SessionKeepAlive methods are mutex-protected, so a straggler
+	// goroutine calling notePeerActivity/Reset after Close just sees
+	// closed=true and no-ops on the ticker.
+	defer ctx.KeepAlive().Close()
+	defer close(keepAliveDone)
+
 	// go gossh.DiscardRequests(reqs)
 	go srv.handleRequests(ctx, reqs)
 	for ch := range chans {
@@ -334,12 +487,117 @@ func (srv *Server) HandleConn(newConn net.Conn) {
 			_ = ch.Reject(gossh.UnknownChannelType, "unsupported channel type")
 			continue
 		}
-		go handler(srv, sshConn, ch, ctx)
+		tracked := &trackingNewChannel{
+			NewChannel: ch,
+			set:        openChans,
+			onAccept:   openChans.add,
+			onClose:    openChans.remove,
+			notePeerActivity: func() {
+				if ka := ctx.KeepAlive(); ka != nil {
+					ka.notePeerActivity()
+				}
+			},
+			ctx: ctx,
+		}
+		go handler(srv, sshConn, tracked, ctx)
+	}
+
+	// Fire the closing callback synchronously, before any deferred cleanup
+	// runs, so downstream callers have a deterministic hook the moment the
+	// channels stream ends. Deferred hooks that block on the transport
+	// (e.g., waiting on the mux) may be delayed indefinitely on a stuck
+	// transport; this path is not.
+	if srv.ConnectionClosingCallback != nil {
+		srv.ConnectionClosingCallback(ctx, sshConn)
+	}
+}
+
+// connectionKeepAlive drives transport-level keep-alive pings for the life
+// of sshConn. It mirrors OpenSSH's client_alive_check(): if at least one
+// channel is open, the keepalive is sent as a SSH2_MSG_CHANNEL_REQUEST on
+// that channel; otherwise it falls back to a SSH2_MSG_GLOBAL_REQUEST. After
+// ClientAliveCountMax consecutive intervals with no successful reply,
+// sshConn is closed so HandleConn unblocks. Stops when `done` closes
+// (HandleConn returning).
+func (srv *Server) connectionKeepAlive(
+	ctx Context,
+	sshConn *gossh.ServerConn,
+	done <-chan struct{},
+) {
+	countMax := srv.ClientAliveCountMax
+	if srv.ClientAliveInterval <= 0 || countMax <= 0 {
+		return
+	}
+
+	// Reuse the SessionKeepAlive already stashed on ctx so request-handler
+	// resets (KeepAliveRequestHandler) and metrics keep working. The ticker
+	// is Stop()ped from HandleConn's defer chain after this goroutine has
+	// been signaled to return.
+	keepAlive := ctx.KeepAlive()
+
+	openChans := ctx.Value(contextKeyOpenChannels).(*openChannelSet)
+
+	inFlight := make(chan struct{}, 1)
+	for {
+		select {
+		case <-done:
+			return
+		case <-keepAlive.Ticks():
+			if keepAlive.TimeIsUp() {
+				log.Printf(
+					"ssh: connection keep-alive timeout after %d intervals; closing transport",
+					countMax,
+				)
+				if err := sshConn.Close(); err != nil {
+					log.Printf("ssh: failed to close stalled transport: %v", err)
+				}
+				return
+			}
+			select {
+			case inFlight <- struct{}{}:
+			default:
+				continue
+			}
+			go func() {
+				defer func() { <-inFlight }()
+				keepAlive.ServerRequestedKeepAliveCallback()
+				// Mirror OpenSSH client_alive_check(): prefer a channel
+				// request on an open channel; fall back to a global
+				// request if no channel is open or the channel send
+				// fails (channel was closed mid-flight).
+				//
+				// No outer timeout is needed here: the inFlight semaphore
+				// already prevents overlapping probes, TimeIsUp() at the
+				// next tick enforces the deadline, and if SendRequest
+				// hangs forever it will be unblocked when the TimeIsUp
+				// branch closes sshConn.
+				var err error
+				ch := openChans.any()
+				if ch != nil {
+					_, err = ch.SendRequest(keepAliveRequestType, true, nil)
+					if err != nil {
+						openChans.remove(ch)
+						ch = nil
+					}
+				}
+				if ch == nil {
+					_, _, err = sshConn.SendRequest(keepAliveRequestType, true, nil)
+				}
+				if err == nil {
+					keepAlive.Reset()
+				} else {
+					log.Printf("ssh: keepalive request failed: %v", err)
+				}
+			}()
+		}
 	}
 }
 
 func (srv *Server) handleRequests(ctx Context, in <-chan *gossh.Request) {
 	for req := range in {
+		if ka := ctx.KeepAlive(); ka != nil {
+			ka.notePeerActivity()
+		}
 		handler := srv.RequestHandlers[req.Type]
 		if handler == nil {
 			handler = srv.RequestHandlers["default"]
diff --git a/server_test.go b/server_test.go
index 3cc0ab7..298acc8 100644
--- a/server_test.go
+++ b/server_test.go
@@ -4,8 +4,12 @@ import (
 	"bytes"
 	"context"
 	"io"
+	"net"
+	"sync/atomic"
 	"testing"
 	"time"
+
+	gossh "golang.org/x/crypto/ssh"
 )
 
 func TestAddHostKey(t *testing.T) {
@@ -80,6 +84,463 @@ func TestServerShutdown(t *testing.T) {
 	}
 }
 
+// TestConnectionClosingCallback verifies the closing callback fires
+// synchronously when HandleConn observes the channels stream close, even
+// when the client disconnects abruptly. It must fire before
+// ConnectionCompleteCallback (which blocks on sshConn.Wait()).
+func TestConnectionClosingCallback(t *testing.T) {
+	t.Parallel()
+
+	// Use a single ordered channel so a regression that swapped the order of
+	// the two callbacks would be caught. Reading from independent channels
+	// with separate timeouts would not detect such a swap.
+	events := make(chan string, 2)
+	var closingCtx atomic.Value // ssh.Context observed at callback time
+
+	srv := &Server{
+		Handler: func(s Session) {
+			_, _ = io.WriteString(s, "hi")
+		},
+		ConnectionClosingCallback: func(ctx Context, _ *gossh.ServerConn) {
+			closingCtx.Store(ctx)
+			events <- "closing"
+		},
+		ConnectionCompleteCallback: func(_ *gossh.ServerConn, _ error) {
+			events <- "complete"
+		},
+	}
+
+	l := newLocalTCPListener()
+	go func() { _ = srv.serveOnce(l) }()
+
+	sess, client, _ := newClientSession(t, l.Addr().String(), nil)
+	if err := sess.Run(""); err != nil && err != io.EOF {
+		t.Fatalf("session run: %v", err)
+	}
+	_ = sess.Close()
+	_ = client.Close()
+
+	readEvent := func() string {
+		select {
+		case ev := <-events:
+			return ev
+		case <-time.After(2 * time.Second):
+			t.Fatal("timed out waiting for callback event")
+			return ""
+		}
+	}
+
+	if first := readEvent(); first != "closing" {
+		t.Fatalf("expected first event to be \"closing\", got %q", first)
+	}
+	if second := readEvent(); second != "complete" {
+		t.Fatalf("expected second event to be \"complete\", got %q", second)
+	}
+
+	if got := closingCtx.Load(); got == nil {
+		t.Fatal("ConnectionClosingCallback received nil Context")
+	}
+}
+
+// TestConnectionKeepAliveClosesStalledConn is the regression test for the
+// connection-level keep-alive: when a client stops responding to keepalive
+// global requests (e.g., idle ControlMaster whose peer is wedged) and no
+// session is open, the server's connectionKeepAlive goroutine must close
+// sshConn after roughly ClientAliveInterval * ClientAliveCountMax, which
+// unblocks HandleConn's `for ch := range chans` loop and fires
+// ConnectionClosingCallback. Before the fix, the keep-alive was only driven
+// from within an active session, so an idle-but-stuck transport would
+// linger forever.
+//
+// We dial with a raw gossh.NewClientConn and intentionally do NOT reply to
+// the incoming requests channel, simulating a stuck peer. We do not open a
+// session — the bug is specifically about the no-session case.
+func TestConnectionKeepAliveClosesStalledConn(t *testing.T) {
+	t.Parallel()
+
+	closingFired := make(chan struct{})
+
+	srv := &Server{
+		Handler:             func(_ Session) {},
+		ClientAliveInterval: 100 * time.Millisecond,
+		ClientAliveCountMax: 2,
+		ConnectionClosingCallback: func(_ Context, _ *gossh.ServerConn) {
+			close(closingFired)
+		},
+	}
+
+	l := newLocalTCPListener()
+	defer func() { _ = l.Close() }()
+	go func() { _ = srv.serveOnce(l) }()
+
+	cfg := &gossh.ClientConfig{
+		User:            "testuser",
+		Auth:            []gossh.AuthMethod{gossh.Password("testpass")},
+		HostKeyCallback: gossh.InsecureIgnoreHostKey(), //nolint:gosec // test code
+	}
+	netConn, err := net.Dial("tcp", l.Addr().String())
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	sshConn, chans, reqs, err := gossh.NewClientConn(netConn, l.Addr().String(), cfg)
+	if err != nil {
+		t.Fatalf("NewClientConn: %v", err)
+	}
+	defer func() { _ = sshConn.Close() }()
+
+	// Drain both streams WITHOUT replying. By not calling req.Reply on
+	// incoming global requests (including keepalive@openssh.com), we
+	// simulate a stuck peer. The server's SendRequest with wantReply=true
+	// will not see a response and connectionKeepAlive will eventually
+	// close sshConn after ClientAliveCountMax intervals.
+	go func() {
+		for range chans { //nolint:revive // intentional drain
+		}
+	}()
+	go func() {
+		for req := range reqs {
+			_ = req // intentionally never reply
+		}
+	}()
+
+	// 100ms * 2 = 200ms expected; allow generous slack for CI scheduling
+	// and any per-SendRequest timeout the other agent may layer on.
+	select {
+	case <-closingFired:
+	case <-time.After(5 * time.Second):
+		t.Fatal("ConnectionClosingCallback did not fire; " +
+			"stalled client was not torn down by connection-level keep-alive")
+	}
+}
+
+// TestConnectionKeepAliveClosesStalledConnWithOpenSession is the counterpart
+// to TestConnectionKeepAliveClosesStalledConn: when a session channel is open
+// (so probes go as channel requests per client_alive_check), and the peer
+// stops replying to those probes, the connection-level keep-alive must still
+// detect the stall and tear the transport down. This exercises the
+// channel-request path of the deadline check; the no-session variant exercises
+// the global-request path.
+func TestConnectionKeepAliveClosesStalledConnWithOpenSession(t *testing.T) {
+	t.Parallel()
+
+	closingFired := make(chan struct{})
+
+	srv := &Server{
+		Handler:             func(s Session) { <-s.Context().Done() },
+		ClientAliveInterval: 100 * time.Millisecond,
+		ClientAliveCountMax: 3,
+		ConnectionClosingCallback: func(_ Context, _ *gossh.ServerConn) {
+			close(closingFired)
+		},
+	}
+
+	l := newLocalTCPListener()
+	defer func() { _ = l.Close() }()
+	go func() { _ = srv.serveOnce(l) }()
+
+	cfg := &gossh.ClientConfig{
+		User:            "testuser",
+		Auth:            []gossh.AuthMethod{gossh.Password("testpass")},
+		HostKeyCallback: gossh.InsecureIgnoreHostKey(), //nolint:gosec // test code
+	}
+	netConn, err := net.Dial("tcp", l.Addr().String())
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	sshConn, chans, reqs, err := gossh.NewClientConn(netConn, l.Addr().String(), cfg)
+	if err != nil {
+		t.Fatalf("NewClientConn: %v", err)
+	}
+	defer func() { _ = sshConn.Close() }()
+
+	ch, chReqs, err := sshConn.OpenChannel("session", nil)
+	if err != nil {
+		t.Fatalf("OpenChannel: %v", err)
+	}
+	defer func() { _ = ch.Close() }()
+
+	// Drain all three request streams (global, per-channel, and the
+	// NewChannel stream) WITHOUT replying. The server's channel-typed
+	// keepalive probes go on chReqs and will never get a reply, so
+	// notePeerActivity is never called on the session-request side and
+	// TimeIsUp eventually fires.
+	go func() {
+		for range chans { //nolint:revive // intentional drain
+		}
+	}()
+	go func() {
+		for req := range reqs {
+			_ = req
+		}
+	}()
+	go func() {
+		for req := range chReqs {
+			_ = req
+		}
+	}()
+
+	// 100ms * 3 = 300ms expected; allow generous slack for CI.
+	select {
+	case <-closingFired:
+	case <-time.After(5 * time.Second):
+		t.Fatal("ConnectionClosingCallback did not fire; " +
+			"stalled-with-session client was not torn down by connection-level keep-alive")
+	}
+}
+
+// TestConnectionKeepAliveUsesChannelRequestWhenSessionOpen verifies that the
+// server's connection-level keepalive mirrors OpenSSH's client_alive_check():
+// while at least one channel is open it sends keepalive@openssh.com as a
+// channel request on that channel; after the channel closes it falls back
+// to a global request.
+func TestConnectionKeepAliveUsesChannelRequestWhenSessionOpen(t *testing.T) {
+	t.Parallel()
+
+	srv := &Server{
+		Handler: func(s Session) {
+			<-s.Context().Done()
+		},
+		ClientAliveInterval: 100 * time.Millisecond,
+		ClientAliveCountMax: 100, // generous so the conn doesn't get torn down mid-test
+	}
+
+	l := newLocalTCPListener()
+	defer func() { _ = l.Close() }()
+	go func() { _ = srv.serveOnce(l) }()
+
+	cfg := &gossh.ClientConfig{
+		User:            "testuser",
+		Auth:            []gossh.AuthMethod{gossh.Password("testpass")},
+		HostKeyCallback: gossh.InsecureIgnoreHostKey(), //nolint:gosec // test code
+	}
+	netConn, err := net.Dial("tcp", l.Addr().String())
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	sshConn, chans, globalReqs, err := gossh.NewClientConn(netConn, l.Addr().String(), cfg)
+	if err != nil {
+		t.Fatalf("NewClientConn: %v", err)
+	}
+	defer func() { _ = sshConn.Close() }()
+	go func() {
+		for range chans { //nolint:revive // intentional drain
+		}
+	}()
+
+	var globalCount, channelCount atomic.Int64
+	go func() {
+		for req := range globalReqs {
+			if req.Type == "keepalive@openssh.com" {
+				globalCount.Add(1)
+			}
+			_ = req.Reply(true, nil)
+		}
+	}()
+
+	ch, chReqs, err := sshConn.OpenChannel("session", nil)
+	if err != nil {
+		t.Fatalf("OpenChannel: %v", err)
+	}
+	chDone := make(chan struct{})
+	go func() {
+		defer close(chDone)
+		for req := range chReqs {
+			if req.Type == "keepalive@openssh.com" {
+				channelCount.Add(1)
+			}
+			if req.WantReply {
+				_ = req.Reply(true, nil)
+			}
+		}
+	}()
+
+	// Observe ~10 intervals with the channel open.
+	time.Sleep(1 * time.Second)
+	if got := channelCount.Load(); got < 1 {
+		t.Fatalf("expected >=1 channel-typed keepalive while channel open, got %d", got)
+	}
+	if got := globalCount.Load(); got != 0 {
+		t.Fatalf("expected 0 global-typed keepalives while channel open, got %d", got)
+	}
+
+	// Close the channel and observe the fallback to global requests.
+	_ = ch.Close()
+	<-chDone
+	before := globalCount.Load()
+	time.Sleep(1 * time.Second)
+	if got := globalCount.Load() - before; got < 1 {
+		t.Fatalf("expected >=1 global-typed keepalive after channel close, got %d", got)
+	}
+}
+
+// TestConnectionKeepAlivePrunesClosedChannels verifies that the
+// per-channel close hook prunes channels from the openChannelSet as
+// soon as the client closes them, BEFORE the next keepalive probe
+// fires. This is a regression test for the case where openChans.any()
+// returned a dead channel because per-channel removal only happened on
+// SendRequest failure (i.e., one probe was wasted on a dead channel
+// before the prune-on-failure path kicked in).
+//
+// Strategy: use a long ClientAliveInterval (3s) so we can observe the
+// state of openChans in the window between channel close and the first
+// post-close probe. We capture the openChannelSet from inside the
+// session handler so the test can inspect it without exporting helpers.
+func TestConnectionKeepAlivePrunesClosedChannels(t *testing.T) {
+	t.Parallel()
+
+	openChansCh := make(chan *openChannelSet, 1)
+	srv := &Server{
+		Handler: func(s Session) {
+			ctx := s.Context()
+			if oc, ok := ctx.Value(contextKeyOpenChannels).(*openChannelSet); ok {
+				select {
+				case openChansCh <- oc:
+				default:
+				}
+			}
+			<-ctx.Done()
+		},
+		ClientAliveInterval: 3 * time.Second,
+		ClientAliveCountMax: 100, // generous so the conn isn't torn down mid-test
+	}
+
+	l := newLocalTCPListener()
+	defer func() { _ = l.Close() }()
+	go func() { _ = srv.serveOnce(l) }()
+
+	cfg := &gossh.ClientConfig{
+		User:            "testuser",
+		Auth:            []gossh.AuthMethod{gossh.Password("testpass")},
+		HostKeyCallback: gossh.InsecureIgnoreHostKey(), //nolint:gosec // test code
+	}
+	netConn, err := net.Dial("tcp", l.Addr().String())
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	sshConn, chans, globalReqs, err := gossh.NewClientConn(netConn, l.Addr().String(), cfg)
+	if err != nil {
+		t.Fatalf("NewClientConn: %v", err)
+	}
+	defer func() { _ = sshConn.Close() }()
+	go func() {
+		for range chans { //nolint:revive // intentional drain
+		}
+	}()
+
+	var globalCount atomic.Int64
+	go func() {
+		for req := range globalReqs {
+			if req.Type == "keepalive@openssh.com" {
+				globalCount.Add(1)
+			}
+			_ = req.Reply(true, nil)
+		}
+	}()
+
+	// Open three sessions so that the server handler runs and registers
+	// channels in openChans. Use a "shell" request so DefaultSessionHandler
+	// considers the session "handled" and our top-level Handler runs.
+	var chReqWg [3]chan struct{}
+	channels := make([]gossh.Channel, 0, 3)
+	for i := 0; i < 3; i++ {
+		ch, chReqs, err := sshConn.OpenChannel("session", nil)
+		if err != nil {
+			t.Fatalf("OpenChannel[%d]: %v", i, err)
+		}
+		done := make(chan struct{})
+		chReqWg[i] = done
+		go func() {
+			defer close(done)
+			for req := range chReqs {
+				if req.WantReply {
+					_ = req.Reply(true, nil)
+				}
+			}
+		}()
+		// Send a shell request to trigger the user Handler so it can
+		// publish openChans into openChansCh.
+		if _, err := ch.SendRequest("shell", true, nil); err != nil {
+			t.Fatalf("shell request[%d]: %v", i, err)
+		}
+		channels = append(channels, ch)
+	}
+
+	// Grab the openChannelSet from the first session that ran.
+	var openChans *openChannelSet
+	select {
+	case openChans = <-openChansCh:
+	case <-time.After(2 * time.Second):
+		t.Fatalf("never received openChannelSet from handler")
+	}
+
+	// Sanity: all three channels are registered.
+	openChans.mu.Lock()
+	regBefore := len(openChans.chans)
+	openChans.mu.Unlock()
+	if regBefore != 3 {
+		t.Fatalf("expected 3 channels registered before close, got %d", regBefore)
+	}
+
+	// Close all channels.
+	for _, ch := range channels {
+		_ = ch.Close()
+	}
+	for i := 0; i < 3; i++ {
+		<-chReqWg[i]
+	}
+
+	// Poll for the close hook to drain. ClientAliveInterval is 3s, so we
+	// have ample time to observe the prune BEFORE the next probe fires.
+	// If the close hook works, len(openChans.chans) drops to 0 quickly
+	// (just goroutine scheduling). If the hook is broken, it stays at 3
+	// until the next keepalive tick fires and SendRequest fails.
+	//
+	// We use a 500ms poll deadline — well above goroutine-scheduling
+	// jitter but well below the 3s tick interval. This guarantees the
+	// assertion is unambiguous: it passes ONLY if the close hook ran,
+	// not because a keepalive tick happened to fire and the
+	// prune-on-failure path cleared the set.
+	deadline := time.Now().Add(500 * time.Millisecond)
+	for time.Now().Before(deadline) {
+		openChans.mu.Lock()
+		n := len(openChans.chans)
+		openChans.mu.Unlock()
+		if n == 0 {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	openChans.mu.Lock()
+	regAfter := len(openChans.chans)
+	openChans.mu.Unlock()
+	// Invariant: no keepalive tick should have fired during the 500ms
+	// window (interval is 3s). If one did, the test result is
+	// ambiguous — the prune may have happened via the
+	// SendRequest-failure path instead of the close hook.
+	if globalCount.Load() != 0 {
+		t.Fatalf(
+			"keepalive tick fired before assertion window (globalCount=%d); test invalid",
+			globalCount.Load(),
+		)
+	}
+	if regAfter != 0 {
+		t.Fatalf(
+			"openChannelSet still has %d entries 500ms after close; close hook did not run",
+			regAfter,
+		)
+	}
+
+	// And verify that the next keepalive probe (which fires ~3s after the
+	// last reset) goes out as a global request — confirming behavior
+	// end-to-end. Wait 4s to leave room for at least one tick after the
+	// 3s interval.
+	before := globalCount.Load()
+	time.Sleep(4 * time.Second)
+	if got := globalCount.Load() - before; got < 1 {
+		t.Fatalf("expected >=1 global-typed keepalive after channels closed, got %d", got)
+	}
+}
+
 func TestServerClose(t *testing.T) {
 	l := newLocalTCPListener()
 	s := &Server{
diff --git a/session.go b/session.go
index f21988b..2b9d56e 100644
--- a/session.go
+++ b/session.go
@@ -4,8 +4,6 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
-	"io"
-	"log"
 	"net"
 	"sync"
 
@@ -282,203 +280,172 @@ func (sess *session) Break(c chan<- bool) {
 	sess.breakCh = c
 }
 
-func (sess *session) handleRequests(ctx Context, reqs <-chan *gossh.Request) {
-	keepAlive := ctx.KeepAlive()
-	defer keepAlive.Close()
-
-	var keepAliveRequestInProgress sync.Mutex
-	for {
-		select {
-		case <-keepAlive.Ticks():
-			if keepAlive.TimeIsUp() {
-				log.Println("Keep-alive reply not received. Close down the session.")
-				_ = sess.Close()
-				return
+func (sess *session) handleRequests(_ Context, reqs <-chan *gossh.Request) {
+	// Transport-level keep-alive is driven from Server.HandleConn so it
+	// runs for the full connection lifetime, not just while a session is
+	// active. We only need to route per-session channel requests here.
+	defer func() {
+		// winch is created on pty-req; close it once when this loop
+		// exits (i.e., reqs closed) so any consumer ranging over it
+		// terminates cleanly.
+		if sess.winch != nil {
+			close(sess.winch)
+		}
+	}()
+	for req := range reqs {
+		if ka := sess.ctx.KeepAlive(); ka != nil {
+			ka.notePeerActivity()
+		}
+		switch req.Type {
+		case "shell", "exec":
+			if sess.handled {
+				_ = req.Reply(false, nil)
+				continue
 			}
 
-			done := keepAliveRequestInProgress.TryLock()
-			if !done {
+			payload := struct{ Value string }{}
+			_ = gossh.Unmarshal(req.Payload, &payload)
+			sess.rawCmd = payload.Value
+
+			// If there's a session policy callback, we need to confirm before
+			// accepting the session.
+			if sess.sessReqCb != nil && !sess.sessReqCb(sess, req.Type) {
+				sess.rawCmd = ""
+				_ = req.Reply(false, nil)
 				continue
 			}
 
+			sess.handled = true
+			_ = req.Reply(true, nil)
+
 			go func() {
-				defer keepAliveRequestInProgress.Unlock()
-
-				// Server-initiated keep-alive flow on the client side:
-				// client: receive packet: type 98 (SSH_MSG_CHANNEL_REQUEST)
-				// client: client_input_channel_req: channel 0 rtype keepalive@openssh.com reply 1
-				// client: send packet: type 100 (SSH_MSG_CHANNEL_FAILURE)
-				//
-				// Apparently, OpenSSH client always replies with 100, but it does not matter
-				// as the server considers it as alive (only the response status is ignored).
-				_, err := sess.SendRequest(keepAliveRequestType, true, nil)
-				keepAlive.ServerRequestedKeepAliveCallback()
-				if err != nil && err != io.EOF {
-					log.Printf("Sending keep-alive request failed: %v", err)
-				} else if err == nil {
-					keepAlive.Reset()
-				}
+				sess.handler(sess)
+				_ = sess.Exit(0)
 			}()
-		case req, ok := <-reqs:
-			if !ok {
-				return
+		case "subsystem":
+			if sess.handled {
+				_ = req.Reply(false, nil)
+				continue
 			}
 
-			switch req.Type {
-			case "shell", "exec":
-				if sess.handled {
-					_ = req.Reply(false, nil)
-					continue
-				}
-
-				payload := struct{ Value string }{}
-				_ = gossh.Unmarshal(req.Payload, &payload)
-				sess.rawCmd = payload.Value
-
-				// If there's a session policy callback, we need to confirm before
-				// accepting the session.
-				if sess.sessReqCb != nil && !sess.sessReqCb(sess, req.Type) {
-					sess.rawCmd = ""
-					_ = req.Reply(false, nil)
-					continue
-				}
-
-				sess.handled = true
-				_ = req.Reply(true, nil)
+			payload := struct{ Value string }{}
+			_ = gossh.Unmarshal(req.Payload, &payload)
+			sess.subsystem = payload.Value
 
-				go func() {
-					sess.handler(sess)
-					_ = sess.Exit(0)
-				}()
-			case "subsystem":
-				if sess.handled {
-					_ = req.Reply(false, nil)
-					continue
-				}
-
-				payload := struct{ Value string }{}
-				_ = gossh.Unmarshal(req.Payload, &payload)
-				sess.subsystem = payload.Value
-
-				// If there's a session policy callback, we need to confirm before
-				// accepting the session.
-				if sess.sessReqCb != nil && !sess.sessReqCb(sess, req.Type) {
-					sess.rawCmd = ""
-					_ = req.Reply(false, nil)
-					continue
-				}
+			// If there's a session policy callback, we need to confirm before
+			// accepting the session.
+			if sess.sessReqCb != nil && !sess.sessReqCb(sess, req.Type) {
+				sess.rawCmd = ""
+				_ = req.Reply(false, nil)
+				continue
+			}
 
-				handler := sess.subsystemHandlers[payload.Value]
-				if handler == nil {
-					handler = sess.subsystemHandlers["default"]
-				}
-				if handler == nil {
-					_ = req.Reply(false, nil)
-					continue
-				}
+			handler := sess.subsystemHandlers[payload.Value]
+			if handler == nil {
+				handler = sess.subsystemHandlers["default"]
+			}
+			if handler == nil {
+				_ = req.Reply(false, nil)
+				continue
+			}
 
-				sess.handled = true
-				_ = req.Reply(true, nil)
+			sess.handled = true
+			_ = req.Reply(true, nil)
 
-				go func() {
-					handler(sess)
-					_ = sess.Exit(0)
-				}()
-			case "env":
-				if sess.handled {
-					_ = req.Reply(false, nil)
-					continue
-				}
-				var kv struct{ Key, Value string }
-				_ = gossh.Unmarshal(req.Payload, &kv)
-				sess.env = append(sess.env, fmt.Sprintf("%s=%s", kv.Key, kv.Value))
-				_ = req.Reply(true, nil)
-			case "signal":
-				var payload struct{ Signal string }
-				_ = gossh.Unmarshal(req.Payload, &payload)
-				sess.sigMu.Lock()
-				if sess.sigCh != nil {
-					sess.sigCh <- Signal(payload.Signal)
-				} else if len(sess.sigBuf) < maxSigBufSize {
-					sess.sigBuf = append(sess.sigBuf, Signal(payload.Signal))
-				}
-				sess.sigMu.Unlock()
-			case "pty-req":
-				if sess.handled || sess.pty != nil {
-					_ = req.Reply(false, nil)
-					continue
-				}
-				ptyReq, ok := parsePtyRequest(req.Payload)
+			go func() {
+				handler(sess)
+				_ = sess.Exit(0)
+			}()
+		case "env":
+			if sess.handled {
+				_ = req.Reply(false, nil)
+				continue
+			}
+			var kv struct{ Key, Value string }
+			_ = gossh.Unmarshal(req.Payload, &kv)
+			sess.env = append(sess.env, fmt.Sprintf("%s=%s", kv.Key, kv.Value))
+			_ = req.Reply(true, nil)
+		case "signal":
+			var payload struct{ Signal string }
+			_ = gossh.Unmarshal(req.Payload, &payload)
+			sess.sigMu.Lock()
+			if sess.sigCh != nil {
+				sess.sigCh <- Signal(payload.Signal)
+			} else if len(sess.sigBuf) < maxSigBufSize {
+				sess.sigBuf = append(sess.sigBuf, Signal(payload.Signal))
+			}
+			sess.sigMu.Unlock()
+		case "pty-req":
+			if sess.handled || sess.pty != nil {
+				_ = req.Reply(false, nil)
+				continue
+			}
+			ptyReq, ok := parsePtyRequest(req.Payload)
+			if !ok {
+				_ = req.Reply(false, nil)
+				continue
+			}
+			if sess.ptyCb != nil {
+				ok := sess.ptyCb(sess.ctx, ptyReq)
 				if !ok {
 					_ = req.Reply(false, nil)
 					continue
 				}
-				if sess.ptyCb != nil {
-					ok := sess.ptyCb(sess.ctx, ptyReq)
-					if !ok {
-						_ = req.Reply(false, nil)
-						continue
-					}
-				}
-				sess.pty = &ptyReq
-				sess.winch = make(chan Window, 1)
-				sess.winch <- ptyReq.Window
-				defer func() {
-					// when reqs is closed
-					close(sess.winch)
-				}()
-				_ = req.Reply(ok, nil)
-			case x11RequestType:
-				if sess.handled || sess.x11 != nil {
-					_ = req.Reply(false, nil)
-					continue
-				}
-				x11Req, ok := parseX11Request(req.Payload)
+			}
+			sess.pty = &ptyReq
+			sess.winch = make(chan Window, 1)
+			sess.winch <- ptyReq.Window
+			_ = req.Reply(ok, nil)
+		case x11RequestType:
+			if sess.handled || sess.x11 != nil {
+				_ = req.Reply(false, nil)
+				continue
+			}
+			x11Req, ok := parseX11Request(req.Payload)
+			if !ok {
+				_ = req.Reply(false, nil)
+				continue
+			}
+			if sess.x11Cb != nil {
+				ok := sess.x11Cb(sess.ctx, x11Req)
 				if !ok {
 					_ = req.Reply(false, nil)
 					continue
 				}
-				if sess.x11Cb != nil {
-					ok := sess.x11Cb(sess.ctx, x11Req)
-					if !ok {
-						_ = req.Reply(false, nil)
-						continue
-					}
-				}
-				sess.x11 = &x11Req
-				_ = req.Reply(ok, nil)
-			case "window-change":
-				if sess.pty == nil {
-					_ = req.Reply(false, nil)
-					continue
-				}
-				win, _, ok := parseWindow(req.Payload)
-				if ok {
-					sess.pty.Window = win
-					sess.winch <- win
-				}
-				_ = req.Reply(ok, nil)
-			case agentRequestType:
-				// TODO: option/callback to allow agent forwarding
-				SetAgentRequested(sess.ctx)
-				_ = req.Reply(true, nil)
-			case keepAliveRequestType:
-				if req.WantReply {
-					_ = req.Reply(true, nil)
-				}
-			case "break":
-				ok := false
-				sess.Lock()
-				if sess.breakCh != nil {
-					sess.breakCh <- true
-					ok = true
-				}
-				_ = req.Reply(ok, nil)
-				sess.Unlock()
-			default:
-				// TODO: debug log
+			}
+			sess.x11 = &x11Req
+			_ = req.Reply(ok, nil)
+		case "window-change":
+			if sess.pty == nil {
 				_ = req.Reply(false, nil)
+				continue
+			}
+			win, _, ok := parseWindow(req.Payload)
+			if ok {
+				sess.pty.Window = win
+				sess.winch <- win
+			}
+			_ = req.Reply(ok, nil)
+		case agentRequestType:
+			// TODO: option/callback to allow agent forwarding
+			SetAgentRequested(sess.ctx)
+			_ = req.Reply(true, nil)
+		case keepAliveRequestType:
+			if req.WantReply {
+				_ = req.Reply(true, nil)
+			}
+		case "break":
+			ok := false
+			sess.Lock()
+			if sess.breakCh != nil {
+				sess.breakCh <- true
+				ok = true
 			}
+			_ = req.Reply(ok, nil)
+			sess.Unlock()
+		default:
+			// TODO: debug log
+			_ = req.Reply(false, nil)
 		}
 	}
 }
diff --git a/ssh.go b/ssh.go
index 2bbfc10..3b4d98f 100644
--- a/ssh.go
+++ b/ssh.go
@@ -91,6 +91,25 @@ type ConnectionFailedCallback func(conn net.Conn, err error)
 // Please note: the ServerConn is closed at this point.
 type ConnectionCompleteCallback func(conn *gossh.ServerConn, err error)
 
+// ConnectionClosingCallback is invoked synchronously the moment HandleConn
+// observes the inbound channels stream close (i.e., the SSH transport is
+// ending), BEFORE sshConn.Wait() is called and BEFORE any deferred cleanup
+// runs. Use it for fast-path resource cleanup. Note: this fires when
+// HandleConn observes the channels stream close. If the transport is
+// permanently stuck and ClientAliveInterval is unset, neither this callback
+// nor ConnectionCompleteCallback will fire — configure keep-alive to bound
+// the wait. The Context is the same one threaded through auth and channel
+// handlers, so per-connection state stashed via ctx.SetValue is reachable
+// without an external sync.Map keyed by *gossh.ServerConn.
+//
+// Note: channel handler goroutines spawned by HandleConn may still be
+// running and mutating per-connection state when this callback fires.
+// The callback is intentionally synchronous and pre-defer; it does NOT
+// imply that all per-connection work has finished. For "all work done"
+// semantics, use ConnectionCompleteCallback (which runs after
+// sshConn.Wait).
+type ConnectionClosingCallback func(ctx Context, conn *gossh.ServerConn)
+
 // Window represents the size of a PTY window.
 //
 // See https://datatracker.ietf.org/doc/html/rfc4254#section-6.2