Fix resource leak in session storage cleanup

Implements fix by calling Close() on sessions that implement
io.Closer before deleting them from storage during cleanup.
This prevents connection leaks when vMCP sessions with backend clients
expire.

Closes: #3871

Author: taskbot

pkg/transport/session/storage_local.go

@@ -6,6 +6,8 @@ package session
 import (
 	"context"
 	"fmt"
+	"io"
+	"log/slog"
 	"sync"
 	"time"
 )
@@ -66,9 +68,12 @@ func (s *LocalStorage) Delete(_ context.Context, id string) error {
 
 // DeleteExpired removes all sessions that haven't been updated since the given time.
 func (s *LocalStorage) DeleteExpired(ctx context.Context, before time.Time) error {
-	var toDelete []string
+	var toDelete []struct {
+		id      string
+		session Session
+	}
 
-	// First pass: collect IDs of expired sessions
+	// First pass: collect expired sessions
 	s.sessions.Range(func(key, val any) bool {
 		// Check for context cancellation
 		select {
@@ -80,16 +85,27 @@ func (s *LocalStorage) DeleteExpired(ctx context.Context, before time.Time) erro
 		if session, ok := val.(Session); ok {
 			if session.UpdatedAt().Before(before) {
 				if id, ok := key.(string); ok {
-					toDelete = append(toDelete, id)
+					toDelete = append(toDelete, struct {
+						id      string
+						session Session
+					}{id, session})
 				}
 			}
 		}
 		return true
 	})
 
-	// Second pass: delete expired sessions
-	for _, id := range toDelete {
-		s.sessions.Delete(id)
+	// Second pass: close and delete expired sessions
+	for _, item := range toDelete {
+		// Call Close() on sessions that implement io.Closer
+		if closer, ok := item.session.(io.Closer); ok {
+			if err := closer.Close(); err != nil {
+				slog.Warn("failed to close session during cleanup",
+					"session_id", item.id,
+					"error", err)
+			}
+		}
+		s.sessions.Delete(item.id)
 	}
 
 	return nil

pkg/transport/session/storage_test.go

@@ -5,6 +5,7 @@ package session
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"testing"
 	"time"
@@ -13,6 +14,24 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// mockClosableSession is a test session that implements io.Closer
+type mockClosableSession struct {
+	*ProxySession
+	closeCalled bool
+	closeError  error
+}
+
+func newMockClosableSession(id string) *mockClosableSession {
+	return &mockClosableSession{
+		ProxySession: NewProxySession(id),
+	}
+}
+
+func (m *mockClosableSession) Close() error {
+	m.closeCalled = true
+	return m.closeError
+}
+
 // TestLocalStorage tests the LocalStorage implementation
 func TestLocalStorage(t *testing.T) {
 	t.Parallel()
@@ -302,6 +321,146 @@ func TestLocalStorage(t *testing.T) {
 		// Should not error, just stop early
 		assert.NoError(t, err)
 	})
+
+	t.Run("DeleteExpired calls Close on io.Closer sessions", func(t *testing.T) {
+		t.Parallel()
+		storage := NewLocalStorage()
+		defer storage.Close()
+
+		ctx := context.Background()
+
+		// Create a closable session (implements io.Closer)
+		closableSession := newMockClosableSession("closable-session")
+		closableSession.updated = time.Now().Add(-2 * time.Hour)
+
+		// Create a regular session (does not implement io.Closer)
+		regularSession := NewProxySession("regular-session")
+		regularSession.updated = time.Now().Add(-2 * time.Hour)
+
+		// Store both sessions
+		err := storage.Store(ctx, closableSession)
+		require.NoError(t, err)
+		err = storage.Store(ctx, regularSession)
+		require.NoError(t, err)
+
+		// Delete sessions older than 1 hour
+		cutoff := time.Now().Add(-1 * time.Hour)
+		err = storage.DeleteExpired(ctx, cutoff)
+		require.NoError(t, err)
+
+		// Both sessions should be deleted
+		_, err = storage.Load(ctx, "closable-session")
+		assert.Equal(t, ErrSessionNotFound, err)
+		_, err = storage.Load(ctx, "regular-session")
+		assert.Equal(t, ErrSessionNotFound, err)
+
+		// Close() should have been called on the closable session
+		assert.True(t, closableSession.closeCalled, "Close() should have been called on closable session")
+	})
+
+	t.Run("DeleteExpired continues deletion even if Close fails", func(t *testing.T) {
+		t.Parallel()
+		storage := NewLocalStorage()
+		defer storage.Close()
+
+		ctx := context.Background()
+
+		// Create a closable session that returns an error on Close()
+		failingSession := newMockClosableSession("failing-session")
+		failingSession.closeError = errors.New("close failed")
+		failingSession.updated = time.Now().Add(-2 * time.Hour)
+
+		// Store the session
+		err := storage.Store(ctx, failingSession)
+		require.NoError(t, err)
+
+		// Delete expired sessions
+		cutoff := time.Now().Add(-1 * time.Hour)
+		err = storage.DeleteExpired(ctx, cutoff)
+		require.NoError(t, err)
+
+		// Session should still be deleted despite Close() error
+		_, err = storage.Load(ctx, "failing-session")
+		assert.Equal(t, ErrSessionNotFound, err)
+
+		// Close() should have been called
+		assert.True(t, failingSession.closeCalled, "Close() should have been called")
+	})
+
+	t.Run("DeleteExpired handles non-io.Closer sessions without error", func(t *testing.T) {
+		t.Parallel()
+		storage := NewLocalStorage()
+		defer storage.Close()
+
+		ctx := context.Background()
+
+		// Create multiple regular sessions (do not implement io.Closer)
+		for i := 0; i < 5; i++ {
+			session := NewProxySession(fmt.Sprintf("session-%d", i))
+			session.updated = time.Now().Add(-2 * time.Hour)
+			err := storage.Store(ctx, session)
+			require.NoError(t, err)
+		}
+
+		// Delete expired sessions
+		cutoff := time.Now().Add(-1 * time.Hour)
+		err := storage.DeleteExpired(ctx, cutoff)
+		require.NoError(t, err)
+
+		// All sessions should be deleted
+		for i := 0; i < 5; i++ {
+			_, err := storage.Load(ctx, fmt.Sprintf("session-%d", i))
+			assert.Equal(t, ErrSessionNotFound, err)
+		}
+	})
+
+	t.Run("DeleteExpired with mixed session types", func(t *testing.T) {
+		t.Parallel()
+		storage := NewLocalStorage()
+		defer storage.Close()
+
+		ctx := context.Background()
+
+		// Create a mix of closable and regular expired sessions
+		closable1 := newMockClosableSession("closable-1")
+		closable1.updated = time.Now().Add(-2 * time.Hour)
+		closable2 := newMockClosableSession("closable-2")
+		closable2.updated = time.Now().Add(-2 * time.Hour)
+
+		regular1 := NewProxySession("regular-1")
+		regular1.updated = time.Now().Add(-2 * time.Hour)
+		regular2 := NewProxySession("regular-2")
+		regular2.updated = time.Now().Add(-2 * time.Hour)
+
+		// Store all sessions
+		err := storage.Store(ctx, closable1)
+		require.NoError(t, err)
+		err = storage.Store(ctx, closable2)
+		require.NoError(t, err)
+		err = storage.Store(ctx, regular1)
+		require.NoError(t, err)
+		err = storage.Store(ctx, regular2)
+		require.NoError(t, err)
+
+		// Delete expired sessions
+		cutoff := time.Now().Add(-1 * time.Hour)
+		err = storage.DeleteExpired(ctx, cutoff)
+		require.NoError(t, err)
+
+		// All sessions should be deleted
+		_, err = storage.Load(ctx, "closable-1")
+		assert.Equal(t, ErrSessionNotFound, err)
+		_, err = storage.Load(ctx, "closable-2")
+		assert.Equal(t, ErrSessionNotFound, err)
+		_, err = storage.Load(ctx, "regular-1")
+		assert.Equal(t, ErrSessionNotFound, err)
+		_, err = storage.Load(ctx, "regular-2")
+		assert.Equal(t, ErrSessionNotFound, err)
+
+		// Close() should have been called on closable sessions
+		assert.True(t, closable1.closeCalled, "Close() should have been called on closable-1")
+		assert.True(t, closable2.closeCalled, "Close() should have been called on closable-2")
+	})
 }
 
 // TestManagerWithStorage tests the Manager with the Storage interface