Wire Redis session storage and fix MultiSession lookup

Two bugs prevented Redis-backed VirtualMCPServer from working correctly
across pods:

1. The vmcp process never read the SessionStorage field from its config,
   always using in-memory (local) storage regardless of operator config.
   Fixed by adding SessionStorageConfig to server.Config and a
   buildRedisConfig helper in commands.go that reads the provider,
   address, DB, and password (via THV_SESSION_REDIS_PASSWORD) from the
   vmcp config at startup.

2. With Redis-backed storage, sessionmanager.Manager.GetMultiSession()
   always returned (nil, false) for fully-formed sessions. The
   transportsession.Manager.Get() deserialises from Redis into a plain
   *StreamableSession, losing the defaultMultiSession type. The type
   assertion sess.(vmcpsession.MultiSession) therefore always failed,
   causing GetAdaptedTools to error, which triggered the cleanup defer
   to call Terminate(), marking the session terminated=true in Redis.
   The client's next request got a 404 "session terminated".

   Fixed by adding a node-local sync.Map (multiSessions) to
   sessionmanager.Manager as the authoritative store for live
   MultiSession objects. CreateSession stores the live session there
   after UpsertSession succeeds; GetMultiSession and Terminate consult
   the local map first before falling back to the pluggable backend.

Also adds:
- E2E test: VirtualMCPServer Redis Session Continuity — verifies that
  MCP sessions survive complete pod replacement when Redis is configured
  (ServiceAffinityNone, 2 replicas, all pods deleted and replaced).
- Redis image constant and DeployRedis/CleanupRedis helpers in the
  e2e test package.

Related-to: #4220

Author: taskbot

cmd/vmcp/app/commands.go

@@ -21,6 +21,7 @@ import (
 	"github.com/stacklok/toolhive/pkg/container/runtime"
 	"github.com/stacklok/toolhive/pkg/groups"
 	"github.com/stacklok/toolhive/pkg/telemetry"
+	transportsession "github.com/stacklok/toolhive/pkg/transport/session"
 	"github.com/stacklok/toolhive/pkg/vmcp"
 	"github.com/stacklok/toolhive/pkg/vmcp/aggregator"
 	vmcpauth "github.com/stacklok/toolhive/pkg/vmcp/auth"
@@ -550,6 +551,11 @@ func runServe(cmd *cobra.Command, _ []string) error {
 		OptimizerConfig:         optCfg,
 		SessionFactory:          sessionFactory,
 	}
+	redisCfg, err := buildRedisConfig(cfg.SessionStorage)
+	if err != nil {
+		return fmt.Errorf("invalid session storage configuration: %w", err)
+	}
+	serverCfg.SessionStorageConfig = redisCfg
 
 	// Convert composite tool configurations to workflow definitions
 	workflowDefs, err := vmcpserver.ConvertConfigToWorkflowDefinitions(cfg.CompositeTools)
@@ -570,3 +576,31 @@ func runServe(cmd *cobra.Command, _ []string) error {
 	slog.Info(fmt.Sprintf("Starting Virtual MCP Server at %s", srv.Address()))
 	return srv.Start(ctx)
 }
+
+const defaultRedisKeyPrefix = "thv:session:"
+
+// buildRedisConfig converts a vmcp SessionStorageConfig into a transport-layer
+// RedisConfig. Returns (nil, nil) when sessionStorage is nil or provider is not "redis".
+// Returns an error when provider is "redis" but required fields are invalid.
+// The Redis password is read from the THV_SESSION_REDIS_PASSWORD environment variable.
+func buildRedisConfig(sessionStorage *config.SessionStorageConfig) (*transportsession.RedisConfig, error) {
+	if sessionStorage == nil || sessionStorage.Provider != "redis" {
+		return nil, nil
+	}
+	if sessionStorage.Address == "" {
+		return nil, fmt.Errorf("session storage provider is \"redis\" but address is empty")
+	}
+	keyPrefix := sessionStorage.KeyPrefix
+	if keyPrefix == "" {
+		keyPrefix = defaultRedisKeyPrefix
+	}
+	if keyPrefix[len(keyPrefix)-1] != ':' {
+		return nil, fmt.Errorf("session storage key prefix %q must end with ':'", keyPrefix)
+	}
+	return &transportsession.RedisConfig{
+		Addr:      sessionStorage.Address,
+		Password:  os.Getenv(config.RedisPasswordEnvVar),
+		DB:        int(sessionStorage.DB),
+		KeyPrefix: keyPrefix,
+	}, nil
+}

pkg/transport/session/manager.go

@@ -276,6 +276,21 @@ func (m *Manager) Count() int {
 	return 0
 }
 
+// TTL returns the session time-to-live configured for this manager.
+func (m *Manager) TTL() time.Duration {
+	return m.ttl
+}
+
+// Peek reports whether a session exists without refreshing its eviction TTL.
+// Returns (true, nil) if found, (false, nil) if definitively absent, and
+// (false, err) if the storage backend could not be reached.
+// Callers must treat a non-nil error as "unknown" rather than "not found".
+func (m *Manager) Peek(id string) (bool, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultOperationTimeout)
+	defer cancel()
+	return m.storage.Peek(ctx, id)
+}
+
 func (m *Manager) cleanupExpiredOnce() error {
 	cutoff := time.Now().Add(-m.ttl)
 	ctx, cancel := context.WithTimeout(context.Background(), cleanupOperationTimeout)

pkg/transport/session/storage.go

@@ -26,6 +26,13 @@ type Storage interface {
 	// that DeleteExpired does not evict sessions that are actively being accessed.
 	Load(ctx context.Context, id string) (Session, error)
 
+	// Peek reports whether a session exists without refreshing its eviction TTL.
+	// Returns (true, nil) if the session exists, (false, nil) if it definitively
+	// does not exist, and (false, err) if the check could not be completed (e.g.
+	// storage unavailable). Callers must treat non-nil errors as "unknown"
+	// rather than "not found".
+	Peek(ctx context.Context, id string) (bool, error)
+
 	// Delete removes a session from the storage backend.
 	// It is not an error if the session doesn't exist.
 	Delete(ctx context.Context, id string) error

pkg/transport/session/storage_local.go

@@ -90,6 +90,17 @@ func (s *LocalStorage) Load(_ context.Context, id string) (Session, error) {
 	return entry.session, nil
 }
 
+// Peek reports whether a session exists in local storage without updating its
+// last-access timestamp. This avoids extending the TTL of idle sessions during
+// eviction-loop probes.
+func (s *LocalStorage) Peek(_ context.Context, id string) (bool, error) {
+	if id == "" {
+		return false, fmt.Errorf("cannot peek session with empty ID")
+	}
+	_, ok := s.sessions.Load(id)
+	return ok, nil
+}
+
 // Delete removes a session from local storage.
 func (s *LocalStorage) Delete(_ context.Context, id string) error {
 	if id == "" {

pkg/transport/session/storage_redis.go

@@ -195,6 +195,20 @@ func (s *RedisStorage) Load(ctx context.Context, id string) (Session, error) {
 	return session, nil
 }
 
+// Peek checks whether a session key exists in Redis without refreshing its TTL.
+// Uses EXISTS rather than GETEX so that idle sessions are not kept alive by the
+// eviction-loop probes in sessionmanager.
+func (s *RedisStorage) Peek(ctx context.Context, id string) (bool, error) {
+	if id == "" {
+		return false, fmt.Errorf("cannot peek session with empty ID")
+	}
+	n, err := s.client.Exists(ctx, s.key(id)).Result()
+	if err != nil {
+		return false, fmt.Errorf("failed to check session existence: %w", err)
+	}
+	return n > 0, nil
+}
+
 // Delete removes the Redis key. A missing key is not an error.
 func (s *RedisStorage) Delete(ctx context.Context, id string) error {
 	if id == "" {

pkg/vmcp/server/server.go

@@ -171,6 +171,12 @@ type Config struct {
 	// SessionFactory creates MultiSessions for session management.
 	// Required; must not be nil.
 	SessionFactory vmcpsession.MultiSessionFactory
+
+	// SessionStorageConfig configures the session storage backend.
+	// When nil, an in-process local store is used (single-pod only).
+	// When non-nil, a Redis-backed store is used so that session metadata
+	// survives pod restarts, enabling horizontal scaling without sticky routing.
+	SessionStorageConfig *transportsession.RedisConfig
 }
 
 // Server is the Virtual MCP Server that aggregates multiple backends.
@@ -359,7 +365,18 @@ func New(
 	// It intentionally carries no vmcp-specific state — backend connections, routing
 	// tables, tool lists, and token binding all live in the separate sessionmanager.Manager,
 	// keyed by the same session ID.
-	sessionManager := transportsession.NewManager(cfg.SessionTTL, transportsession.NewStreamableSession)
+	var sessionManager *transportsession.Manager
+	if cfg.SessionStorageConfig != nil {
+		sessionManager, err = transportsession.NewManagerWithRedis(
+			ctx, cfg.SessionTTL, transportsession.NewStreamableSession, *cfg.SessionStorageConfig,
+		)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create Redis-backed session manager: %w", err)
+		}
+		slog.Info("using Redis-backed session storage", "addr", cfg.SessionStorageConfig.Addr)
+	} else {
+		sessionManager = transportsession.NewManager(cfg.SessionTTL, transportsession.NewStreamableSession)
+	}
 
 	// Create handler factory (used by adapter and for future dynamic registration)
 	handlerFactory := adapter.NewDefaultHandlerFactory(rt, backendClient)