Add E2E test for MCPServer cross-replica session routing with Redis (#4525)

* Add E2E test for MCPServer cross-replica session routing with Redis

MCPServer supports horizontal scaling with Redis session storage, but
there was no E2E test verifying that a session established on one pod
is accessible from a different pod. This test deploys an MCPServer with
replicas=2 and Redis session storage, initializes an MCP session, then
sends raw JSON-RPC requests directly to each pod IP using the same
Mcp-Session-Id header to prove sessions are shared via Redis.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Use kubectl port-forward instead of pod IPs for cross-replica test

Pod IPs are not reachable from the CI runner host in Kind clusters.
Replace direct pod IP HTTP calls with kubectl port-forward to each
pod, which tunnels through the Kind node's network.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Wire Redis session storage into MCPServer proxy runner

The MCPServer CRD's sessionStorage config was populated by the operator
into RunConfig but the proxy runner never read it — sessions always used
in-memory LocalStorage, making cross-replica routing non-functional.

Add WithSessionStorage transport option and wire ScalingConfig.SessionRedis
from RunConfig into the transport layer so both StdioTransport and
HTTPTransport (transparent proxy) use Redis-backed session storage when
configured.

Rewrite the E2E test to use mcp-go clients throughout, including
transport.WithSession to create a client on pod B that reuses the
session established on pod A.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Move session storage to transport Config, fix codespell

Pass SessionStorage through types.Config instead of a factory option
with interface assertion. The factory now sets the field directly on
each transport type during construction.

Add clientA to codespell ignore list.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Address review comments

- Move Redis image to test/e2e/images/images.go (RedisImage constant)
- Move RedisPasswordEnvVar to pkg/transport/session/redis_config.go to
  avoid proxy runner depending on pkg/vmcp/config
- Remove unused vmcpconfig import from runner.go

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: jerm-dro

.codespellrc

@@ -1,3 +1,3 @@
 [codespell]
-ignore-words-list = NotIn,notin,AfterAll,ND,aks,deriver,te
+ignore-words-list = NotIn,notin,AfterAll,ND,aks,deriver,te,clientA
 skip = *.svg,*.mod,*.sum

pkg/runner/runner.go

@@ -33,6 +33,7 @@ import (
 	"github.com/stacklok/toolhive/pkg/secrets"
 	"github.com/stacklok/toolhive/pkg/telemetry"
 	"github.com/stacklok/toolhive/pkg/transport"
+	"github.com/stacklok/toolhive/pkg/transport/session"
 	"github.com/stacklok/toolhive/pkg/transport/types"
 	"github.com/stacklok/toolhive/pkg/workloads/statuses"
 )
@@ -353,6 +354,31 @@ func (r *Runner) Run(ctx context.Context) error {
 		}
 	}
 
+	// When Redis session storage is configured, create a Redis-backed session store
+	// so sessions are shared across proxy replicas instead of being pod-local.
+	if r.Config.ScalingConfig != nil && r.Config.ScalingConfig.SessionRedis != nil {
+		redisCfg := r.Config.ScalingConfig.SessionRedis
+		keyPrefix := redisCfg.KeyPrefix
+		if keyPrefix == "" {
+			keyPrefix = "thv:proxy:session:"
+		}
+		storage, err := session.NewRedisStorage(ctx, session.RedisConfig{
+			Addr:      redisCfg.Address,
+			Password:  os.Getenv(session.RedisPasswordEnvVar),
+			DB:        int(redisCfg.DB),
+			KeyPrefix: keyPrefix,
+		}, session.DefaultSessionTTL)
+		if err != nil {
+			return fmt.Errorf("failed to create Redis session storage: %w", err)
+		}
+		slog.Info("using Redis session storage",
+			"address", redisCfg.Address,
+			"db", redisCfg.DB,
+			"key_prefix", keyPrefix,
+		)
+		transportConfig.SessionStorage = storage
+	}
+
 	// Create transport with options
 	transportHandler, err := transport.NewFactory().Create(transportConfig, transportOpts...)
 	if err != nil {

pkg/transport/factory.go

@@ -47,13 +47,17 @@ func (*Factory) Create(config types.Config, opts ...Option) (types.Transport, er
 
 	switch config.Type {
 	case types.TransportTypeStdio:
-		tr = NewStdioTransport(
+		stdio := NewStdioTransport(
 			config.Host, config.ProxyPort, config.Deployer, config.Debug, config.TrustProxyHeaders,
 			config.PrometheusHandler, config.Middlewares...,
 		)
-		tr.(*StdioTransport).SetProxyMode(config.ProxyMode)
+		stdio.SetProxyMode(config.ProxyMode)
+		if config.SessionStorage != nil {
+			stdio.SetSessionStorage(config.SessionStorage)
+		}
+		tr = stdio
 	case types.TransportTypeSSE:
-		tr = NewHTTPTransport(
+		httpTransport := NewHTTPTransport(
 			types.TransportTypeSSE,
 			config.Host,
 			config.ProxyPort,
@@ -68,8 +72,10 @@ func (*Factory) Create(config types.Config, opts ...Option) (types.Transport, er
 			config.TrustProxyHeaders,
 			config.Middlewares...,
 		)
+		httpTransport.sessionStorage = config.SessionStorage
+		tr = httpTransport
 	case types.TransportTypeStreamableHTTP:
-		tr = NewHTTPTransport(
+		httpTransport := NewHTTPTransport(
 			types.TransportTypeStreamableHTTP,
 			config.Host,
 			config.ProxyPort,
@@ -84,6 +90,8 @@ func (*Factory) Create(config types.Config, opts ...Option) (types.Transport, er
 			config.TrustProxyHeaders,
 			config.Middlewares...,
 		)
+		httpTransport.sessionStorage = config.SessionStorage
+		tr = httpTransport
 	case types.TransportTypeInspector:
 		// HTTP transport is not implemented yet
 		return nil, errors.ErrUnsupportedTransport

pkg/transport/http.go

@@ -22,6 +22,7 @@ import (
 	transporterrors "github.com/stacklok/toolhive/pkg/transport/errors"
 	"github.com/stacklok/toolhive/pkg/transport/middleware"
 	"github.com/stacklok/toolhive/pkg/transport/proxy/transparent"
+	"github.com/stacklok/toolhive/pkg/transport/session"
 	"github.com/stacklok/toolhive/pkg/transport/types"
 )
 
@@ -73,6 +74,10 @@ type HTTPTransport struct {
 	// Mutex for protecting shared state
 	mutex sync.Mutex
 
+	// sessionStorage overrides the default in-memory session store when set.
+	// Used for Redis-backed session sharing across replicas.
+	sessionStorage session.Storage
+
 	// Transparent proxy
 	proxy types.Proxy
 
@@ -236,6 +241,8 @@ func (t *HTTPTransport) setTargetURI(targetURI string) {
 
 // Start initializes the transport and begins processing messages.
 // The transport is responsible for starting the container.
+//
+//nolint:gocyclo // Start is a candidate for refactoring; keeping this PR focused on the Redis wiring
 func (t *HTTPTransport) Start(ctx context.Context) error {
 	t.mutex.Lock()
 	defer t.mutex.Unlock()
@@ -319,6 +326,9 @@ func (t *HTTPTransport) Start(ctx context.Context) error {
 		proxyOptions = append(proxyOptions, transparent.WithRemoteBasePath(remoteBasePath))
 	}
 	proxyOptions = append(proxyOptions, transparent.WithRemoteRawQuery(remoteRawQuery))
+	if t.sessionStorage != nil {
+		proxyOptions = append(proxyOptions, transparent.WithSessionStorage(t.sessionStorage))
+	}
 
 	// Create the transparent proxy
 	t.proxy = transparent.NewTransparentProxyWithOptions(

pkg/transport/session/redis_config.go

@@ -5,6 +5,12 @@ package session
 
 import "time"
 
+// RedisPasswordEnvVar is the environment variable name for the Redis session storage password.
+// The operator injects this as a SecretKeyRef when sessionStorage.provider is "redis"
+// and passwordRef is set.
+// #nosec G101 -- This is an environment variable name, not a hardcoded credential
+const RedisPasswordEnvVar = "THV_SESSION_REDIS_PASSWORD"
+
 // Default timeouts for Redis operations.
 const (
 	DefaultDialTimeout  = 5 * time.Second

pkg/transport/types/transport.go

@@ -19,6 +19,7 @@ import (
 	"github.com/stacklok/toolhive/pkg/authserver/server/keys"
 	rt "github.com/stacklok/toolhive/pkg/container/runtime"
 	"github.com/stacklok/toolhive/pkg/transport/errors"
+	"github.com/stacklok/toolhive/pkg/transport/session"
 )
 
 // MiddlewareFunction is a function that wraps an http.Handler with additional functionality.
@@ -270,6 +271,11 @@ type Config struct {
 	//	  "/.well-known/oauth-authorization-server": authServerHandler,
 	//	}
 	PrefixHandlers map[string]http.Handler
+
+	// SessionStorage overrides the default in-memory session store when set.
+	// Used for Redis-backed session sharing across replicas.
+	// When nil, transports use their default in-memory LocalStorage.
+	SessionStorage session.Storage
 }
 
 // ProxyMode represents the proxy mode for stdio transport.

test/e2e/images/images.go

@@ -97,4 +97,9 @@ const (
 	// PagerDutyMCPServerImage is used for testing multi-backend optimizer scenarios.
 	// Provides ~64 PagerDuty incident management tools (incidents, services, schedules, etc.).
 	PagerDutyMCPServerImage = pagerdutyMCPServerImageURL + ":" + pagerdutyMCPServerImageTag
+
+	redisImageURL = "redis"
+	redisImageTag = "7-alpine"
+	// RedisImage is used for Redis-backed session storage in scaling tests.
+	RedisImage = redisImageURL + ":" + redisImageTag
 )