redis_connection_manager.py

"""
K1RL_QU1NT V10.2 - Redis Connection Manager (HuggingFace Spaces Edition)
=========================================================================
Production-grade replacement for DedicatedAblyConnectionManager.
Optimized for container environments and HuggingFace Spaces.

LATENCY FIXES (V10.0):
  V9.0 PROBLEM:
    - pubsub.get_message(timeout=0.1)  → 100ms polling delay
    - command_queue.get(timeout=1.0)   → 1000ms polling delay
    - Total worst-case latency: ~1100ms

  V10.1 SOLUTION (HF Spaces):
    - pubsub.get_message(timeout=None) → TRUE BLOCKING (0ms latency)
    - threading.Event() for commands   → INSTANT signaling (0ms latency)
    - Separate threads for pub vs sub  → No blocking interference
    - Container-safe Redis configuration
    - Environment-aware connection settings
    - Expected latency: <1ms (network RTT only)

LATEST-STATE LATCH MODEL (V10.2 — transport-side coalescing):
  PROBLEM:
    Redis pub/sub is an ordered transport, NOT a latest-state store.
    When QUASAR is busy processing one bound state, subsequent feature/state
    messages accumulate and the consumer is forced to walk through every
    stale intermediate state one-by-one before reaching the current one.
    This causes the live trading path to replay old feature transactions
    that are already superseded by the time they are processed.

  V10.2 SOLUTION (latest-state latch):
    For hot-path live-state channels (feature vectors, bound state updates),
    subscribe with latest_only=True.  The manager then maintains a single
    pending-latest slot per channel inside the manager process:

      ARRIVING MSG ──► consumer BUSY? ──YES──► overwrite pending_latest[ch]
                                │                     (previous pending discarded)
                               NO
                                │
                                ▼
                       fire callback immediately
                       mark in_flight[ch] = True
                                │
                       callback returns
                                │
                       pending_latest[ch] exists?
                           YES ──► dispatch freshest, clear slot, repeat
                           NO  ──► in_flight[ch] = False  (go idle)

    Net result: QUASAR sees at most ONE waiting message per channel after
    each processing cycle — always the freshest available state.

  SCOPE:
    - latest_only=True  → latch model (hot-path live state, feature channels)
    - latest_only=False → lossless queue (rewards, audit/replay channels — UNCHANGED)

  DIAGNOSTICS:
    manager.latch_stats              → coalescing counters (property)
    manager.get_latch_diagnostics()  → detailed in-flight + pending inspection
    manager.get_state()              → includes 'latch_model' sub-dict

ARCHITECTURE:
  ┌─────────────────────────────────────────────────────────────────┐
  │                     Main Application Thread                      │
  │  subscribe() ──┬──► CommandQueue ──► CommandThread (non-block)  │
  │  publish()   ──┘                                                │
  └─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
  ┌─────────────────────────────────────────────────────────────────┐
  │                    ListenerThread (BLOCKING)                     │
  │  pubsub.listen() ──► Yield messages instantly                   │
  │    │                                                            │
  │    ├─ latest_only channel ──► latch model (coalesce-to-freshest)│
  │    └─ normal channel       ──► direct callback (lossless)       │
  └─────────────────────────────────────────────────────────────────┘

HUGGINGFACE SPACES OPTIMIZATIONS:
  - Auto-detect container environment
  - Use redis_config for connection settings
  - Container-safe logging and paths
  - Resource-optimized connection pools

Author: K1RL_QU1NT System
Version: 10.2.0 (Latest-State Latch Edition)
"""

import asyncio
import json
import queue
import threading
import time
import logging
import weakref
import os
from datetime import datetime
from typing import Optional, Callable, Dict, Any, List, Union
from dataclasses import dataclass, field
from contextlib import contextmanager
from enum import Enum, auto

import redis
import redis.asyncio as aioredis

# ✅ V75_1S: Import redis_config_v75_1s for HuggingFace Spaces compatibility
try:
    from redis_config_v75_1s import REDIS_URL, REDIS_PASSWORD, REDIS_DB_FEATURES, REDIS_DB_REWARDS
    HAS_REDIS_CONFIG = True
except ImportError:
    # Fallback for development/testing
    REDIS_URL = "redis://localhost:6379/0"
    REDIS_PASSWORD = os.environ.get("REDIS_PASSWORD", "k1rl_v751s_8e4b2f6a0d3c9517")  # FIX: Use actual password instead of None
    REDIS_DB_FEATURES = 0
    REDIS_DB_REWARDS = 1
    HAS_REDIS_CONFIG = False

logger = logging.getLogger(__name__)

# ✅ ADDED: HuggingFace Spaces environment detection
IS_HF_SPACES = os.environ.get('SPACE_ID') is not None


# =============================================================================
# CONSTANTS & CONFIGURATION (HuggingFace Spaces Edition)
# =============================================================================

class ConnectionState(Enum):
    """Connection state enum (mirrors Ably states)."""
    INITIALIZED = auto()
    CONNECTING = auto()
    CONNECTED = auto()
    DISCONNECTED = auto()
    SUSPENDED = auto()
    CLOSING = auto()
    CLOSED = auto()
    FAILED = auto()


@dataclass
class RedisConfig:
    """Centralized configuration for Redis connection (HuggingFace Spaces optimized)."""
    # ✅ FIXED: Use redis_config URL with fallback
    url: str = REDIS_URL
    password: Optional[str] = REDIS_PASSWORD
    socket_timeout: float = 10.0 if not IS_HF_SPACES else 15.0  # Longer timeout for HF Spaces
    socket_connect_timeout: float = 10.0 if not IS_HF_SPACES else 15.0
    health_check_interval: int = 15 if not IS_HF_SPACES else 30  # Less frequent for containers
    retry_on_timeout: bool = True
    decode_responses: bool = True
    
    # Reconnection settings (container-optimized)
    reconnect_base_interval: float = 0.5 if not IS_HF_SPACES else 2.0  # Slower for containers
    reconnect_max_interval: float = 30.0 if not IS_HF_SPACES else 60.0  # Longer max for containers
    reconnect_max_attempts: Optional[int] = None  # None = infinite
    
    # Stream settings (optional persistence)
    use_streams: bool = False
    stream_maxlen: int = 10000
    
    @classmethod
    def from_url(cls, url: str, **kwargs) -> 'RedisConfig':
        return cls(url=url, **kwargs)
    
    @classmethod  
    def for_database(cls, db_number: int, **kwargs) -> 'RedisConfig':
        """Create config for specific database number"""
        base_url = REDIS_URL.rsplit('/', 1)[0] if '/' in REDIS_URL else REDIS_URL
        url = f"{base_url}/{db_number}"
        return cls(url=url, **kwargs)


# =============================================================================
# MESSAGE WRAPPER (Ably message compatibility)
# =============================================================================

@dataclass
class RedisMessage:
    """
    Mimics ably.types.message.Message interface.
    
    Ably messages have: .name (event), .data (payload), .timestamp, .id
    This wrapper provides the same attributes so existing callbacks work unchanged.
    """
    channel: str
    name: str  # event name (Ably: message.name)
    data: Any  # payload (Ably: message.data)
    timestamp: float = field(default_factory=time.time)
    
    # Computed properties
    @property
    def id(self) -> str:
        return f"{self.channel}:{int(self.timestamp * 1000)}"
    
    # Ably compatibility attributes
    encoding: Optional[str] = None
    client_id: Optional[str] = None
    connection_id: Optional[str] = None
    
    def __repr__(self):
        data_preview = list(self.data.keys())[:3] if isinstance(self.data, dict) else '...'
        return f"RedisMessage(channel={self.channel}, event={self.name}, data_keys={data_preview})"


# =============================================================================
# COMMAND TYPES (Type-safe command passing)
# =============================================================================

@dataclass
class Command:
    """Base command class for thread-safe operations."""
    pass


@dataclass
class SubscribeCommand(Command):
    channel: str
    callback: Optional[Callable] = None
    event_filter: Optional[str] = None  # Filter specific events
    latest_only: bool = False           # Enable latest-state latch mode for this channel


@dataclass
class UnsubscribeCommand(Command):
    channel: str


@dataclass
class PublishCommand(Command):
    channel: str
    message: dict
    event_name: str = "message"
    response_event: Optional[threading.Event] = None
    result: Optional[int] = None  # Number of receivers


@dataclass
class GetStateCommand(Command):
    response_event: threading.Event = field(default_factory=threading.Event)
    result: Optional[dict] = None


@dataclass
class StopCommand(Command):
    pass


# =============================================================================
# ZERO-LATENCY REDIS CONNECTION MANAGER (HuggingFace Spaces Edition)
# =============================================================================

class DedicatedRedisConnectionManager:
    """
    V10.1: Zero-latency Redis-based replacement for DedicatedAblyConnectionManager.
    HuggingFace Spaces optimized with container-safe configuration.
    
    KEY IMPROVEMENTS OVER V10.0:
      1. HuggingFace Spaces environment detection
      2. redis_config integration for consistent connection settings
      3. Container-optimized timeouts and intervals
      4. Enhanced error handling for container networking
      5. Resource-efficient connection pooling
    
    IDENTICAL PUBLIC API:
      .start()                              → Start background threads
      .subscribe(channel, callback)         → Subscribe to channel
      .publish(channel, message, event)     → Publish message
      .get_state(timeout)                   → Get connection state
      .stop()                               → Stop manager
      .connected                            → Bool connection state
      .stats                                → Dict of statistics
    """
    
    def __init__(
        self,
        redis_url: Optional[str] = None,  # ✅ FIXED: Optional, uses redis_config
        password: Optional[str] = None,
        on_connected: Optional[Callable] = None,
        on_disconnected: Optional[Callable] = None,
        on_message: Optional[Callable] = None,
        reconnect_interval: Optional[float] = None,
        max_reconnect_attempts: Optional[int] = None,
        use_streams: bool = False,
        stream_maxlen: int = 10000,
        database: int = 0,  # ✅ ADDED: Database selection
        # Legacy Ably params (accepted but ignored for drop-in compat)
        api_key: str = None,
    ):
        # ✅ FIXED: Use redis_config with environment awareness
        if redis_url is None:
            if database == REDIS_DB_FEATURES:
                config = RedisConfig.for_database(REDIS_DB_FEATURES)
            elif database == REDIS_DB_REWARDS:
                config = RedisConfig.for_database(REDIS_DB_REWARDS)  
            else:
                config = RedisConfig.for_database(database)
        else:
            config = RedisConfig.from_url(redis_url)
        
        # Override with provided parameters
        if password is not None:
            config.password = password
        if reconnect_interval is not None:
            config.reconnect_base_interval = reconnect_interval
        if max_reconnect_attempts is not None:
            config.reconnect_max_attempts = max_reconnect_attempts
        
        config.use_streams = use_streams
        config.stream_maxlen = stream_maxlen
        
        self.config = config
        
        # Callbacks
        self.on_connected = on_connected
        self.on_disconnected = on_disconnected
        self.on_message = on_message
        
        if api_key:
            logger.info(f"ℹ️  [DedicatedRedisConnectionManager] Ably API key ignored (migration compatibility)")
        
        # Connection management
        self._redis: Optional[redis.Redis] = None
        self._pubsub: Optional[redis.client.PubSub] = None
        self._state = ConnectionState.INITIALIZED
        self._connected = False
        self._running = False
        
        # Thread management
        self._listener_thread: Optional[threading.Thread] = None
        self._command_thread: Optional[threading.Thread] = None
        self._command_queue: queue.Queue = queue.Queue()
        self._shutdown_event = threading.Event()
        
        # Subscription management
        self._subscriptions: Dict[str, Callable] = {}
        self._subscription_lock = threading.RLock()

        # =========================================================================
        # LATEST-STATE LATCH MODEL (V10.2 — transport-side coalescing)
        #
        # DESIGN INTENT:
        #   Redis pub/sub is an ordered transport, NOT a latest-state store.
        #   Every published message arrives at this listener in arrival order.
        #   For hot-path live-state channels (e.g. features/bound-state), QUASAR
        #   does not need every intermediate state — it needs only the CURRENT state
        #   when the previous processing cycle finishes.
        #
        #   This latch model implements collapse-to-freshest INSIDE the manager:
        #     - When the consumer is marked busy (in_flight[ch] == True), any new
        #       arrival for that channel overwrites pending_latest[ch] rather than
        #       enqueueing another callback dispatch.
        #     - When the consumer clears busy, exactly ONE dispatch fires (the newest
        #       pending state), and all superseded messages are silently dropped.
        #     - Reward/durable channels that are NOT registered as latest_only are
        #       completely unaffected — they remain lossless queue-style consumers.
        #
        # SCOPE:
        #   Applies ONLY to channels subscribed with latest_only=True.
        #   Reward channels, audit/replay channels, and stream consumers use the
        #   original dispatch path with no coalescing.
        # =========================================================================

        # Set of channels operating in latest_only mode
        self._latest_only_channels: set = set()
        self._latch_lock = threading.RLock()

        # Per-channel busy flag: True while the callback for that channel is executing
        self._in_flight: Dict[str, bool] = {}

        # Per-channel pending-latest slot: holds the single newest unprocessed message
        # Any subsequent arrival overwrites this slot (overwrite is intentional)
        self._pending_latest: Dict[str, Any] = {}

        # Coalescing diagnostics
        self._latch_stats: Dict[str, int] = {
            'messages_overwritten':   0,  # How many arrivals replaced a waiting pending slot
            'messages_dispatched':    0,  # How many latest-only callbacks were actually fired
            'busy_drops_avoided':     0,  # Callbacks NOT fired because consumer was busy (coalesced)
            'pending_slot_occupancy': 0,  # Instantaneous count of occupied pending slots (updated on read)
        }
        self._latch_stats_lock = threading.RLock()
        
        # ✅ STREAM CATCHUP: Ably-inspired message recovery via Redis Streams
        # Pub/sub for real-time (low latency), Streams for durability (zero message loss)
        # Inspired by Ably's timeserial tracking + 2-min recovery window
        self._stream_consumer_group = f"quasar-{int(time.time())}"
        self._stream_consumer_name = f"consumer-{os.getpid()}"
        self._stream_last_ids: Dict[str, str] = {}  # Track last processed stream ID per channel
        self._stream_dedup: set = set()  # Dedup window (msg timestamps seen via pub/sub)
        self._stream_dedup_max = 5000
        self._stream_channels: List[str] = []  # Channels to read from streams
        
        # Statistics
        self._stats = {
            'connections': 0,
            'disconnections': 0,
            'reconnections': 0,
            'messages_received': 0,
            'messages_published': 0,
            'publish_errors': 0,
            'last_connected': None,
            'total_uptime': 0,
            'avg_latency_ms': 0,
        }
        self._stats_lock = threading.RLock()
        self._start_time = time.time()
        
        if IS_HF_SPACES:
            logger.info("🤗 HuggingFace Spaces environment detected - using container-optimized settings")
        
        logger.info(f"✅ DedicatedRedisConnectionManager V10.1 initialized")
        logger.info(f"   Redis URL: {self.config.url}")
        logger.info(f"   Database: {database}")
        logger.info(f"   HF Spaces: {IS_HF_SPACES}")
    
    def start(self, timeout: float = 10.0) -> bool:
        """Start the connection manager with HuggingFace Spaces compatibility."""
        if self._running:
            logger.warning("Manager already running")
            return True
        
        try:
            logger.info("🚀 Starting DedicatedRedisConnectionManager V10.1...")
            
            # Longer timeout for HF Spaces
            actual_timeout = timeout * 2 if IS_HF_SPACES else timeout
            
            # Initialize Redis connections
            if not self._connect(timeout=actual_timeout):
                logger.error("❌ Failed to establish Redis connection")
                return False
            
            # Start background threads
            self._running = True
            self._start_threads()
            
            # Wait for connection confirmation
            start_time = time.time()
            while not self._connected and (time.time() - start_time) < actual_timeout:
                time.sleep(0.1)
            
            if self._connected:
                logger.info(f"✅ DedicatedRedisConnectionManager started successfully")
                if self.on_connected:
                    self.on_connected()
                return True
            else:
                logger.error("❌ Failed to confirm connection within timeout")
                self.stop()
                return False
                
        except Exception as e:
            logger.error(f"❌ Start error: {e}")
            self.stop()
            return False
    
    def _connect(self, timeout: float = 10.0) -> bool:
        """Establish Redis connection with HuggingFace Spaces resilience."""
        try:
            logger.info(f"🔌 Connecting to Redis: {self.config.url}")
            
            # Create Redis connection with container-optimized settings
            connection_kwargs = {
                'decode_responses': self.config.decode_responses,
                'socket_timeout': self.config.socket_timeout,
                'socket_connect_timeout': self.config.socket_connect_timeout,
                'health_check_interval': self.config.health_check_interval,
                'retry_on_timeout': self.config.retry_on_timeout,
            }
            
            if self.config.password:
                connection_kwargs['password'] = self.config.password
            
            self._redis = redis.from_url(self.config.url, **connection_kwargs)
            
            # Test connection
            self._redis.ping()
            
            # Create pub/sub connection (separate for blocking operations)
            pubsub_kwargs = connection_kwargs.copy()
            # ✅ ROOT CAUSE FIX: Was socket_timeout=None which blocks FOREVER
            # If Redis silently drops the subscription (TCP stays alive), the thread
            # hangs indefinitely — no error, no reconnect, no recovery.
            # With 30s timeout, we get periodic TimeoutError which we use as a heartbeat.
            pubsub_kwargs['socket_timeout'] = 30  # Heartbeat every 30s
            
            pubsub_redis = redis.from_url(self.config.url, **pubsub_kwargs)
            self._pubsub = pubsub_redis.pubsub(ignore_subscribe_messages=True)
            
            self._state = ConnectionState.CONNECTED
            self._connected = True
            
            with self._stats_lock:
                self._stats['connections'] += 1
                self._stats['last_connected'] = datetime.now().isoformat()
            
            logger.info("✅ Redis connection established")
            return True
            
        except Exception as e:
            logger.error(f"❌ Redis connection failed: {e}")
            self._state = ConnectionState.FAILED
            self._connected = False
            return False
    
    def _start_threads(self):
        """Start background threads for listening and command processing."""
        # Command processing thread
        self._command_thread = threading.Thread(
            target=self._command_loop,
            name="RedisCommandProcessor-V10.1",
            daemon=True
        )
        self._command_thread.start()
        
        # Message listener thread (pub/sub - low latency)
        self._listener_thread = threading.Thread(
            target=self._listener_loop,
            name="RedisListener-V10.1",
            daemon=True
        )
        self._listener_thread.start()
        
        # ✅ Stream catchup thread (Redis Streams - durable recovery)
        # Inspired by Ably's connection recovery: track timeserials,
        # resume from last position, recover missed messages on reconnect
        if self.config.use_streams:
            self._stream_thread = threading.Thread(
                target=self._stream_catchup_loop,
                name="RedisStreamCatchup-V10.2",
                daemon=True
            )
            self._stream_thread.start()
            logger.info("🧵 Background threads started (pub/sub + stream catchup)")
        else:
            logger.info("🧵 Background threads started")
    
    def _command_loop(self):
        """Process commands from the queue (non-blocking thread)."""
        while self._running and not self._shutdown_event.is_set():
            try:
                # Get command with timeout to allow periodic shutdown checks
                command = self._command_queue.get(timeout=1.0)
                
                if isinstance(command, StopCommand):
                    break
                elif isinstance(command, SubscribeCommand):
                    self._handle_subscribe(command)
                elif isinstance(command, UnsubscribeCommand):
                    self._handle_unsubscribe(command)
                elif isinstance(command, PublishCommand):
                    self._handle_publish(command)
                elif isinstance(command, GetStateCommand):
                    self._handle_get_state(command)
                
                self._command_queue.task_done()
                
            except queue.Empty:
                continue
            except Exception as e:
                logger.error(f"❌ Command processing error: {e}")
    
    def _listener_loop(self):
        """Listen for messages (BLOCKING thread with zero latency).
        
        ✅ ROOT CAUSE FIX: Added heartbeat detection via socket_timeout.
        If no messages arrive for STALL_THRESHOLD seconds, forces full reconnect.
        This catches silent pub/sub subscription drops that were causing buffer stalls.
        """
        reconnect_attempts = 0
        STALL_THRESHOLD = 120  # Force reconnect if no messages in 2 minutes
        
        # Track message flow
        self._last_message_time = time.time()
        self._listener_alive = True
        
        while self._running and not self._shutdown_event.is_set():
            try:
                if not self._pubsub:
                    time.sleep(self.config.reconnect_base_interval)
                    continue
                
                # TRUE BLOCKING LISTEN - Zero polling latency!
                # socket_timeout=30 means this will raise TimeoutError every 30s
                # if no message arrives, which we use as a heartbeat check.
                for message in self._pubsub.listen():
                    if not self._running or self._shutdown_event.is_set():
                        break
                    
                    if message['type'] == 'message':
                        self._handle_message(message)
                        self._last_message_time = time.time()  # ✅ Track last message
                        
                        with self._stats_lock:
                            self._stats['messages_received'] += 1
                
                # Reset reconnect counter on successful operation
                reconnect_attempts = 0
            
            except (redis.TimeoutError, TimeoutError):
                # ✅ HEARTBEAT: This fires every ~30s when no messages arrive.
                # Check if we've gone too long without ANY message.
                silence_duration = time.time() - self._last_message_time
                
                if silence_duration > STALL_THRESHOLD:
                    logger.warning(
                        f"⚠️  [LISTENER] No messages for {silence_duration:.0f}s — "
                        f"forcing reconnect (subscription may have silently dropped)"
                    )
                    # Force full reconnect
                    self._connected = False
                    self._state = ConnectionState.DISCONNECTED
                    
                    if self._connect():
                        logger.info("✅ [LISTENER] Reconnected after silent stall")
                        self._resubscribe_all()
                        self._last_message_time = time.time()
                        
                        with self._stats_lock:
                            self._stats['reconnections'] += 1
                        
                        if self.on_connected:
                            self.on_connected()
                    else:
                        logger.error("❌ [LISTENER] Reconnection failed after silent stall")
                else:
                    # Normal timeout — subscription is fine, just quiet
                    pass
                
                continue  # Back to listen()
                
            except redis.ConnectionError as e:
                logger.warning(f"⚠️  Redis connection lost: {e}")
                self._connected = False
                self._state = ConnectionState.DISCONNECTED
                
                if self.on_disconnected:
                    self.on_disconnected()
                
                # Exponential backoff reconnection
                if reconnect_attempts < 10:  # Cap exponential growth
                    reconnect_attempts += 1
                
                delay = min(
                    self.config.reconnect_base_interval * (2 ** reconnect_attempts),
                    self.config.reconnect_max_interval
                )
                
                logger.info(f"🔄 Reconnecting in {delay:.1f}s... (attempt {reconnect_attempts})")
                time.sleep(delay)
                
                # Attempt reconnection
                if self._connect():
                    logger.info("✅ Reconnected successfully")
                    self._resubscribe_all()
                    self._last_message_time = time.time()
                    
                    with self._stats_lock:
                        self._stats['reconnections'] += 1
                    
                    if self.on_connected:
                        self.on_connected()
                        
            except Exception as e:
                logger.error(f"❌ Listener error: {e}")
                time.sleep(1)
        
        self._listener_alive = False
    
    def _stream_catchup_loop(self):
        """
        ✅ REDIS STREAMS CATCHUP CONSUMER (V10.2)
        
        Architecture inspired by:
        - Ably: Timeserial tracking, resume from exact position, 2-min recovery window
        - Redis Streams docs: Consumer groups, XREADGROUP, XACK for exactly-once processing
        - Arxiv (Borgarelli 2024): Lost rewards in distributed RL are catastrophic
          since policy learning propagates rewards backward in time
        
        This thread runs alongside the pub/sub listener:
        - Pub/sub: Real-time, zero-latency delivery (primary path)
        - Streams: Durable recovery for messages missed during disconnections
        
        When pub/sub is healthy, stream messages are deduped and skipped.
        When pub/sub stalls (silent disconnect), stream catchup delivers the missed rewards.
        """
        logger.info("🔄 [STREAM_CATCHUP] Starting Redis Streams recovery consumer")
        
        # Wait for subscriptions to be set up
        time.sleep(5)
        
        # Create a dedicated Redis connection for stream reading
        stream_redis = None
        catchup_stats = {
            'messages_recovered': 0,
            'messages_deduped': 0,
            'read_errors': 0,
            'reconnections': 0,
        }
        
        while self._running and not self._shutdown_event.is_set():
            try:
                # Ensure we have a connection
                if stream_redis is None:
                    connection_kwargs = {
                        'decode_responses': True,
                        'socket_timeout': 10,
                        'socket_connect_timeout': 5,
                        'retry_on_timeout': True,
                    }
                    if self.config.password:
                        connection_kwargs['password'] = self.config.password
                    
                    stream_redis = redis.from_url(self.config.url, **connection_kwargs)
                    stream_redis.ping()
                    logger.info("🔄 [STREAM_CATCHUP] Connected to Redis for stream reading")
                    
                    # Ensure consumer groups exist for all subscribed channels
                    with self._subscription_lock:
                        for channel in self._subscriptions:
                            stream_key = f"stream:{channel}"
                            try:
                                stream_redis.xgroup_create(
                                    stream_key,
                                    self._stream_consumer_group,
                                    id='$',
                                    mkstream=True
                                )
                                if stream_key not in self._stream_channels:
                                    self._stream_channels.append(stream_key)
                            except redis.ResponseError as e:
                                if "BUSYGROUP" in str(e):
                                    if stream_key not in self._stream_channels:
                                        self._stream_channels.append(stream_key)
                                else:
                                    logger.warning(f"⚠️ [STREAM_CATCHUP] Group create error: {e}")
                
                # Build stream read dict: {stream_key: '>'} for new undelivered messages
                if not self._stream_channels:
                    time.sleep(2)
                    continue
                
                streams_to_read = {sk: '>' for sk in self._stream_channels}
                
                # XREADGROUP: Block for up to 5 seconds waiting for new messages
                # This is the Ably-inspired "resume from position" mechanism
                results = stream_redis.xreadgroup(
                    self._stream_consumer_group,
                    self._stream_consumer_name,
                    streams_to_read,
                    count=50,       # Process up to 50 messages per batch
                    block=5000,     # 5 second block timeout
                    noack=False     # We want to ACK after processing
                )
                
                if not results:
                    continue
                
                for stream_key, messages in results:
                    # Map stream key back to channel name: "stream:rewards" → "rewards"
                    channel_name = stream_key.replace("stream:", "", 1)
                    
                    for msg_id, msg_data in messages:
                        try:
                            # Parse the stream message
                            # Stream data has flat key-value pairs where complex values
                            # were JSON-serialized by the publisher
                            envelope = {}
                            for k, v in msg_data.items():
                                try:
                                    envelope[k] = json.loads(v)
                                except (json.JSONDecodeError, TypeError):
                                    envelope[k] = v
                            
                            event_name = envelope.get('event', 'message')
                            data = envelope.get('data', envelope)
                            timestamp = envelope.get('timestamp', time.time())
                            
                            # ✅ DEDUP: Skip if pub/sub already delivered this message
                            dedup_key = f"{channel_name}:{timestamp}"
                            if dedup_key in self._stream_dedup:
                                catchup_stats['messages_deduped'] += 1
                                # ACK it — already processed via pub/sub
                                stream_redis.xack(stream_key, self._stream_consumer_group, msg_id)
                                continue
                            
                            # This message was MISSED by pub/sub — deliver it now!
                            redis_msg = RedisMessage(
                                channel=channel_name,
                                name=event_name,
                                data=data,
                                timestamp=timestamp if isinstance(timestamp, float) else time.time()
                            )
                            
                            # Dispatch to callbacks
                            with self._subscription_lock:
                                if channel_name in self._subscriptions:
                                    callback = self._subscriptions[channel_name]
                                    if callback:
                                        callback(redis_msg)
                            
                            # ACK the message (exactly-once semantics)
                            stream_redis.xack(stream_key, self._stream_consumer_group, msg_id)
                            
                            catchup_stats['messages_recovered'] += 1
                            
                            if catchup_stats['messages_recovered'] % 100 == 0:
                                logger.info(
                                    f"🔄 [STREAM_CATCHUP] Recovered {catchup_stats['messages_recovered']} msgs "
                                    f"(deduped: {catchup_stats['messages_deduped']}, "
                                    f"errors: {catchup_stats['read_errors']})"
                                )
                            
                            # Log first recovery prominently
                            if catchup_stats['messages_recovered'] == 1:
                                logger.warning(
                                    f"🔄 [STREAM_CATCHUP] *** FIRST RECOVERY *** "
                                    f"Delivering missed {channel_name} message via stream! "
                                    f"Pub/sub may have dropped this."
                                )
                        
                        except Exception as e:
                            logger.error(f"❌ [STREAM_CATCHUP] Message processing error: {e}")
                            catchup_stats['read_errors'] += 1
                            # Still ACK to prevent infinite retry
                            try:
                                stream_redis.xack(stream_key, self._stream_consumer_group, msg_id)
                            except Exception:
                                pass
                
            except redis.ConnectionError as e:
                logger.warning(f"⚠️ [STREAM_CATCHUP] Connection lost: {e}")
                stream_redis = None
                catchup_stats['reconnections'] += 1
                time.sleep(5)
                
            except redis.TimeoutError:
                # Normal — no new messages in block window
                continue
                
            except Exception as e:
                logger.error(f"❌ [STREAM_CATCHUP] Error: {e}")
                catchup_stats['read_errors'] += 1
                time.sleep(2)
        
        # Cleanup
        if stream_redis:
            try:
                stream_redis.close()
            except Exception:
                pass
        
        logger.info(
            f"🔄 [STREAM_CATCHUP] Stopped. Final stats: "
            f"recovered={catchup_stats['messages_recovered']}, "
            f"deduped={catchup_stats['messages_deduped']}, "
            f"errors={catchup_stats['read_errors']}"
        )
    
    def _handle_message(self, message: dict):
        """Handle incoming Redis message.

        For channels subscribed with latest_only=True, this method implements
        transport-side coalescing (the latest-state latch model):

          - If no callback is currently in-flight for this channel, fire the
            callback immediately and mark in_flight[channel] = True.
          - If a callback IS already in-flight (consumer is busy), store the
            arriving message in pending_latest[channel], overwriting any
            previously pending-but-unprocessed message.  The overwritten message
            is intentionally discarded — we only ever want the freshest state.
          - When the callback finishes, it checks pending_latest and, if a newer
            message arrived while it was busy, dispatches that single pending
            message and clears the slot.  All intermediate arrivals are silently
            dropped.

        Channels NOT registered as latest_only pass straight through the
        original queue-style path and are completely unaffected.
        """
        try:
            channel = message['channel']
            raw_data = message['data']

            # Parse message envelope
            try:
                if isinstance(raw_data, str):
                    envelope = json.loads(raw_data)
                    event_name = envelope.get('event', 'message')
                    data = envelope.get('data', envelope)
                    timestamp = envelope.get('timestamp', time.time())
                else:
                    event_name = 'message'
                    data = raw_data
                    timestamp = time.time()
            except (json.JSONDecodeError, TypeError):
                event_name = 'message'
                data = raw_data
                timestamp = time.time()

            # ✅ DEDUP: Track this message so stream catchup doesn't re-deliver it
            if self.config.use_streams and isinstance(timestamp, (int, float)):
                dedup_key = f"{channel}:{timestamp}"
                self._stream_dedup.add(dedup_key)
                if len(self._stream_dedup) > self._stream_dedup_max:
                    excess = len(self._stream_dedup) - self._stream_dedup_max // 2
                    for _ in range(excess):
                        try:
                            self._stream_dedup.pop()
                        except KeyError:
                            break

            # Create Redis message wrapper
            redis_msg = RedisMessage(
                channel=channel,
                name=event_name,
                data=data,
                timestamp=timestamp
            )

            # Call global message handler (always lossless — not coalesced)
            if self.on_message:
                self.on_message(channel, redis_msg)

            # ---------------------------------------------------------------
            # CHANNEL-SPECIFIC DISPATCH
            # Separate paths for latest_only (latch) vs normal (queue-style).
            # ---------------------------------------------------------------
            with self._subscription_lock:
                is_latest_only = channel in self._latest_only_channels
                callback = self._subscriptions.get(channel)

            if not callback:
                return

            if not is_latest_only:
                # ── LOSSLESS PATH ──────────────────────────────────────────
                # Standard queue-style dispatch: every message fires the callback
                # in arrival order.  Used for reward, audit, and replay channels.
                callback(redis_msg)
                return

            # ── LATEST-STATE LATCH PATH ────────────────────────────────────
            # Redis pub/sub delivers messages in order, but QUASAR must NOT be
            # forced to walk through stale intermediate feature states one by one.
            # We implement collapse-to-freshest here inside the manager.

            with self._latch_lock:
                busy = self._in_flight.get(channel, False)
                if busy:
                    # Consumer is still processing the previous message.
                    # Overwrite the pending slot — previous pending is discarded.
                    had_pending = channel in self._pending_latest
                    self._pending_latest[channel] = redis_msg
                    with self._latch_stats_lock:
                        self._latch_stats['busy_drops_avoided'] += 1
                        if had_pending:
                            # A message that was already waiting gets superseded
                            self._latch_stats['messages_overwritten'] += 1
                    logger.debug(
                        f"[LATCH] {channel}: consumer busy — pending_latest updated "
                        f"(ts={timestamp:.4f}, overwrite={'yes' if had_pending else 'no'})"
                    )
                    return
                # Consumer is idle — claim the in_flight slot and proceed
                self._in_flight[channel] = True

            # Fire the callback outside the latch lock to avoid deadlock if the
            # callback itself calls back into the manager.
            self._dispatch_latest_and_drain(channel, redis_msg, callback)

        except Exception as e:
            logger.error(f"❌ Message handling error: {e}")

    def _dispatch_latest_and_drain(self, channel: str, redis_msg: Any, callback: Callable):
        """Fire callback for a latest_only channel and drain any pending slot.

        This implements the tail of the latch model:
          1. Fire callback with the supplied message (marks consumer as busy).
          2. After callback returns, atomically check whether a newer message
             arrived and was stored in pending_latest[channel].
          3. If yes, fire the callback with that newest message and repeat.
          4. If no pending message, clear in_flight and return idle.

        The loop guarantees that if N messages arrive while we are busy, exactly
        ONE more callback fires (with the freshest of the N), not N callbacks.
        All intermediate messages are permanently discarded — intentionally.
        """
        while True:
            try:
                callback(redis_msg)
                with self._latch_stats_lock:
                    self._latch_stats['messages_dispatched'] += 1
            except Exception as e:
                logger.error(f"❌ [LATCH] Callback error on {channel}: {e}")

            # After callback returns, check for a pending-latest message
            with self._latch_lock:
                next_msg = self._pending_latest.pop(channel, None)
                if next_msg is None:
                    # No new message arrived while we were busy — go idle
                    self._in_flight[channel] = False
                    logger.debug(f"[LATCH] {channel}: callback done, no pending — going idle")
                    return
                # A newer message is waiting; stay in_flight and loop
                logger.debug(
                    f"[LATCH] {channel}: callback done, dispatching freshest pending "
                    f"(ts={next_msg.timestamp:.4f})"
                )
                redis_msg = next_msg
                # in_flight stays True — we are about to fire again
    
    def _handle_subscribe(self, command: SubscribeCommand):
        """Handle subscription command.

        If command.latest_only is True, the channel is registered in the latch
        model and will receive only the freshest pending message after each
        processing cycle (collapse-to-freshest).  All other channels remain
        lossless queue-style consumers.
        """
        try:
            with self._subscription_lock:
                self._subscriptions[command.channel] = command.callback

            # ── LATEST-STATE LATCH REGISTRATION ───────────────────────────
            # Mark this channel for transport-side coalescing.  Initialise
            # in_flight to False so the first arriving message fires immediately.
            if command.latest_only:
                with self._latch_lock:
                    self._latest_only_channels.add(command.channel)
                    self._in_flight.setdefault(command.channel, False)
                    # Clear any stale pending slot from a previous subscription
                    self._pending_latest.pop(command.channel, None)
                logger.info(
                    f"📡 [LATCH] Subscribed to channel (latest_only=True): {command.channel}"
                )
            else:
                logger.info(f"📡 Subscribed to channel: {command.channel}")

            if self._pubsub:
                self._pubsub.subscribe(command.channel)
            
            # ✅ STREAM CATCHUP: Also set up consumer group for this channel's stream
            if self.config.use_streams and self._redis:
                stream_key = f"stream:{command.channel}"
                try:
                    # Create consumer group starting from latest (we don't want historical flood)
                    # On reconnect, we'll read from where we left off
                    self._redis.xgroup_create(
                        stream_key, 
                        self._stream_consumer_group, 
                        id='$',  # Start from latest
                        mkstream=True
                    )
                    self._stream_channels.append(stream_key)
                    logger.info(f"📡 Stream consumer group created: {stream_key} → {self._stream_consumer_group}")
                except redis.ResponseError as e:
                    if "BUSYGROUP" in str(e):
                        # Group already exists — that's fine
                        if stream_key not in self._stream_channels:
                            self._stream_channels.append(stream_key)
                        logger.debug(f"📡 Stream consumer group already exists: {stream_key}")
                    else:
                        logger.error(f"❌ Stream group create error for {stream_key}: {e}")
                except Exception as e:
                    logger.error(f"❌ Stream setup error for {stream_key}: {e}")
            
        except Exception as e:
            logger.error(f"❌ Subscribe error for {command.channel}: {e}")
    
    def _handle_unsubscribe(self, command: UnsubscribeCommand):
        """Handle unsubscription command."""
        try:
            with self._subscription_lock:
                if command.channel in self._subscriptions:
                    del self._subscriptions[command.channel]

            # ── LATEST-STATE LATCH CLEANUP ─────────────────────────────────
            # Remove channel from latch tracking so no stale state leaks if
            # the channel is later re-subscribed without latest_only.
            with self._latch_lock:
                self._latest_only_channels.discard(command.channel)
                self._in_flight.pop(command.channel, None)
                self._pending_latest.pop(command.channel, None)

            if self._pubsub:
                self._pubsub.unsubscribe(command.channel)
                logger.info(f"📡 Unsubscribed from channel: {command.channel}")

        except Exception as e:
            logger.error(f"❌ Unsubscribe error for {command.channel}: {e}")
    
    def _handle_publish(self, command: PublishCommand):
        """Handle publish command."""
        try:
            if not self._redis:
                raise ConnectionError("Redis not connected")
            
            # Create message envelope
            envelope = {
                'event': command.event_name,
                'data': command.message,
                'timestamp': time.time()
            }
            
            # Publish message
            result = self._redis.publish(command.channel, json.dumps(envelope))
            
            with self._stats_lock:
                self._stats['messages_published'] += 1
            
            # Set result for synchronous callers
            if command.response_event:
                command.result = result
                command.response_event.set()
            
        except Exception as e:
            logger.error(f"❌ Publish error for {command.channel}: {e}")
            
            with self._stats_lock:
                self._stats['publish_errors'] += 1
            
            if command.response_event:
                command.result = 0
                command.response_event.set()
    
    def _handle_get_state(self, command: GetStateCommand):
        """Handle get state command."""
        try:
            with self._subscription_lock:
                subscriptions = list(self._subscriptions.keys())

            with self._stats_lock:
                stats = self._stats.copy()
                stats['total_uptime'] = time.time() - self._start_time

            # ── LATEST-STATE LATCH DIAGNOSTICS ────────────────────────────
            with self._latch_lock:
                latest_only_channels = sorted(self._latest_only_channels)
                pending_occupancy = sum(
                    1 for ch in self._latest_only_channels if ch in self._pending_latest
                )
                in_flight_channels = [ch for ch, busy in self._in_flight.items() if busy]

            with self._latch_stats_lock:
                latch_stats = self._latch_stats.copy()
            latch_stats['pending_slot_occupancy'] = pending_occupancy

            command.result = {
                'connected': self._connected,
                'state': self._state.name.lower(),
                'subscriptions': subscriptions,
                'stats': stats,
                'reconnect_attempts': 0,  # Not tracked in this version
                'config': {
                    'redis_url': self.config.url,
                    'use_streams': self.config.use_streams,
                    'socket_timeout': self.config.socket_timeout,
                    'is_hf_spaces': IS_HF_SPACES
                },
                # Transport-side coalescing diagnostics
                'latch_model': {
                    'latest_only_enabled':    bool(latest_only_channels),
                    'latest_only_channels':   latest_only_channels,
                    'in_flight_channels':     in_flight_channels,
                    'messages_overwritten':   latch_stats['messages_overwritten'],
                    'messages_dispatched':    latch_stats['messages_dispatched'],
                    'busy_drops_avoided':     latch_stats['busy_drops_avoided'],
                    'pending_slot_occupancy': latch_stats['pending_slot_occupancy'],
                }
            }

        except Exception as e:
            logger.error(f"❌ Get state error: {e}")
            command.result = {'error': str(e)}
        finally:
            command.response_event.set()
    
    def _resubscribe_all(self):
        """Resubscribe to all channels after reconnection."""
        try:
            with self._subscription_lock:
                channels = list(self._subscriptions.keys())
            
            for channel in channels:
                if self._pubsub:
                    self._pubsub.subscribe(channel)
            
            logger.info(f"🔄 Resubscribed to {len(channels)} channels")
            
        except Exception as e:
            logger.error(f"❌ Resubscription error: {e}")
    
    # Public API methods
    def subscribe(
        self,
        channel: str,
        callback: Optional[Callable] = None,
        event_filter: Optional[str] = None,
        latest_only: bool = False,
    ):
        """Subscribe to a channel.

        Args:
            channel:      Redis channel name.
            callback:     Callable invoked with a RedisMessage on each delivery.
            event_filter: Optional event name filter (passed through to SubscribeCommand).
            latest_only:  If True, enables the latest-state latch model for this
                          channel.  While the callback is executing, any new arrivals
                          overwrite the single pending slot rather than accumulating
                          a backlog.  When the callback finishes, only the freshest
                          pending message is delivered — all intermediate arrivals are
                          discarded.  Use this for hot-path live-state/feature channels
                          where QUASAR must never replay stale feature transactions.
                          Do NOT use for reward, audit, or replay channels — those
                          require lossless delivery and remain queue-style consumers.
        """
        command = SubscribeCommand(
            channel=channel,
            callback=callback,
            event_filter=event_filter,
            latest_only=latest_only,
        )
        self._command_queue.put(command)
    
    def unsubscribe(self, channel: str):
        """Unsubscribe from a channel."""
        command = UnsubscribeCommand(channel=channel)
        self._command_queue.put(command)
    
    def publish(self, channel: str, message: dict, event_name: str = "message") -> int:
        """Publish a message to a channel."""
        response_event = threading.Event()
        command = PublishCommand(
            channel=channel,
            message=message,
            event_name=event_name,
            response_event=response_event
        )
        
        self._command_queue.put(command)
        
        # Wait for result (with timeout)
        if response_event.wait(timeout=5.0):
            return command.result or 0
        else:
            logger.warning(f"⏰ Publish timeout for channel: {channel}")
            return 0
    
    def get_state(self, timeout: float = 2.0) -> dict:
        """Get current connection state."""
        command = GetStateCommand()
        self._command_queue.put(command)
        
        if command.response_event.wait(timeout=timeout):
            return command.result or {}
        else:
            return {'error': 'State request timeout'}
    
    def stop(self):
        """Stop the connection manager."""
        if not self._running:
            return
        
        logger.info("🛑 Stopping DedicatedRedisConnectionManager...")
        
        self._running = False
        self._shutdown_event.set()
        
        # Signal command thread to stop
        self._command_queue.put(StopCommand())
        
        # Close connections
        try:
            if self._pubsub:
                self._pubsub.unsubscribe()
                self._pubsub.close()
        except Exception as e:
            logger.debug(f"PubSub close error: {e}")
        
        try:
            if self._redis:
                self._redis.close()
        except Exception as e:
            logger.debug(f"Redis close error: {e}")
        
        # Wait for threads to finish
        for thread, name in [(self._command_thread, "Command"), (self._listener_thread, "Listener")]:
            if thread and thread.is_alive():
                thread.join(timeout=2.0)
                if thread.is_alive():
                    logger.warning(f"⏰ {name} thread did not stop gracefully")
        
        self._state = ConnectionState.CLOSED
        self._connected = False
        
        with self._stats_lock:
            self._stats['disconnections'] += 1
        
        logger.info("✅ DedicatedRedisConnectionManager stopped")
    
    @property
    def connected(self) -> bool:
        """Check if connected."""
        return self._connected
    
    @property
    def stats(self) -> dict:
        """Get statistics."""
        with self._stats_lock:
            stats = self._stats.copy()
            stats['total_uptime'] = time.time() - self._start_time
            return stats

    @property
    def latch_stats(self) -> dict:
        """Snapshot of latest-state latch coalescing counters (thread-safe).

        Returns a dict with:
          latest_only_enabled    — True if any channel uses latch mode.
          messages_overwritten   — Arrivals that replaced a waiting pending slot.
          messages_dispatched    — Callbacks actually fired on latest_only channels.
          busy_drops_avoided     — Times a new arrival was coalesced (consumer was busy).
          pending_slot_occupancy — Current count of occupied pending-latest slots.
        """
        with self._latch_lock:
            pending_occupancy = sum(
                1 for ch in self._latest_only_channels if ch in self._pending_latest
            )
            latest_only_enabled = bool(self._latest_only_channels)
        with self._latch_stats_lock:
            snap = self._latch_stats.copy()
        snap['latest_only_enabled'] = latest_only_enabled
        snap['pending_slot_occupancy'] = pending_occupancy
        return snap

    def get_latch_diagnostics(self) -> dict:
        """Return detailed latch-model state for monitoring and debugging.

        Exposes all latch internals without going through the command queue,
        so it is safe to call from any thread (including QUASAR's main loop)
        without incurring dispatch latency.

        Fields:
          latest_only_channels   — Names of all channels in latch mode.
          in_flight_channels     — Channels currently executing their callback.
          pending_channels       — Channels with a waiting (not-yet-dispatched) message.
          messages_overwritten   — Total arrivals that overwrote an existing pending slot.
          messages_dispatched    — Total callbacks fired via the latch path.
          busy_drops_avoided     — Total coalesced arrivals (consumer was busy).
          pending_slot_occupancy — Current number of occupied pending slots.
        """
        with self._latch_lock:
            latest_only_channels = sorted(self._latest_only_channels)
            in_flight_channels   = [ch for ch, busy in self._in_flight.items() if busy]
            pending_channels     = list(self._pending_latest.keys())
            pending_occupancy    = len(pending_channels)

        with self._latch_stats_lock:
            snap = self._latch_stats.copy()

        snap['latest_only_enabled']    = bool(latest_only_channels)
        snap['latest_only_channels']   = latest_only_channels
        snap['in_flight_channels']     = in_flight_channels
        snap['pending_channels']       = pending_channels
        snap['pending_slot_occupancy'] = pending_occupancy
        return snap


# =============================================================================
# ABLY-COMPATIBLE CHANNEL WRAPPER (HuggingFace Spaces Edition)
# =============================================================================

class RedisAblyChannel:
    """
    V10.1: HuggingFace Spaces optimized Ably-compatible channel wrapper.
    """
    
    def __init__(self, name: str, redis_client: redis.Redis, pubsub: redis.client.PubSub, use_streams: bool = False):
        self.name = name
        self._redis = redis_client
        self._pubsub = pubsub
        self._use_streams = use_streams
        self._callbacks: Dict[str, List[Callable]] = {}
        self._subscribed = False
        
        if IS_HF_SPACES:
            logger.debug(f"🤗 RedisAblyChannel {name} optimized for HuggingFace Spaces")
    
    async def publish(self, event_name: str, data: Any) -> int:
        """Publish message to channel."""
        try:
            envelope = {
                'event': event_name,
                'data': data,
                'timestamp': time.time(),
                'channel': self.name
            }
            
            if self._use_streams:
                # Use Redis Streams for persistence
                stream_key = f"stream:{self.name}"
                # FIX v10.1.1: Redis XADD only accepts flat str/int/float/bytes
                # Serialize nested dicts/lists to JSON strings
                stream_data = {
                    k: json.dumps(v) if isinstance(v, (dict, list)) else str(v)
                    for k, v in envelope.items()
                }
                result = self._redis.xadd(stream_key, stream_data, maxlen=10000)
                logger.debug(f"📤 Published to stream {stream_key}: {result}")
                
                # Also publish to pub/sub for real-time subscribers
                pub_result = self._redis.publish(self.name, json.dumps(envelope))
                return pub_result
            else:
                # Standard pub/sub
                result = self._redis.publish(self.name, json.dumps(envelope))
                logger.debug(f"📤 Published to channel {self.name}: {result} subscribers")
                return result
                
        except Exception as e:
            logger.error(f"❌ Publish error on channel {self.name}: {e}")
            return 0
    
    async def subscribe(self, event_name: str, callback: Callable):
        """Subscribe to events on this channel."""
        if event_name not in self._callbacks:
            self._callbacks[event_name] = []
        
        self._callbacks[event_name].append(callback)
        
        # Subscribe to Redis channel if not already subscribed
        if not self._subscribed:
            self._pubsub.subscribe(self.name)
            self._subscribed = True
            logger.info(f"📡 Subscribed to Redis channel: {self.name}")
        
        logger.debug(f"📡 Added callback for {self.name}:{event_name}")
    
    def _dispatch(self, event_name: str, data: Any, timestamp: float = None):
        """Dispatch message to callbacks (called by listener thread)."""
        if event_name in self._callbacks:
            message = RedisMessage(
                channel=self.name,
                name=event_name,
                data=data,
                timestamp=timestamp or time.time()
            )
            
            for callback in self._callbacks[event_name]:
                try:
                    callback(message)
                except Exception as e:
                    logger.error(f"❌ Callback error on {self.name}:{event_name}: {e}")


class RedisAblyChannelCollection:
    """Collection of Redis channels (HuggingFace Spaces optimized)."""
    
    def __init__(self, redis_client: redis.Redis, pubsub: redis.client.PubSub, use_streams: bool = False):
        self._redis = redis_client
        self._pubsub = pubsub
        self._use_streams = use_streams
        self._channels: Dict[str, RedisAblyChannel] = {}
    
    def get(self, channel_name: str) -> RedisAblyChannel:
        """Get or create a channel."""
        if channel_name not in self._channels:
            self._channels[channel_name] = RedisAblyChannel(
                channel_name, self._redis, self._pubsub, self._use_streams
            )
        return self._channels[channel_name]


class RedisConnectionState:
    """Ably-compatible connection state object."""
    
    def __init__(self):
        self.state = 'initialized'
        self._callbacks: Dict[str, List[Callable]] = {}
    
    def on(self, event: str, callback: Callable):
        """Register event callback."""
        if event not in self._callbacks:
            self._callbacks[event] = []
        self._callbacks[event].append(callback)
    
    def _trigger(self, event: str, *args):
        """Trigger event callbacks."""
        if event in self._callbacks:
            for callback in self._callbacks[event]:
                try:
                    callback(*args)
                except Exception as e:
                    logger.error(f"❌ Connection event callback error: {e}")


# =============================================================================
# SIMPLIFIED REDIS-ABLY CLIENT (HuggingFace Spaces Edition)
# =============================================================================

class RedisAblyClient:
    """
    V10.1: Simplified Redis client with Ably-compatible interface.
    HuggingFace Spaces optimized with environment-aware configuration.
    """
    
    def __init__(
        self,
        redis_url: Optional[str] = None,
        key: Optional[str] = None,  # Ignored (Ably compatibility)
        password: Optional[str] = None,
        use_streams: bool = False,
        database: int = 0,  # ✅ ADDED: Database selection
    ):
        if key:
            logger.info("ℹ️  [RedisAblyClient] Ably key ignored (migration compatibility)")
        
        # ✅ FIXED: Use redis_config with database support
        if redis_url is None:
            if database == REDIS_DB_FEATURES:
                redis_url = RedisConfig.for_database(REDIS_DB_FEATURES).url
            elif database == REDIS_DB_REWARDS:
                redis_url = RedisConfig.for_database(REDIS_DB_REWARDS).url
            else:
                redis_url = RedisConfig.for_database(database).url
        
        self._redis_url = redis_url
        
        # Connection settings optimized for HuggingFace Spaces
        connection_kwargs = {
            'decode_responses': True,
            'socket_timeout': 15 if IS_HF_SPACES else 10,
            'socket_connect_timeout': 15 if IS_HF_SPACES else 10,
            'retry_on_timeout': True,
        }
        
        if password or REDIS_PASSWORD:
            connection_kwargs['password'] = password or REDIS_PASSWORD
        
        self._redis = redis.Redis.from_url(redis_url, **connection_kwargs)
        
        # Separate connection for pub/sub
        # ✅ FIX: socket_timeout=30 instead of None — enables heartbeat detection
        # Same fix as DedicatedRedisConnectionManager. Without this,
        # Rewards.py's signal listener can silently die, stopping ALL reward generation.
        pubsub_kwargs = connection_kwargs.copy()
        pubsub_kwargs['socket_timeout'] = 30  # Heartbeat every 30s
        
        self._sub_redis = redis.Redis.from_url(redis_url, **pubsub_kwargs)
        self._pubsub = self._sub_redis.pubsub(ignore_subscribe_messages=True)
        
        # Ably-compatible interfaces
        self.channels = RedisAblyChannelCollection(
            self._redis, self._pubsub, use_streams
        )
        self.connection = RedisConnectionState()
        
        # Start BLOCKING listener thread
        self._running = True
        self._listener_thread = threading.Thread(
            target=self._listener_loop,
            daemon=True,
            name="RedisAblyClientListener-V10.1"
        )
        self._listener_thread.start()
        
        self.connection.state = 'connected'
        self.connection._trigger('connected')
        
        if IS_HF_SPACES:
            logger.info(f"✅ RedisAblyClient V10.1 (HF Spaces) connected to {redis_url}")
        else:
            logger.info(f"✅ RedisAblyClient V10.1 connected to {redis_url}")
    
    def _listener_loop(self):
        """
        BLOCKING listener using pubsub.listen() - zero polling latency!
        HuggingFace Spaces optimized with enhanced error handling.
        
        ✅ FIX: Added heartbeat via socket_timeout=30 and stall detection.
        This is the Rewards.py side — if this listener dies, Rewards.py stops
        receiving signals and NO rewards get calculated for ANY consumer.
        """
        reconnect_attempts = 0
        STALL_THRESHOLD = 120  # Force reconnect after 2 min silence
        self._last_ably_message_time = time.time()
        
        while self._running:
            try:
                # TRUE BLOCKING LISTEN (with 30s heartbeat timeout)
                for message in self._pubsub.listen():
                    if not self._running:
                        break
                    
                    if message['type'] == 'message':
                        channel_name = message['channel']
                        raw_data = message['data']
                        self._last_ably_message_time = time.time()
                        
                        try:
                            envelope = json.loads(raw_data)
                            event_name = envelope.get('event', 'message')
                            data = envelope.get('data', envelope)
                            timestamp = envelope.get('timestamp', time.time())
                        except (json.JSONDecodeError, TypeError):
                            event_name = 'message'
                            data = raw_data
                            timestamp = time.time()
                        
                        if channel_name in self.channels._channels:
                            self.channels._channels[channel_name]._dispatch(
                                event_name, data, timestamp
                            )
                
                # Reset reconnect counter on successful operation
                reconnect_attempts = 0
            
            except (redis.TimeoutError, TimeoutError):
                # ✅ HEARTBEAT: Fires every ~30s when no messages arrive
                silence = time.time() - self._last_ably_message_time
                
                if silence > STALL_THRESHOLD:
                    logger.warning(
                        f"[RedisAblyClient] ⚠️ No messages for {silence:.0f}s — "
                        f"forcing reconnect (subscription may have silently dropped)"
                    )
                    # Force full reconnect
                    try:
                        connection_kwargs = {
                            'decode_responses': True,
                            'socket_timeout': 30,  # ✅ FIX: was None
                            'retry_on_timeout': True,
                        }
                        if REDIS_PASSWORD:
                            connection_kwargs['password'] = REDIS_PASSWORD
                        
                        self._sub_redis = redis.Redis.from_url(self._redis_url, **connection_kwargs)
                        self._pubsub = self._sub_redis.pubsub(ignore_subscribe_messages=True)
                        
                        # Re-subscribe to all channels
                        for ch in self.channels._channels:
                            self._pubsub.subscribe(ch)
                        
                        self._last_ably_message_time = time.time()
                        logger.info(f"[RedisAblyClient] ✅ Reconnected after silent stall, re-subscribed to {len(self.channels._channels)} channels")
                    except Exception as e:
                        logger.error(f"[RedisAblyClient] ❌ Reconnection failed: {e}")
                        time.sleep(5)
                
                continue  # Back to listen()
                            
            except redis.ConnectionError:
                logger.warning("[RedisAblyClient] ⚠️  Connection lost, reconnecting...")
                
                # Exponential backoff for HuggingFace Spaces
                if reconnect_attempts < 10:
                    reconnect_attempts += 1
                
                delay = min(2 ** reconnect_attempts, 60) if IS_HF_SPACES else min(2 ** reconnect_attempts, 30)
                time.sleep(delay)
                
                try:
                    self._sub_redis.ping()
                except Exception:
                    # Recreate connection
                    connection_kwargs = {
                        'decode_responses': True,
                        'socket_timeout': 30,  # ✅ FIX: was None
                        'retry_on_timeout': True,
                    }
                    if REDIS_PASSWORD:
                        connection_kwargs['password'] = REDIS_PASSWORD
                        
                    self._sub_redis = redis.Redis.from_url(self._redis_url, **connection_kwargs)
                    self._pubsub = self._sub_redis.pubsub(ignore_subscribe_messages=True)
                    
                    # Re-subscribe
                    for ch in self.channels._channels:
                        self._pubsub.subscribe(ch)
                    
                    self._last_ably_message_time = time.time()
                        
            except Exception as e:
                if self._running:
                    logger.error(f"[RedisAblyClient] ❌ Listener error: {e}")
                time.sleep(1 if not IS_HF_SPACES else 2)  # Slower retry for containers
    
    def close(self):
        """Close connections."""
        self._running = False
        try:
            self._pubsub.unsubscribe()
            self._pubsub.close()
        except Exception:
            pass
        try:
            self._redis.close()
            self._sub_redis.close()
        except Exception:
            pass
        self.connection.state = 'closed'
        logger.info("✅ RedisAblyClient closed")


# =============================================================================
# DIAGNOSTIC UTILITY (HuggingFace Spaces Edition)
# =============================================================================

def diagnose_redis_connection(system, verbose: bool = True) -> Optional[dict]:
    """
    V10.2: Redis diagnostics with HuggingFace Spaces awareness and latch model reporting.
    
    Args:
        system: Object with .ably_manager attribute
        verbose: Print detailed output
    
    Returns:
        Connection state dict (includes 'latch_model' sub-dict)
    """
    manager = getattr(system, 'ably_manager', None)
    if manager is None:
        logger.warning("⚠️  No Redis manager found on system")
        return None
    
    if verbose:
        print(f"\n{'='*80}")
        print(f"📡 REDIS CONNECTION DIAGNOSTICS (V10.2 - HuggingFace Spaces)")
        print(f"{'='*80}")
        print(f"🤗 HuggingFace Spaces: {IS_HF_SPACES}")
        print(f"📋 Redis Config Available: {HAS_REDIS_CONFIG}")
    
    try:
        state = manager.get_state(timeout=5 if IS_HF_SPACES else 2)
        
        if verbose:
            print(f"  Connected:          {state.get('connected', 'unknown')}")
            print(f"  State:              {state.get('state', 'unknown')}")
            print(f"  Reconnect attempts: {state.get('reconnect_attempts', 0)}")
            
            config = state.get('config', {})
            print(f"  Redis URL:          {config.get('redis_url', 'unknown')}")
            print(f"  Socket timeout:     {config.get('socket_timeout', 'unknown')}")
            
            stats = state.get('stats', {})
            print(f"\n  📊 Statistics:")
            print(f"    Connections:      {stats.get('connections', 0)}")
            print(f"    Disconnections:   {stats.get('disconnections', 0)}")
            print(f"    Reconnections:    {stats.get('reconnections', 0)}")
            print(f"    Messages recv:    {stats.get('messages_received', 0)}")
            print(f"    Messages pub:     {stats.get('messages_published', 0)}")
            print(f"    Publish errors:   {stats.get('publish_errors', 0)}")
            print(f"    Avg latency:      {stats.get('avg_latency_ms', 0):.2f}ms")
            print(f"    Last connected:   {stats.get('last_connected', 'never')}")
            print(f"    Uptime:           {stats.get('total_uptime', 0):.0f}s")

            # ── LATCH MODEL DIAGNOSTICS ────────────────────────────────────
            latch = state.get('latch_model', {})
            print(f"\n  🔒 Latest-State Latch Model (V10.2):")
            print(f"    Enabled:              {latch.get('latest_only_enabled', False)}")
            lo_channels = latch.get('latest_only_channels', [])
            if lo_channels:
                print(f"    Latest-only channels: {', '.join(lo_channels)}")
            else:
                print(f"    Latest-only channels: (none)")
            if_channels = latch.get('in_flight_channels', [])
            print(f"    In-flight now:        {', '.join(if_channels) if if_channels else '(none)'}")
            print(f"    Pending slot occupancy: {latch.get('pending_slot_occupancy', 0)}")
            print(f"    Messages dispatched:  {latch.get('messages_dispatched', 0)}")
            print(f"    Messages overwritten: {latch.get('messages_overwritten', 0)}")
            print(f"    Busy drops avoided:   {latch.get('busy_drops_avoided', 0)}")

            subs = state.get('subscriptions', [])
            print(f"\n  📌 Subscriptions ({len(subs)}):")
            for ch in subs[:10]:
                tag = " [latest_only]" if ch in lo_channels else ""
                print(f"    • {ch}{tag}")
            if len(subs) > 10:
                print(f"    ... and {len(subs) - 10} more")
            
            print(f"{'='*80}\n")
        
        return state
        
    except Exception as e:
        if verbose:
            print(f"  ❌ Diagnostic error: {e}")
            print(f"{'='*80}\n")
        return {'error': str(e)}


# =============================================================================
# FEATURE BATCH GATEWAY
# =============================================================================
#
# Architecture (Karl's diagram, 2026-04):
#
#   Market → p_now → Features → 8 per-agent publishes → Redis
#                                                            │
#                                                   FeatureBatchGateway
#                                                            │
#                                          discards waitlist on new tick
#                                          fires only when all N agents
#                                          share the same tick / price
#                                                            │
#                                                     complete batch
#                                                            │
#                                                  Erasure / QSAP engine
#                                                  produce signal at p=p_now
#
# The gateway is the synchronisation point:
#   • Features.py publishes each agent's features independently to its own
#     per-agent Redis channel (xs, s, m, l, xl, xxl, 5m, 10m).
#   • FeatureBatchGateway subscribes to all N agent channels via the
#     DedicatedRedisConnectionManager (latest_only=True, so stale same-agent
#     messages are already coalesced at the transport layer).
#   • Each arriving message updates that agent's slot in the current-tick
#     accumulator.
#   • If a new tick_index arrives while the previous tick's batch is still
#     incomplete, the incomplete batch is DISCARDED ("waitlist discard") and
#     the accumulator resets — ensuring Quasar only ever sees feature sets
#     where all agents share the same p_now.
#   • When all N agents have contributed to the same tick, the gateway fires
#     on_batch_ready(batch_payload) exactly once.

class FeatureBatchGateway:
    """
    Redis-side batch synchronisation gate.

    Subscribes to N per-agent feature channels via a
    DedicatedRedisConnectionManager.  Accumulates per-tick contributions
    and fires on_batch_ready only when all N agents have arrived for the
    same tick_index.  Any partial batch for a superseded tick is silently
    discarded — the downstream consumer (QSAP / Erasure engine) therefore
    always receives a coherent, same-price, all-agents feature set.

    Usage
    -----
        gateway = FeatureBatchGateway(
            manager        = redis_manager,
            agent_names    = ['xs','s','m','l','xl','xxl','5m','10m'],
            channel_prefix = 'V75_1S:',
            on_batch_ready = lambda batch: asyncio.ensure_future(
                system.on_tick_batch_message(batch)
            ),
        )
        gateway.start()
    """

    def __init__(
        self,
        manager,                        # DedicatedRedisConnectionManager
        agent_names: list,
        on_batch_ready,                 # callable(batch_dict) — fired on complete tick
        channel_prefix: str = '',       # e.g. 'V75_1S:' to match namespaced channels
        n_required: int = None,         # agents required; defaults to len(agent_names)
        stale_timeout_s: float = 10.0,  # discard incomplete batch after this many seconds
    ):
        self._manager       = manager
        self._agent_names   = list(agent_names)
        self._n_required    = n_required or len(agent_names)
        self._on_batch      = on_batch_ready
        self._prefix        = channel_prefix
        self._stale_timeout = stale_timeout_s

        # Accumulator state (guarded by _lock)
        self._lock          = threading.Lock()
        self._current_tick  = None          # tick_index for in-progress batch
        self._batch: dict   = {}            # agent_name → feature payload
        self._batch_price: float = 0.0      # price from first agent in current tick
        self._batch_ts: float    = 0.0      # monotonic time when batch was opened

        # Diagnostics
        self._batches_fired:    int = 0
        self._partial_discards: int = 0
        self._agent_overwrites: int = 0     # same agent sent twice in one tick

        self._started = False
        logger.info("[FeatureBatchGateway] Created: agents=%s n_required=%d prefix=%r",
                    self._agent_names, self._n_required, self._prefix)

    # ── public API ────────────────────────────────────────────────────────────

    def start(self) -> None:
        """Subscribe to all per-agent channels through the manager."""
        if self._started:
            logger.warning("[FeatureBatchGateway] Already started — ignoring duplicate start()")
            return

        for agent in self._agent_names:
            channel = f"{self._prefix}{agent}"
            self._manager.subscribe(
                channel    = channel,
                callback   = self._make_handler(agent),
                latest_only = True,   # latch model: coalesce same-agent bursts
            )
            logger.info("[FeatureBatchGateway] Subscribed to channel: %s (latest_only=True)", channel)

        self._started = True
        logger.info("[FeatureBatchGateway] Started — waiting for features on %d channels",
                    len(self._agent_names))

    def stop(self) -> None:
        """Unsubscribe from all channels."""
        for agent in self._agent_names:
            channel = f"{self._prefix}{agent}"
            self._manager.unsubscribe(channel)
        self._started = False
        logger.info("[FeatureBatchGateway] Stopped")

    @property
    def diagnostics(self) -> dict:
        with self._lock:
            return {
                'batches_fired':      self._batches_fired,
                'partial_discards':   self._partial_discards,
                'agent_overwrites':   self._agent_overwrites,
                'current_tick':       self._current_tick,
                'agents_in_batch':    len(self._batch),
                'n_required':         self._n_required,
            }

    # ── internal ──────────────────────────────────────────────────────────────

    def _make_handler(self, agent_name: str):
        """Return a per-agent message handler closed over agent_name."""
        def _handler(message):
            try:
                self._receive(agent_name, message)
            except Exception as e:
                logger.error("[FeatureBatchGateway] Handler error for %s: %s", agent_name, e)
        return _handler

    def _receive(self, agent_name: str, message) -> None:
        """
        Process one arriving per-agent feature message.

        Called by the DedicatedRedisConnectionManager's listener thread —
        must be fast and must not block.
        """
        # Unpack the Redis envelope
        data = getattr(message, 'data', None)
        if data is None:
            if isinstance(message, dict):
                data = message
            else:
                logger.debug("[FeatureBatchGateway] %s: unrecognised message shape", agent_name)
                return

        # Extract tick_index (price epoch key) and raw price
        tick_index  = (
            data.get('tick_index')
            or (data.get('features') or {}).get('tick_index')
            or (data.get('features') or {}).get('tick_count')
        )
        raw_price   = data.get('price')          # envelope-level raw Deriv tick
        features    = data.get('features', {})

        now_mono = time.monotonic()

        with self._lock:
            # ── Stale-batch eviction ──────────────────────────────────────
            # If the in-progress batch is older than stale_timeout_s, discard it
            # before deciding whether the incoming tick matches.
            if (self._current_tick is not None
                    and (now_mono - self._batch_ts) > self._stale_timeout):
                n_stale = len(self._batch)
                logger.warning(
                    "[FeatureBatchGateway] Discarding stale partial batch: "
                    "tick=%s agents=%d/%d age=%.1fs",
                    self._current_tick, n_stale, self._n_required,
                    now_mono - self._batch_ts,
                )
                self._partial_discards += 1
                self._batch       = {}
                self._current_tick = None

            # ── New-tick reset (waitlist discard) ─────────────────────────
            # This is the core of Karl's diagram:
            # When a new tick_index arrives while a partial batch is open,
            # the old partial batch is DISCARDED and we start fresh.
            # This guarantees: every complete batch that reaches the consumer
            # has ALL agents at the SAME price / tick.
            if tick_index is not None and tick_index != self._current_tick:
                if self._batch:
                    logger.info(
                        "[FeatureBatchGateway] New tick %s arrived — discarding partial "
                        "batch for tick %s (%d/%d agents). Waitlist cleared.",
                        tick_index, self._current_tick,
                        len(self._batch), self._n_required,
                    )
                    self._partial_discards += 1
                self._batch        = {}
                self._current_tick = tick_index
                self._batch_ts     = now_mono
                self._batch_price  = float(raw_price) if raw_price else 0.0

            # ── Accumulate ────────────────────────────────────────────────
            if agent_name in self._batch:
                self._agent_overwrites += 1
                logger.debug("[FeatureBatchGateway] Agent %s overwrote slot in tick %s",
                             agent_name, self._current_tick)

            self._batch[agent_name] = {
                'features':  features,
                'price':     raw_price,
                'tick_index': tick_index,
            }

            # Update price from latest arrival if previous was missing
            if self._batch_price == 0.0 and raw_price:
                try:
                    self._batch_price = float(raw_price)
                except (ValueError, TypeError):
                    pass

            n_now = len(self._batch)
            logger.debug("[FeatureBatchGateway] tick=%s agents=%d/%d (+%s)",
                         self._current_tick, n_now, self._n_required, agent_name)

            # ── Fire batch when complete ──────────────────────────────────
            if n_now >= self._n_required:
                batch_payload = {
                    'tick_index':  self._current_tick,
                    'price':       self._batch_price,
                    'timestamp':   time.strftime('%H:%M:%S'),
                    'agent_count': n_now,
                    'agents': {
                        ag: {'features': v['features'], 'price': v['price']}
                        for ag, v in self._batch.items()
                    },
                }
                self._batches_fired += 1
                logger.info(
                    "[FeatureBatchGateway] ✅ BATCH COMPLETE tick=%s price=%.4f "
                    "agents=%d (batch #%d)",
                    self._current_tick, self._batch_price, n_now, self._batches_fired,
                )
                # Reset accumulator BEFORE firing so re-entrant calls don't corrupt state
                self._batch        = {}
                self._current_tick = None
                self._batch_price  = 0.0

                # Fire the callback OUTSIDE the lock to avoid deadlock if the
                # consumer calls back into the manager.
                batch_to_fire = batch_payload

            else:
                batch_to_fire = None

        # ── Fire outside lock ─────────────────────────────────────────────
        if batch_to_fire is not None:
            try:
                self._on_batch(batch_to_fire)
            except Exception as e:
                logger.error("[FeatureBatchGateway] on_batch_ready raised: %s", e)


# =============================================================================
# EXAMPLE USAGE (HuggingFace Spaces Compatible)
# =============================================================================

if __name__ == "__main__":
    import sys
    
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
    )
    
    def on_message(msg: RedisMessage):
        print(f"📨 Received: {msg}")
    
    def on_connected():
        print("🟢 Connected!")
    
    def on_disconnected():
        print("🔴 Disconnected!")
    
    # Example: Synchronous usage with HuggingFace Spaces support
    print(f"\n=== Testing DedicatedRedisConnectionManager V10.2 ===")
    print(f"🤗 HuggingFace Spaces: {IS_HF_SPACES}")
    print(f"📋 Redis Config: {'Available' if HAS_REDIS_CONFIG else 'Using defaults'}")
    print()
    
    manager = DedicatedRedisConnectionManager(
        on_connected=on_connected,
        on_disconnected=on_disconnected,
        on_message=lambda ch, msg: print(f"[Global] {ch}: {msg}"),
        database=0  # Use default database
    )
    
    timeout = 10 if IS_HF_SPACES else 5
    
    if manager.start(timeout=timeout):
        print("✅ Manager started successfully")
        
        # ── Lossless subscription (rewards / audit channels) ──────────────
        manager.subscribe("reward-channel", callback=on_message)

        # ── Latest-only subscription (live feature / bound-state channels) ─
        # Messages arriving while the callback executes are coalesced:
        # only the FRESHEST pending state is delivered after the callback
        # finishes; all intermediate arrivals are intentionally discarded.
        manager.subscribe(
            "features-live",
            callback=on_message,
            latest_only=True,   # ← enables transport-side latch model
        )
        
        # Publish (with timing)
        time.sleep(1)  # Wait for subscriptions
        
        start = time.time()
        result = manager.publish(
            "features-live",
            {"hello": "world", "platform": "HuggingFace Spaces" if IS_HF_SPACES else "local"},
            "feature-update"
        )
        latency = (time.time() - start) * 1000
        
        print(f"📤 Published to {result} subscribers (latency: {latency:.2f}ms)")
        
        # Wait for message
        time.sleep(2)
        
        # Get state
        state = manager.get_state()
        print(f"\n📊 Final State:")
        print(f"  Connected: {state.get('connected')}")
        print(f"  Messages sent: {state.get('stats', {}).get('messages_published', 0)}")
        print(f"  Messages received: {state.get('stats', {}).get('messages_received', 0)}")

        # Print latch model diagnostics
        latch = manager.get_latch_diagnostics()
        print(f"\n🔒 Latch Model Diagnostics:")
        print(f"  Latest-only channels: {latch.get('latest_only_channels')}")
        print(f"  Messages dispatched:  {latch.get('messages_dispatched')}")
        print(f"  Messages overwritten: {latch.get('messages_overwritten')}")
        print(f"  Busy drops avoided:   {latch.get('busy_drops_avoided')}")
        
        manager.stop()
    else:
        print("❌ Failed to start manager")
        sys.exit(1)