Feature Request: Implement Session Persistence & Pooling for Improved Performance and State Continuity

[crivetimihai]

Executive Summary

This feature request proposes implementing session persistence across tool calls and session pooling to address performance bottlenecks and state continuity issues in Context Forge. The approach prioritizes using existing stateful session support (USE_STATEFUL_SESSIONS) and extending it to other transports.

Key Goals:

• Session persistence: reuse one MCP session across multiple tool calls for the same client workflow
• Session pooling: cache sessions for reuse across separate HTTP requests where a persistent channel is not used
• Enable state continuity without weakening isolation or multi-tenant boundaries

Problem Statement

Context Forge currently implements a per-request session strategy for SSE transport, where every tool invocation creates a new MCP session. While this prioritizes isolation and fault tolerance, it introduces significant performance penalties and breaks stateful tool workflows.

Current Behavior Issues:

• Network Overhead: Each tool call requires 4-6 network round trips before execution
• State Loss: Conversational context lost between sequential tool calls
• Resource Consumption: Memory and TCP connections grow linearly with tool invocation frequency
• Protocol Compliance: Violates MCP specification expectations for stateful sessions

Performance Impact:

• Resource Consumption Per Session: 2-4KB memory (asyncio.Queue + Event + registry entry)
• Network Overhead: 4-6 HTTP round trips (initialize + initialized + 3x notifications)
• Scaling Issues: At 100 tools/second = 400KB/sec memory + 400-600 network calls/sec

Root Cause Analysis:

SSE Endpoint (mcpgateway/main.py:1313-1347):

@server_router.get("/{server_id}/sse")
async def sse_endpoint(...):
    transport = SSETransport(base_url=server_sse_url)  # ❌ NEW EVERY TIME
    await transport.connect()
    await session_registry.add_session(transport.session_id, transport)

Session ID Generation (mcpgateway/transports/sse_transport.py:114):

def __init__(self, base_url: str = None):
    self._session_id = str(uuid.uuid4())  # ❌ FRESH UUID PER INSTANCE

Current Configuration Landscape

Existing Stateful Sessions (already implemented but disabled):

# Session behavior flag (used by streamable HTTP; not yet wired for SSE/WS)
USE_STATEFUL_SESSIONS=false

# Cache backend selection (impacts session storage)
CACHE_TYPE=database
# CACHE_TYPE=redis

Current Limitations:

• No session pooling configuration options
• No per-server session strategy control
• No session reuse timeout settings
• No user-scoped session limits

Quick Win Opportunity

Enable existing stateful sessions for streamable HTTP transport:

• Set USE_STATEFUL_SESSIONS=true in .env
• Behavior: single client workflow reuses one MCP session across multiple tool invocations
• Reference: mcpgateway/transports/streamablehttp_transport.py already supports this

Acceptance Criteria

• Streamable HTTP with USE_STATEFUL_SESSIONS=true reuses single MCP session across multiple tool calls
• SSE/WS with pooling enabled reuses sessions for same (user, server) across requests
• Per-server override can enable/disable pooling regardless of global defaults
• User isolation maintained: sessions never shared across users or tenants
• Auth context preserved and validated on session reuse
• Observability: metrics expose pool hit/miss, active/idle counts, cleanup events
• Backward compatibility: with pooling disabled, behavior matches current per-request sessions

Expected Performance Improvements

• Response Time: 30%+ improvement expected
• Memory Usage: 70%+ reduction expected
• Network Calls: 75%+ reduction expected
• Initialization Skipped: 80%+ of calls should skip init sequence

Impact Analysis

Cache Type Compatibility:

• Memory Backend (CACHE_TYPE=memory): ❌ Not suitable for multiple workers
• Redis Backend (CACHE_TYPE=redis): ✅ Fully compatible with multi-worker deployments
• Database Backend (CACHE_TYPE=database): ⚠️ Works but creates DB performance bottleneck

Multi-Worker/Container Impact:

• Memory backend + Multiple Workers: ❌ Session loss across workers
• Redis backend + Multiple Workers: ✅ Sessions shared across workers
• Database backend + Multiple Workers: ⚠️ Works but 3x DB load increase

---

Labels: enhancement, performance, session-management
Priority: high

Note: Design details, implementation specifications, and comprehensive testing strategy will be provided in follow-up comments.

[crivetimihai]

Design Architecture & Solution Approach

Two-Track Implementation Strategy

The solution addresses two complementary scenarios:

1. Streamable HTTP: Prefer single persistent client↔gateway↔server session (controlled by USE_STATEFUL_SESSIONS)
2. SSE/WebSocket: Introduce in-gateway session pool keyed by (user, server) to avoid re-initialization

Session Pool Data Structure

Core Classes (mcpgateway/cache/session_pool.py - new file):

class SessionPool:
    def __init__(self, session_registry: SessionRegistry, config: SessionPoolConfig):
        self._pool: Dict[str, PooledSession] = {}  # user:server -> session
        self._registry = session_registry
        self._lock = asyncio.Lock()
        self._cleanup_task: Optional[asyncio.Task] = None
        self._config = config

class PooledSession:
    def __init__(self, transport: SSETransport, user_id: str, server_id: str):
        self.transport = transport
        self.user_id = user_id
        self.server_id = server_id
        self.created_at = time.time()
        self.last_used = time.time()
        self.use_count = 0
        self.active_connections = 0
        self._respond_task: Optional[asyncio.Task] = None

Configuration Integration

Environment Variables (.env.example additions):

#####################################
# Session Persistence Configuration
#####################################

# Enable stateful sessions for streamable HTTP transport
# When enabled, single client workflow reuses one MCP session
USE_STATEFUL_SESSIONS=false

# Enable session pooling for SSE/WebSocket transports
# Default: false (maintains current per-request behavior)
SESSION_POOLING_ENABLED=false

# Session pooling strategy
# - user-server: Pool sessions per user+server combination (recommended)
# - global: Single global session pool (not recommended for multi-tenant)
# - disabled: Force disable pooling
SESSION_POOL_STRATEGY=user-server

# Session pool time-to-live in seconds (default: 1800 = 30 minutes)
SESSION_POOL_TTL=1800

# Maximum pooled sessions per user (default: 10)
SESSION_POOL_MAX_PER_USER=10

# Maximum idle time before session cleanup (default: 300 = 5 minutes)
SESSION_POOL_MAX_IDLE_TIME=300

# Session pool cleanup task interval (default: 60 = 1 minute)
SESSION_POOL_CLEANUP_INTERVAL=60

# Session pool metrics collection (default: true)
SESSION_POOL_METRICS_ENABLED=true

Session Lifecycle Management

stateDiagram-v2
    [*] --> Creating: New session request
    Creating --> Pooled: Registration successful
    Pooled --> Active: Tool invocation
    Active --> Idle: Tool completion
    Idle --> Active: Reuse within TTL
    Idle --> Expired: TTL exceeded
    Active --> Expired: Max idle time
    Expired --> [*]: Cleanup & removal
    
    Pooled --> Failed: Connection lost
    Active --> Failed: Transport error
    Failed --> [*]: Error cleanup

Loading

Transport Architecture Differences

SSE Transport (mcpgateway/transports/sse_transport.py)

• Protocol: Server-Sent Events over HTTP
• Connection: Unidirectional (server → client streaming)
• Session Management: Each SSE connection creates new transport with UUID session ID
• Client Integration: Used by /servers/{server_id}/sse endpoint

Streamable HTTP Transport (mcpgateway/transports/streamablehttp_transport.py)

• Protocol: MCP-compliant Streamable HTTP (bidirectional)
• Connection: Request/response with event streaming capability
• Session Management: Uses StreamableHTTPSessionManager with configurable state
• Client Integration: Used by /servers/{server_id}/mcp endpoint

Cache Backend Impact on Session Distribution

graph TD
    subgraph "Memory Backend"
        A1[Session Pool] --> B1[Local Dict Cache]
        B1 --> C1[Single Worker Only]
        C1 --> D1[Fast Access]
    end
    
    subgraph "Redis Backend"
        A2[Session Pool] --> B2[Redis Hash Storage]
        B2 --> C2[Multi-Worker Shared]
        C2 --> D2[Network Latency]
    end
    
    subgraph "Database Backend"
        A3[Session Pool] --> B3[DB Table Storage]
        B3 --> C3[Multi-Worker + Persistent]
        C3 --> D3[DB Query Overhead]
    end
    
    style A1 fill:#e1f5fe
    style A2 fill:#fff3e0
    style A3 fill:#f3e5f5

Loading

Security Design

User Isolation:

• Pool key format: "user_email:server_id" ensures no cross-user sharing
• Authentication context preserved and validated on session reuse
• Session cleanup triggered on user logout

Resource Protection:

• SESSION_POOL_MAX_PER_USER prevents resource exhaustion attacks
• TTL-based cleanup prevents indefinite memory growth
• Connection counting prevents session hoarding

Multi-Tenant Boundaries:

• Sessions scoped by server ID prevent cross-server contamination
• JWT token validation continues to enforce access controls
• Per-server session strategy allows tenant-specific configuration

[crivetimihai]

Implementation Details & Code Changes

Phase 1: Enable Existing Stateful Sessions (Quick Win)

No code changes required - just configuration:

# In .env file
USE_STATEFUL_SESSIONS=true

Validation:

• Multiple consecutive tool calls occur after single initialize
• Tools/resources/prompts not re-listed between calls unless server-initiated changes
• Session torn down on channel close or TTL expiry

Phase 2: Core Session Pool Infrastructure

Files to Create:

• mcpgateway/cache/session_pool.py - Session pool implementation
• tests/unit/mcpgateway/cache/test_session_pool.py - Unit tests

Key Methods (session_pool.py):

class SessionPool:
    async def get_or_create_session(self, user_id: str, server_id: str, base_url: str) -> SSETransport:
        """Get existing session or create new one based on pooling strategy"""
        pool_key = self._get_pool_key(user_id, server_id)
        
        async with self._lock:
            if pool_key in self._pool:
                session = self._pool[pool_key]
                if await self._is_session_valid(session):
                    session.last_used = time.time()
                    session.use_count += 1
                    return session.transport
                else:
                    await self._cleanup_session(pool_key)
            
            # Create new session
            transport = SSETransport(base_url=base_url, pooled=True, pool_key=pool_key)
            await transport.connect()
            await self._registry.add_session(transport.session_id, transport, pooled=True)
            
            session = PooledSession(transport, user_id, server_id)
            self._pool[pool_key] = session
            return transport

    def _get_pool_key(self, user_id: str, server_id: str) -> str:
        """Generate consistent pool key for user+server combination"""
        return f"{user_id}:{server_id}"

    async def _is_session_valid(self, session: PooledSession) -> bool:
        """Check if pooled session is still valid and connected"""

Phase 3: SSE Endpoint Integration

Files to Modify:

• mcpgateway/main.py:1313-1347 - SSE endpoint with pooling
• mcpgateway/main.py:1350-1389 - Message endpoint affinity
• mcpgateway/transports/sse_transport.py:110+ - Pool-aware transport

SSE Endpoint Changes:

# Current code
async def sse_endpoint(request: Request, server_id: str, user=Depends(...)):
    transport = SSETransport(base_url=server_sse_url)  # Always new
    await transport.connect()
    await session_registry.add_session(transport.session_id, transport)

# Enhanced code  
async def sse_endpoint(request: Request, server_id: str, user=Depends(...)):
    user_id = get_user_email(user)
    
    # Check if session pooling enabled for this server
    if await should_use_session_pooling(server_id):
        transport = await session_pool.get_or_create_session(user_id, server_id, server_sse_url)
    else:
        # Fallback to current behavior
        transport = SSETransport(base_url=server_sse_url)
        await transport.connect()
        await session_registry.add_session(transport.session_id, transport)
    
    response = await transport.create_sse_response(request)
    # ... rest of existing logic

Transport Enhancement (sse_transport.py):

class SSETransport(Transport):
    def __init__(self, base_url: str = None, pooled: bool = False, pool_key: str = None):
        # Existing initialization...
        self._pooled = pooled
        self._pool_key = pool_key
        self._initialization_complete = False
        self._last_activity = time.time()

    async def send_message(self, message: Dict[str, Any]) -> None:
        """Enhanced to skip duplicate initialization for pooled sessions"""
        self._last_activity = time.time()
        
        # Skip duplicate initialization for pooled sessions
        if (self._pooled and 
            message.get("method") == "initialize" and 
            self._initialization_complete):
            logger.debug(f"Skipping duplicate initialization for pooled session {self._session_id}")
            return
            
        await super().send_message(message)
        
        if message.get("method") == "initialize":
            self._initialization_complete = True

Phase 4: Server Service Integration

Files to Modify:

• mcpgateway/services/server_service.py - Session strategy resolution
• mcpgateway/schemas.py - Per-server session strategy
• mcpgateway/config.py:480+ - Add session pooling configuration

Strategy Resolution Logic:

# server_service.py
async def get_session_strategy(self, server_id: str) -> str:
    """Determine effective session strategy for server"""
    server = await self.get_server(server_id)
    
    if server.session_pooling_strategy:
        if server.session_pooling_strategy == "inherit":
            return settings.session_pool_strategy
        return server.session_pooling_strategy
    
    return settings.session_pool_strategy if settings.session_pooling_enabled else "disabled"

async def should_use_pooling(self, server_id: str) -> bool:
    """Check if session pooling should be used for server"""
    strategy = await self.get_session_strategy(server_id)
    return strategy in ["user-server", "global", "enabled"]

Configuration Settings (config.py):

class Settings(BaseSettings):
    # Session Pooling Configuration
    session_pooling_enabled: bool = Field(default=False, description="Enable session pooling for improved performance")
    session_pool_strategy: str = Field(default="user-server", description="Pooling strategy: 'user-server', 'global', 'disabled'")
    session_pool_ttl: int = Field(default=1800, description="Session pool TTL in seconds (30 minutes)")
    session_pool_max_per_user: int = Field(default=10, description="Maximum pooled sessions per user")
    session_pool_max_idle_time: int = Field(default=300, description="Maximum idle time before session cleanup (5 minutes)")
    session_pool_cleanup_interval: int = Field(default=60, description="Cleanup task interval in seconds")

Phase 5: Observability Integration

Files to Create/Modify:

• mcpgateway/observability/ - Session pool metrics integration
• mcpgateway/admin.py - Session pool management UI
• Admin templates for session monitoring

Metrics to Expose:

• session_pool_size - Current number of pooled sessions
• session_pool_hits - Cache hit rate for session reuse
• session_pool_misses - Cache miss rate requiring new sessions
• session_pool_evictions - Sessions removed due to TTL/limits
• session_initialization_skipped - Initialize operations avoided
• session_pool_memory_usage - Memory consumed by pooled sessions

Database Migration (if per-server config added)

Location: alembic/versions/

-- Add session pooling configuration to servers table
ALTER TABLE servers 
ADD COLUMN session_pooling_strategy VARCHAR(50) NULL,
ADD COLUMN session_pool_ttl INTEGER NULL;

Backward Compatibility Strategy

Configuration Defaults:

• session_pooling_enabled: false - Feature disabled by default
• Existing installations continue with current per-request behavior
• Opt-in activation via environment variables or admin UI

Fallback Mechanisms:

• If session pool encounters errors, fall back to per-request sessions
• Per-server override allows gradual rollout
• Existing session registry backends continue to work unchanged

API Compatibility:

• All existing endpoints maintain same signature
• No client-side changes required
• Session pooling operates transparently to MCP clients

[crivetimihai]

Testing Strategy & Multi-Environment Validation

Pre-Implementation Baseline Testing

Test Environment Setup:

# Set up baseline testing environment
export SESSION_POOLING_ENABLED=false
export MCPGATEWAY_METRICS_ENABLED=true
export LOG_LEVEL=DEBUG

# Start monitoring and collect baseline metrics
make dev &
curl http://localhost:4444/metrics > baseline_metrics_start.txt

Performance Baseline Collection (tests/performance/test_session_baseline.py):

class SessionBaselineTests:
    async def test_single_user_sequential_tools(self):
        """Test: 10 sequential tool calls from single user"""
        start_time = time.time()
        memory_before = self.get_memory_usage()
        
        results = []
        for i in range(10):
            tool_start = time.time()
            response = await self.call_tool(f"ping-{i}")
            tool_end = time.time()
            results.append(tool_end - tool_start)
            
        return {
            "total_time": time.time() - start_time,
            "individual_times": results,
            "average_time": statistics.mean(results),
            "memory_delta": self.get_memory_usage() - memory_before,
            "session_count": self.count_active_sessions()
        }

    async def test_network_overhead_measurement(self):
        """Measure network calls per tool invocation"""
        network_counter = NetworkCallCounter()
        
        with network_counter:
            await self.call_tool("test-tool")
            
        return {
            "http_requests": network_counter.request_count,
            "initialize_calls": network_counter.count_method("initialize"),
            "notification_calls": network_counter.count_notifications()
        }

Post-Implementation Validation Testing

A/B Comparison Test Suite (tests/performance/test_session_pooling_comparison.py):

class SessionPoolingComparisonTests:
    async def test_performance_improvement_validation(self):
        """Validate expected performance improvements"""
        
        # Test with pooling disabled (baseline)
        baseline_results = await self.run_test_suite(
            session_pooling_enabled=False,
            test_scenarios=["sequential_tools_single_user", "concurrent_users_burst"]
        )
        
        # Test with pooling enabled
        pooling_results = await self.run_test_suite(
            session_pooling_enabled=True,
            test_scenarios=["sequential_tools_single_user", "concurrent_users_burst"]
        )
        
        # Validate against expected thresholds
        improvements = self.calculate_improvements(baseline_results, pooling_results)
        assert improvements.response_time_improvement > 0.3, "Response time should improve by 30%+"
        assert improvements.memory_reduction > 0.7, "Memory usage should reduce by 70%+"
        assert improvements.network_call_reduction > 0.75, "Network calls should reduce by 75%+"

    async def test_state_continuity_validation(self):
        """Validate that session pooling preserves state across tool calls"""
        
        with self.configure_session_pooling(enabled=True):
            # Call 1: Initialize state
            result1 = await self.call_tool("stateful-counter", {"action": "reset"})
            assert result1["counter"] == 0
            
            # Call 2: Increment (should use same session)
            result2 = await self.call_tool("stateful-counter", {"action": "increment"}) 
            assert result2["counter"] == 1, "State should persist across calls"
            
            # Call 3: Increment again
            result3 = await self.call_tool("stateful-counter", {"action": "increment"})
            assert result3["counter"] == 2, "State should continue to persist"
            
        # Verify only one session was used
        session_metrics = await self.get_session_pool_metrics()
        assert session_metrics["unique_sessions_created"] == 1
        assert session_metrics["session_reuse_count"] == 2

Docker Compose Multi-Container Testing

Complete Test Environment (docker-compose.test.yml):

version: "3.9"

services:
  # Load Balancer
  nginx:
    image: nginx:alpine
    ports: ["4444:80"]
    volumes: ["./test/nginx.conf:/etc/nginx/nginx.conf:ro"]

  # Gateway Instance 1
  gateway-1:
    image: mcpgateway/mcpgateway:latest
    environment:
      - CACHE_TYPE=redis
      - REDIS_URL=redis://redis:6379/0
      - SESSION_POOLING_ENABLED=true
      - SESSION_POOL_STRATEGY=user-server
      - GUNICORN_WORKERS=2

  # Gateway Instance 2  
  gateway-2:
    image: mcpgateway/mcpgateway:latest
    environment:
      - CACHE_TYPE=redis
      - REDIS_URL=redis://redis:6379/0
      - SESSION_POOLING_ENABLED=true
      - SESSION_POOL_STRATEGY=user-server
      - GUNICORN_WORKERS=2

  # Shared Redis for Session Storage
  redis:
    image: redis:latest
    command: redis-server --appendonly yes

  # Test Client for Load Generation
  test-client:
    image: python:3.11-slim
    volumes: ["./tests:/app/tests", "./scripts:/app/scripts"]
    command: ["python", "-m", "pytest", "tests/performance/test_session_pooling_distributed.py", "-v"]

Multi-Container Validation Commands:

# Build and start test environment
docker-compose -f docker-compose.test.yml build
docker-compose -f docker-compose.test.yml up -d

# Run distributed session pooling tests
docker-compose -f docker-compose.test.yml exec test-client \
    python -m pytest tests/performance/test_session_pooling_distributed.py -v

# Check Redis session storage
docker-compose -f docker-compose.test.yml exec redis redis-cli keys "mcp:session:*"

# Monitor session pool metrics from both instances
docker-compose -f docker-compose.test.yml exec gateway-1 \
    curl http://localhost:4444/metrics | grep session_pool
docker-compose -f docker-compose.test.yml exec gateway-2 \
    curl http://localhost:4444/metrics | grep session_pool

# View container logs for session activity
docker-compose -f docker-compose.test.yml logs gateway-1 | grep session
docker-compose -f docker-compose.test.yml logs gateway-2 | grep session

# Cleanup
docker-compose -f docker-compose.test.yml down -v

Multi-Container Session Validation

Distributed Session Pooling Tests (tests/performance/test_session_pooling_distributed.py):

class DistributedSessionPoolingTests:
    """Test session pooling across multiple gateway instances"""
    
    async def test_cross_container_session_reuse(self):
        """Test that sessions are reused across different container instances"""
        
        headers = {"Authorization": f"Bearer {self.auth_token}"}
        
        async with aiohttp.ClientSession() as session:
            # Make multiple requests - load balancer will distribute across containers
            session_ids = set()
            for i in range(20):  # 20 requests should hit both containers
                
                # Establish SSE connection
                async with session.get(
                    f"{self.gateway_url}/servers/{server_id}/sse",
                    headers=headers
                ) as sse_response:
                    # Extract session ID from SSE endpoint response
                    session_id = self.extract_session_id_from_sse(sse_response)
                    session_ids.add(session_id)
            
            # Verify session pooling is working
            # With proper session pooling, we should see session reuse
            # Without pooling, we'd see 20 unique session IDs
            assert len(session_ids) < 10, f"Too many unique sessions: {len(session_ids)} (expected < 10 with pooling)"

    async def test_redis_session_distribution(self):
        """Test Redis-based session sharing between containers"""
        
        import redis.asyncio as redis
        redis_client = redis.Redis.from_url("redis://redis:6379/0")
        
        # Count sessions before test
        initial_session_count = len(await redis_client.keys("mcp:session:*"))
        
        # Create multiple concurrent sessions
        tasks = []
        async with aiohttp.ClientSession() as session:
            for i in range(5):
                task = asyncio.create_task(
                    self.create_sse_connection(session, f"test-server-{i}")
                )
                tasks.append(task)
            
            await asyncio.gather(*tasks)
        
        # Verify sessions are stored in Redis
        final_session_count = len(await redis_client.keys("mcp:session:*"))
        assert final_session_count > initial_session_count, "No new sessions found in Redis"

Automated Testing Pipeline

Complete Test Runner (scripts/run_session_performance_tests.sh):

#!/bin/bash
set -e

RESULTS_DIR="test_results/session_pooling_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$RESULTS_DIR"

echo "🧪 Starting Session Pooling Performance Validation"

# Phase 1: Baseline Testing (Pooling Disabled)
echo "📊 Phase 1: Collecting baseline metrics"
export SESSION_POOLING_ENABLED=false
export MCPGATEWAY_METRICS_ENABLED=true

make dev > "$RESULTS_DIR/baseline_server.log" 2>&1 &
SERVER_PID=$!
sleep 10

python -m pytest tests/performance/test_session_baseline.py \
    --json-report --json-report-file="$RESULTS_DIR/baseline_results.json" -v

kill $SERVER_PID && wait $SERVER_PID 2>/dev/null || true

# Phase 2: Session Pooling Testing
echo "🚀 Phase 2: Testing with session pooling enabled"
export SESSION_POOLING_ENABLED=true
export SESSION_POOL_STRATEGY=user-server

make dev > "$RESULTS_DIR/pooling_server.log" 2>&1 &
SERVER_PID=$!
sleep 10

python -m pytest tests/performance/test_session_pooling_comparison.py \
    --json-report --json-report-file="$RESULTS_DIR/pooling_results.json" -v

kill $SERVER_PID && wait $SERVER_PID 2>/dev/null || true

# Phase 3: Generate Comparison Report
echo "📈 Phase 3: Generating performance comparison report"
python scripts/generate_performance_report.py \
    --baseline "$RESULTS_DIR/baseline_results.json" \
    --pooling "$RESULTS_DIR/pooling_results.json" \
    --output "$RESULTS_DIR/performance_report.html"

echo "🎉 Session pooling validation complete!"
echo "📂 Results saved in: $RESULTS_DIR"

Expected Performance Benchmarks

Target Improvements:

EXPECTED_IMPROVEMENTS = {
    "initialization_skipped_ratio": "> 0.8",      # 80%+ of calls skip init
    "memory_allocation_reduction": "> 0.7",       # 70%+ memory savings  
    "network_call_reduction": "> 0.75",           # 75%+ network savings
    "response_time_improvement": "> 0.3",         # 30%+ faster responses
    "database_query_reduction": "> 0.8"           # 80%+ fewer DB queries
}

Load Testing Scenarios:

1. Burst Load: 1000 tool calls in 10 seconds from 50 users
2. Sustained Load: 100 tools/second for 30 minutes
3. User Churn: High user login/logout with session cleanup
4. Server Diversity: Tools spread across 20+ virtual servers

Regression Testing Requirements

Security Validation:

• User isolation maintained (no cross-user session sharing)
• Server isolation maintained (sessions scoped per server)
• Auth context preserved on session reuse
• Session cleanup on user logout

Functional Validation:

• Backward compatibility with pooling disabled
• Per-server session strategy overrides work correctly
• Session TTL and cleanup behavior functions properly
• Metrics collection and observability operational

Multi-Environment Testing:

• Single worker vs multiple workers (GUNICORN_WORKERS=1 vs GUNICORN_WORKERS=4)
• Different cache backends (memory vs redis vs database)
• Container scaling (single container vs multi-container with load balancer)
• Network partition and Redis failure scenarios

This comprehensive testing framework ensures session persistence and pooling delivers expected performance improvements while maintaining security and compatibility across all deployment configurations.