Fix Redis flush and wait_for_redis in modular mode

- Move modular Redis wait from uWSGI exec-pre to entrypoint (exec-pre runs under 'su -' which strips Docker env vars, so DISPATCHARR_ENV and REDIS_HOST were never available)
- Selective flush in modular mode: clears stale app state (stream locks, proxy metadata) while preserving Celery broker/result keys
- AIO mode unchanged: full flushdb via uWSGI exec-pre
- Update unit tests for both flush paths

scripts/wait_for_redis.py

@@ -12,6 +12,28 @@ import logging
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 logger = logging.getLogger(__name__)
 
+# Key prefixes used by Celery's broker (Kombu) and result backend.
+# These must be preserved in modular mode where Celery runs independently.
+_CELERY_KEY_PREFIXES = ('celery', '_kombu', 'unacked')
+
+
+def _flush_non_celery_keys(client):
+    """Delete all Redis keys except those belonging to Celery."""
+    cursor = '0'
+    deleted = 0
+    while True:
+        cursor, keys = client.scan(cursor=cursor, count=500)
+        to_delete = [
+            k for k in keys
+            if not k.decode('utf-8', errors='replace').startswith(_CELERY_KEY_PREFIXES)
+        ]
+        if to_delete:
+            deleted += client.delete(*to_delete)
+        if cursor == 0:
+            break
+    logger.info(f"Modular mode: selectively cleared {deleted} non-Celery Redis key(s)")
+
+
 def wait_for_redis(host='localhost', port=6379, db=0, password='', username='', max_retries=30, retry_interval=2):
     """Wait for Redis to become available"""
     redis_client = None
@@ -31,6 +53,15 @@ def wait_for_redis(host='localhost', port=6379, db=0, password='', username='',
                 socket_connect_timeout=2
             )
             redis_client.ping()
+            # Clear stale state on startup. In AIO mode, every service restarts
+            # together so a full flush is safe. In modular mode, Celery has its
+            # own lifecycle — preserve its broker/result keys and only wipe
+            # application state (stream locks, proxy metadata, etc.).
+            if os.environ.get('DISPATCHARR_ENV') == 'modular':
+                _flush_non_celery_keys(redis_client)
+            else:
+                redis_client.flushdb()
+                logger.info(f"Flushed Redis database")
             logger.info(f"✅ Redis at {host}:{port}/{db} is now available!")
             return True
         except (redis.exceptions.ConnectionError, redis.exceptions.TimeoutError) as e: