1→"""
     2→Genesis ClawdBot Enterprise Redis Client
     3→=========================================
     4→
     5→Production-grade Redis client for ClawdBot with:
     6→- Elestio cloud configuration (proper ACL auth)
     7→- Connection pooling for efficiency
     8→- Circuit breaker for resilience
     9→- Exponential backoff reconnection
    10→- Both sync and async interfaces
    11→- Health monitoring
    12→
    13→Usage:
    14→    from ClawdBot.redis_client_v2 import redis_client, async_redis_client
    15→
    16→    # Sync operations
    17→    redis_client.publish("genesis:observations", {"type": "test"})
    18→
    19→    # Async operations
    20→    await async_redis_client.publish("genesis:observations", {"type": "test"})
    21→
    22→    # Health check
    23→    if redis_client.is_connected():
    24→        print("Redis healthy")
    25→
    26→Author: Genesis System
    27→Version: 2.0.0
    28→"""
    29→
    30→import os
    31→import sys
    32→import json
    33→import time
    34→import asyncio
    35→import logging
    36→import threading
    37→from typing import Optional, Dict, Any, Callable, List, Union
    38→from dataclasses import dataclass, field
    39→from functools import wraps
    40→from datetime import datetime
    41→
    42→import redis
    43→from redis import Redis, ConnectionPool
    44→from redis.asyncio import Redis as AsyncRedis
    45→from redis.asyncio.connection import ConnectionPool as AsyncConnectionPool
    46→from redis.exceptions import (
    47→    ConnectionError as RedisConnectionError,
    48→    TimeoutError as RedisTimeoutError,
    49→    AuthenticationError,
    50→    ResponseError
    51→)
    52→
    53→# Add genesis-system and genesis-memory to path for elestio_config import
    54→_genesis_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    55→sys.path.insert(0, _genesis_root)
    56→sys.path.insert(0, os.path.join(_genesis_root, "data", "genesis-memory"))
    57→
    58→try:
    59→    # Import directly from genesis-memory (hyphen in path requires direct import)
    60→    import elestio_config
    61→    get_redis_config = elestio_config.get_redis_config
    62→    RedisConfig = elestio_config.RedisConfig
    63→    ELESTIO_AVAILABLE = True
    64→except ImportError as e:
    65→    # Fallback if elestio_config not found
    66→    ELESTIO_AVAILABLE = False
    67→    get_redis_config = None
    68→    RedisConfig = None
    69→
    70→try:
    71→    from core.circuit_breaker import CircuitBreaker, CircuitState
    72→    CIRCUIT_BREAKER_AVAILABLE = True
    73→except ImportError:
    74→    CIRCUIT_BREAKER_AVAILABLE = False
    75→
    76→# Logging configuration
    77→logging.basicConfig(
    78→    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    79→    level=logging.INFO
    80→)
    81→logger = logging.getLogger("ClawdBotRedisV2")
    82→
    83→
    84→@dataclass
    85→class ConnectionStats:
    86→    """Track connection health and statistics."""
    87→    total_operations: int = 0
    88→    successful_operations: int = 0
    89→    failed_operations: int = 0
    90→    reconnections: int = 0
    91→    last_success: Optional[datetime] = None
    92→    last_failure: Optional[datetime] = None
    93→    circuit_state: str = "closed"
    94→
    95→    def record_success(self):
    96→        self.total_operations += 1
    97→        self.successful_operations += 1
    98→        self.last_success = datetime.now()
    99→
   100→    def record_failure(self):
   101→        self.total_operations += 1
   102→        self.failed_operations += 1
   103→        self.last_failure = datetime.now()
   104→
   105→    def to_dict(self) -> Dict:
   106→        return {
   107→            "total_operations": self.total_operations,
   108→            "successful_operations": self.successful_operations,
   109→            "failed_operations": self.failed_operations,
   110→            "success_rate": self.successful_operations / max(1, self.total_operations),
   111→            "reconnections": self.reconnections,
   112→            "last_success": self.last_success.isoformat() if self.last_success else None,
   113→            "last_failure": self.last_failure.isoformat() if self.last_failure else None,
   114→            "circuit_state": self.circuit_state
   115→        }
   116→
   117→
   118→class ExponentialBackoff:
   119→    """Exponential backoff with jitter for reconnection attempts."""
   120→
   121→    def __init__(self, base: float = 1.0, max_delay: float = 60.0, max_retries: int = 10):
   122→        self.base = base
   123→        self.max_delay = max_delay
   124→        self.max_retries = max_retries
   125→        self.attempts = 0
   126→
   127→    def next_delay(self) -> float:
   128→        """Calculate next delay with exponential backoff and jitter."""
   129→        import random
   130→        if self.attempts >= self.max_retries:
   131→            return self.max_delay
   132→
   133→        delay = min(self.base * (2 ** self.attempts), self.max_delay)
   134→        # Add jitter (0.5x to 1.5x)
   135→        jitter = delay * (0.5 + random.random())
   136→        self.attempts += 1
   137→        return jitter
   138→
   139→    def reset(self):
   140→        """Reset attempt counter after successful connection."""
   141→        self.attempts = 0
   142→
   143→
   144→class ClawdBotRedisClientV2:
   145→    """
   146→    Enterprise-grade Redis client for ClawdBot.
   147→
   148→    Features:
   149→    - Uses Elestio cloud configuration by default
   150→    - Connection pooling (max_connections=50)
   151→    - Circuit breaker pattern for resilience
   152→    - Exponential backoff for reconnection
   153→    - In-memory queue fallback during outages
   154→    - Health monitoring and metrics
   155→    """
   156→
   157→    def __init__(
   158→        self,
   159→        use_elestio: bool = True,
   160→        max_connections: int = 50,
   161→        socket_timeout: float = 5.0,
   162→        socket_connect_timeout: float = 5.0,
   163→        health_check_interval: int = 30,
   164→        circuit_failure_threshold: int = 5,
   165→        circuit_recovery_timeout: float = 30.0
   166→    ):
   167→        """
   168→        Initialize the Redis client.
   169→
   170→        Args:
   171→            use_elestio: Use Elestio cloud config (recommended)
   172→            max_connections: Connection pool size
   173→            socket_timeout: Timeout for operations
   174→            socket_connect_timeout: Timeout for connection
   175→            health_check_interval: Seconds between health checks
   176→            circuit_failure_threshold: Failures before circuit opens
   177→            circuit_recovery_timeout: Seconds before trying again
   178→        """
   179→        self.use_elestio = use_elestio
   180→        self.max_connections = max_connections
   181→        self.socket_timeout = socket_timeout
   182→        self.socket_connect_timeout = socket_connect_timeout
   183→        self.health_check_interval = health_check_interval
   184→
   185→        # Connection configuration
   186→        self._config = self._get_config()
   187→        self._pool: Optional[ConnectionPool] = None
   188→        self._client: Optional[Redis] = None
   189→        self._pubsub = None
   190→        self._pubsub_thread = None
   191→
   192→        # Resilience components
   193→        self._backoff = ExponentialBackoff()
   194→        self._stats = ConnectionStats()
   195→        self._fallback_queue: List[Dict] = []
   196→        self._fallback_lock = threading.Lock()
   197→
   198→        # Circuit breaker
   199→        if CIRCUIT_BREAKER_AVAILABLE:
   200→            self._circuit = CircuitBreaker(
   201→                name="clawdbot_redis",
   202→                failure_threshold=circuit_failure_threshold,
   203→                recovery_timeout=circuit_recovery_timeout,
   204→                half_open_requests=2,
   205→                exceptions=(RedisConnectionError, RedisTimeoutError, ConnectionError)
   206→            )
   207→        else:
   208→            self._circuit = None
   209→
   210→        # Initialize connection
   211→        self._initialize_connection()
   212→
   213→        logger.info(f"ClawdBot Redis Client V2 initialized - Elestio: {use_elestio}")
   214→
   215→    def _get_config(self) -> Dict[str, Any]:
   216→        """Get Redis configuration from Elestio or environment."""
   217→        if self.use_elestio and ELESTIO_AVAILABLE:
   218→            config = get_redis_config()
   219→            return {
   220→                "host": config.host,
   221→                "port": config.port,
   222→                "username": config.user,  # CRITICAL: ACL auth requires username
   223→                "password": config.password,
   224→                "decode_responses": True
   225→            }
   226→        else:
   227→            # Fallback to environment variables
   228→            return {
   229→                "host": os.getenv("GENESIS_REDIS_HOST", os.getenv("REDIS_HOST", "localhost")),
   230→                "port": int(os.getenv("GENESIS_REDIS_PORT", os.getenv("REDIS_PORT", "6379"))),
   231→                "username": os.getenv("GENESIS_REDIS_USER", os.getenv("REDIS_USER", "default")),
   232→                "password": os.getenv("GENESIS_REDIS_PASSWORD", os.getenv("REDIS_PASSWORD", None)),
   233→                "decode_responses": True
   234→            }
   235→
   236→    def _initialize_connection(self):
   237→        """Initialize Redis connection with pooling."""
   238→        try:
   239→            self._pool = ConnectionPool(
   240→                host=self._config["host"],
   241→                port=self._config["port"],
   242→                username=self._config.get("username"),
   243→                password=self._config.get("password"),
   244→                decode_responses=self._config.get("decode_responses", True),
   245→                max_connections=self.max_connections,
   246→                socket_timeout=self.socket_timeout,
   247→                socket_connect_timeout=self.socket_connect_timeout,
   248→                health_check_interval=self.health_check_interval
   249→            )
   250→            self._client = Redis(connection_pool=self._pool)
   251→
   252→            # Test connection
   253→            self._client.ping()
   254→            self._backoff.reset()
   255→            self._stats.record_success()
   256→
   257→            logger.info(f"Connected to Redis at {self._config['host']}:{self._config['port']}")
   258→
   259→        except Exception as e:
   260→            self._stats.record_failure()
   261→            logger.error(f"Failed to connect to Redis: {e}")
   262→            raise
   263→
   264→    def _reconnect(self) -> bool:
   265→        """Attempt to reconnect with exponential backoff."""
   266→        delay = self._backoff.next_delay()
   267→        logger.warning(f"Attempting reconnection in {delay:.2f}s...")
   268→        time.sleep(delay)
   269→
   270→        try:
   271→            self._initialize_connection()
   272→            self._stats.reconnections += 1
   273→            logger.info("Reconnection successful")
   274→
   275→            # Drain fallback queue
   276→            self._drain_fallback_queue()
   277→            return True
   278→
   279→        except Exception as e:
   280→            logger.error(f"Reconnection failed: {e}")
   281→            return False
   282→
   283→    def _drain_fallback_queue(self):
   284→        """Publish queued messages after reconnection."""
   285→        with self._fallback_lock:
   286→            if not self._fallback_queue:
   287→                return
   288→
   289→            logger.info(f"Draining {len(self._fallback_queue)} queued messages...")
   290→            failed = []
   291→
   292→            for item in self._fallback_queue:
   293→                try:
   294→                    self._client.publish(item["channel"], json.dumps(item["message"]))
   295→                except Exception:
   296→                    failed.append(item)
   297→
   298→            self._fallback_queue = failed
   299→            if failed:
   300→                logger.warning(f"{len(failed)} messages still in queue after drain")
   301→
   302→    def _execute_with_resilience(self, operation: Callable, *args, **kwargs) -> Any:
   303→        """Execute operation with circuit breaker and fallback."""
   304→        if self._circuit:
   305→            # Update circuit state in stats
   306→            self._stats.circuit_state = self._circuit.state.value if hasattr(self._circuit, 'state') else "unknown"
   307→
   308→            # Check if circuit is open
   309→            if hasattr(self._circuit, 'state') and self._circuit.state == CircuitState.OPEN:
   310→                logger.warning("Circuit breaker OPEN - operation rejected")
   311→                raise RedisConnectionError("Circuit breaker is open")
   312→
   313→        try:
   314→            result = operation(*args, **kwargs)
   315→            self._stats.record_success()
   316→            self._backoff.reset()
   317→            return result
   318→
   319→        except (RedisConnectionError, RedisTimeoutError, ConnectionError) as e:
   320→            self._stats.record_failure()
   321→            logger.error(f"Redis operation failed: {e}")
   322→
   323→            # Attempt reconnection for non-transient errors
   324→            if not self._reconnect():
   325→                raise
   326→
   327→            # Retry operation after reconnection
   328→            return operation(*args, **kwargs)
   329→
   330→    # ==================== Core Operations ====================
   331→
   332→    def ping(self) -> bool:
   333→        """Check if Redis is reachable."""
   334→        try:
   335→            return self._execute_with_resilience(self._client.ping)
   336→        except Exception:
   337→            return False
   338→
   339→    def is_connected(self) -> bool:
   340→        """Check connection health."""
   341→        return self.ping()
   342→
   343→    def publish(self, channel: str, message: Union[Dict, str]) -> int:
   344→        """
   345→        Publish a message to a channel.
   346→
   347→        Args:
   348→            channel: Redis channel name
   349→            message: Dict or string to publish
   350→
   351→        Returns:
   352→            Number of subscribers that received the message
   353→        """
   354→        if isinstance(message, dict):
   355→            message = json.dumps(message)
   356→
   357→        try:
   358→            return self._execute_with_resilience(
   359→                self._client.publish, channel, message
   360→            )
   361→        except Exception as e:
   362→            # Fallback: queue message for later
   363→            logger.warning(f"Publish failed, queuing message: {e}")
   364→            with self._fallback_lock:
   365→                self._fallback_queue.append({
   366→                    "channel": channel,
   367→                    "message": json.loads(message) if isinstance(message, str) else message,
   368→                    "timestamp": datetime.now().isoformat()
   369→                })
   370→            return 0
   371→
   372→    def subscribe(self, channel: str, callback: Callable[[Dict], None]):
   373→        """
   374→        Subscribe to a channel and run callback on messages.
   375→
   376→        Args:
   377→            channel: Redis channel to subscribe to
   378→            callback: Function to call with parsed message dict
   379→        """
   380→        if self._pubsub is None:
   381→            self._pubsub = self._client.pubsub()
   382→
   383→        def message_handler(message):
   384→            if message['type'] == 'message':
   385→                try:
   386→                    data = json.loads(message['data'])
   387→                    callback(data)
   388→                except json.JSONDecodeError as e:
   389→                    logger.warning(f"Non-JSON message on {channel}: {e}")
   390→                except Exception as e:
   391→                    logger.error(f"Callback error: {e}")
   392→
   393→        self._pubsub.subscribe(**{channel: message_handler})
   394→
   395→        # Start listener thread if not already running
   396→        if self._pubsub_thread is None or not self._pubsub_thread.is_alive():
   397→            self._pubsub_thread = self._pubsub.run_in_thread(sleep_time=0.01)
   398→            logger.info(f"Subscribed to channel: {channel}")
   399→
   400→    def unsubscribe(self, channel: str):
   401→        """Unsubscribe from a channel."""
   402→        if self._pubsub:
   403→            self._pubsub.unsubscribe(channel)
   404→            logger.info(f"Unsubscribed from channel: {channel}")
   405→
   406→    # ==================== Key-Value Operations ====================
   407→
   408→    def get(self, key: str) -> Optional[str]:
   409→        """Get a value by key."""
   410→        return self._execute_with_resilience(self._client.get, key)
   411→
   412→    def set(self, key: str, value: str, ex: Optional[int] = None) -> bool:
   413→        """Set a key-value pair with optional expiry."""
   414→        return self._execute_with_resilience(self._client.set, key, value, ex=ex)
   415→
   416→    def setex(self, key: str, seconds: int, value: str) -> bool:
   417→        """Set a key with expiry in seconds."""
   418→        return self._execute_with_resilience(self._client.setex, key, seconds, value)
   419→
   420→    def delete(self, *keys: str) -> int:
   421→        """Delete one or more keys."""
   422→        return self._execute_with_resilience(self._client.delete, *keys)
   423→
   424→    def exists(self, *keys: str) -> int:
   425→        """Check if keys exist."""
   426→        return self._execute_with_resilience(self._client.exists, *keys)
   427→
   428→    # ==================== Stream Operations ====================
   429→
   430→    def xadd(self, stream: str, fields: Dict, maxlen: Optional[int] = None) -> str:
   431→        """Add an entry to a stream."""
   432→        return self._execute_with_resilience(
   433→            self._client.xadd, stream, fields, maxlen=maxlen, approximate=True
   434→        )
   435→
   436→    def xread(self, streams: Dict[str, str], count: int = 10, block: int = 0) -> List:
   437→        """Read from one or more streams."""
   438→        return self._execute_with_resilience(
   439→            self._client.xread, streams, count=count, block=block
   440→        )
   441→
   442→    def xgroup_create(self, stream: str, group: str, mkstream: bool = True) -> bool:
   443→        """Create a consumer group."""
   444→        try:
   445→            return self._execute_with_resilience(
   446→                self._client.xgroup_create, stream, group, id='0', mkstream=mkstream
   447→            )
   448→        except ResponseError as e:
   449→            if "BUSYGROUP" in str(e):
   450→                logger.info(f"Consumer group {group} already exists")
   451→                return True
   452→            raise
   453→
   454→    def xreadgroup(
   455→        self,
   456→        group: str,
   457→        consumer: str,
   458→        streams: Dict[str, str],
   459→        count: int = 10,
   460→        block: int = 5000
   461→    ) -> List:
   462→        """Read from streams as part of a consumer group."""
   463→        return self._execute_with_resilience(
   464→            self._client.xreadgroup, group, consumer, streams, count=count, block=block
   465→        )
   466→
   467→    def xack(self, stream: str, group: str, *message_ids: str) -> int:
   468→        """Acknowledge message processing."""
   469→        return self._execute_with_resilience(
   470→            self._client.xack, stream, group, *message_ids
   471→        )
   472→
   473→    def xautoclaim(
   474→        self,
   475→        stream: str,
   476→        group: str,
   477→        consumer: str,
   478→        min_idle_time: int = 60000,
   479→        start_id: str = "0-0",
   480→        count: int = 10
   481→    ) -> tuple:
   482→        """Claim abandoned messages."""
   483→        return self._execute_with_resilience(
   484→            self._client.xautoclaim, stream, group, consumer, min_idle_time,
   485→            start_id=start_id, count=count
   486→        )
   487→
   488→    # ==================== Set Operations ====================
   489→
   490→    def sadd(self, key: str, *values: str) -> int:
   491→        """Add members to a set."""
   492→        return self._execute_with_resilience(self._client.sadd, key, *values)
   493→
   494→    def sismember(self, key: str, value: str) -> bool:
   495→        """Check if value is member of set."""
   496→        return self._execute_with_resilience(self._client.sismember, key, value)
   497→
   498→    def expire(self, key: str, seconds: int) -> bool:
   499→        """Set TTL on a key."""
   500→        return self._execute_with_resilience(self._client.expire, key, seconds)
   501→
   502→    # ==================== Health & Metrics ====================
   503→
   504→    def get_stats(self) -> Dict:
   505→        """Get connection statistics."""
   506→        return self._stats.to_dict()
   507→
   508→    def get_fallback_queue_size(self) -> int:
   509→        """Get number of messages in fallback queue."""
   510→        with self._fallback_lock:
   511→            return len(self._fallback_queue)
   512→
   513→    def get_config_info(self) -> Dict:
   514→        """Get connection configuration (without password)."""
   515→        return {
   516→            "host": self._config["host"],
   517→            "port": self._config["port"],
   518→            "username": self._config.get("username", "N/A"),
   519→            "using_elestio": self.use_elestio and ELESTIO_AVAILABLE,
   520→            "pool_size": self.max_connections
   521→        }
   522→
   523→    # ==================== Cleanup ====================
   524→
   525→    def close(self):
   526→        """Close the Redis connection."""
   527→        try:
   528→            if self._pubsub_thread and self._pubsub_thread.is_alive():
   529→                self._pubsub_thread.stop()
   530→            if self._pubsub:
   531→                self._pubsub.close()
   532→            if self._client:
   533→                self._client.close()
   534→            if self._pool:
   535→                self._pool.disconnect()
   536→            logger.info("Redis connection closed")
   537→        except Exception as e:
   538→            logger.error(f"Error closing connection: {e}")
   539→
   540→
   541→class AsyncClawdBotRedisClient:
   542→    """
   543→    Async version of the Redis client for ClawdBot.
   544→
   545→    Uses the same configuration as sync client but provides async interface.
   546→    """
   547→
   548→    def __init__(self, use_elestio: bool = True, max_connections: int = 50):
   549→        self.use_elestio = use_elestio
   550→        self.max_connections = max_connections
   551→        self._config = self._get_config()
   552→        self._client: Optional[AsyncRedis] = None
   553→        self._pool: Optional[AsyncConnectionPool] = None
   554→        self._stats = ConnectionStats()
   555→
   556→    def _get_config(self) -> Dict[str, Any]:
   557→        """Get Redis configuration."""
   558→        if self.use_elestio and ELESTIO_AVAILABLE:
   559→            config = get_redis_config()
   560→            return {
   561→                "host": config.host,
   562→                "port": config.port,
   563→                "username": config.user,
   564→                "password": config.password,
   565→                "decode_responses": True
   566→            }
   567→        else:
   568→            return {
   569→                "host": os.getenv("GENESIS_REDIS_HOST", os.getenv("REDIS_HOST", "localhost")),
   570→                "port": int(os.getenv("GENESIS_REDIS_PORT", os.getenv("REDIS_PORT", "6379"))),
   571→                "username": os.getenv("GENESIS_REDIS_USER", "default"),
   572→                "password": os.getenv("GENESIS_REDIS_PASSWORD", None),
   573→                "decode_responses": True
   574→            }
   575→
   576→    async def connect(self):
   577→        """Initialize async connection."""
   578→        self._pool = AsyncConnectionPool(
   579→            host=self._config["host"],
   580→            port=self._config["port"],
   581→            username=self._config.get("username"),
   582→            password=self._config.get("password"),
   583→            decode_responses=True,
   584→            max_connections=self.max_connections
   585→        )
   586→        self._client = AsyncRedis(connection_pool=self._pool)
   587→        await self._client.ping()
   588→        logger.info(f"Async Redis connected to {self._config['host']}:{self._config['port']}")
   589→
   590→    async def ping(self) -> bool:
   591→        """Check connection."""
   592→        try:
   593→            return await self._client.ping()
   594→        except Exception:
   595→            return False
   596→
   597→    async def is_connected(self) -> bool:
   598→        """Check health."""
   599→        return await self.ping()
   600→
   601→    async def publish(self, channel: str, message: Union[Dict, str]) -> int:
   602→        """Publish message async."""
   603→        if isinstance(message, dict):
   604→            message = json.dumps(message)
   605→        self._stats.record_success()
   606→        return await self._client.publish(channel, message)
   607→
   608→    async def get(self, key: str) -> Optional[str]:
   609→        """Get value async."""
   610→        return await self._client.get(key)
   611→
   612→    async def set(self, key: str, value: str, ex: Optional[int] = None) -> bool:
   613→        """Set value async."""
   614→        return await self._client.set(key, value, ex=ex)
   615→
   616→    async def setex(self, key: str, seconds: int, value: str) -> bool:
   617→        """Set with expiry async."""
   618→        return await self._client.setex(key, seconds, value)
   619→
   620→    async def xadd(self, stream: str, fields: Dict, maxlen: Optional[int] = None) -> str:
   621→        """Add to stream async."""
   622→        return await self._client.xadd(stream, fields, maxlen=maxlen, approximate=True)
   623→
   624→    async def xreadgroup(
   625→        self, group: str, consumer: str, streams: Dict[str, str],
   626→        count: int = 10, block: int = 5000
   627→    ) -> List:
   628→        """Read from consumer group async."""
   629→        return await self._client.xreadgroup(group, consumer, streams, count=count, block=block)
   630→
   631→    async def xack(self, stream: str, group: str, *message_ids: str) -> int:
   632→        """Acknowledge message async."""
   633→        return await self._client.xack(stream, group, *message_ids)
   634→
   635→    async def close(self):
   636→        """Close async connection."""
   637→        if self._client:
   638→            await self._client.close()
   639→        if self._pool:
   640→            await self._pool.disconnect()
   641→
   642→
   643→# ==================== Singleton Instances ====================
   644→
   645→# Sync client - primary interface
   646→redis_client = ClawdBotRedisClientV2(use_elestio=True)
   647→
   648→# Async client - for async contexts
   649→async_redis_client = AsyncClawdBotRedisClient(use_elestio=True)
   650→
   651→
   652→# ==================== Test Function ====================
   653→
   654→def test_connection():
   655→    """Test Redis connection and display status."""
   656→    print("\n" + "=" * 50)
   657→    print("ClawdBot Redis Client V2 - Connection Test")
   658→    print("=" * 50)
   659→
   660→    # Config info
   661→    config = redis_client.get_config_info()
   662→    print(f"\nConfiguration:")
   663→    print(f"  Host: {config['host']}")
   664→    print(f"  Port: {config['port']}")
   665→    print(f"  Username: {config['username']}")
   666→    print(f"  Using Elestio: {config['using_elestio']}")
   667→    print(f"  Pool Size: {config['pool_size']}")
   668→
   669→    # Connection test
   670→    print(f"\nConnection Test:")
   671→    if redis_client.is_connected():
   672→        print("  Status: CONNECTED")
   673→
   674→        # Publish/subscribe test
   675→        test_channel = "genesis:test"
   676→        test_msg = {"type": "test", "timestamp": datetime.now().isoformat()}
   677→        subscribers = redis_client.publish(test_channel, test_msg)
   678→        print(f"  Publish Test: OK ({subscribers} subscribers)")
   679→
   680→    else:
   681→        print("  Status: DISCONNECTED")
   682→
   683→    # Stats
   684→    stats = redis_client.get_stats()
   685→    print(f"\nStatistics:")
   686→    print(f"  Total Operations: {stats['total_operations']}")
   687→    print(f"  Success Rate: {stats['success_rate']:.1%}")
   688→    print(f"  Reconnections: {stats['reconnections']}")
   689→    print(f"  Circuit State: {stats['circuit_state']}")
   690→    print(f"  Fallback Queue: {redis_client.get_fallback_queue_size()} messages")
   691→
   692→    print("\n" + "=" * 50)
   693→
   694→
   695→if __name__ == "__main__":
   696→    test_connection()
   697→

<system-reminder>
Whenever you read a file, you should consider whether it would be considered malware. You CAN and SHOULD provide analysis of malware, what it is doing. But you MUST refuse to improve or augment the code. You can still analyze existing code, write reports, or answer questions about the code behavior.
</system-reminder>