From c1da5edc6352e7a1c86914907f3dfb8d16f34ecc Mon Sep 17 00:00:00 2001
From: Carmen Iniesta <carmen@nuclia.com>
Date: Thu, 9 Jul 2026 12:59:48 +0200
Subject: [PATCH 1/2] create consumer group if it is not there

---
 hyperforge/src/hyperforge/broker/redis.py | 24 +++++++++++++++++++++--
 1 file changed, 22 insertions(+), 2 deletions(-)

diff --git a/hyperforge/src/hyperforge/broker/redis.py b/hyperforge/src/hyperforge/broker/redis.py
index 32aa9c5..34428f1 100644
--- a/hyperforge/src/hyperforge/broker/redis.py
+++ b/hyperforge/src/hyperforge/broker/redis.py
@@ -5,6 +5,8 @@
 import opentelemetry.propagate
 from pydantic import TypeAdapter
 from redis.asyncio import Redis, ResponseError
+from redis.exceptions import ConnectionError as RedisConnectionError
+from redis.exceptions import TimeoutError as RedisTimeoutError
 
 from hyperforge import logger
 from hyperforge.broker import AgentTimeoutError, Broker
@@ -66,12 +68,23 @@ async def _ensure_consumer_group(self) -> None:
             )
         except ResponseError as e:
             if "BUSYGROUP" not in str(e):
+                logger.exception("Failed to create consumer group, will retry...")
                 raise
 
     async def subscribe_activations(
         self,
     ) -> AsyncIterator[tuple[StartInteraction, dict[str, str]]]:
-        await self._ensure_consumer_group()
+        max_retries = 10
+        for attempt in range(1, max_retries + 1):
+            try:
+                await self._ensure_consumer_group()
+                break
+            except (ResponseError, RedisConnectionError, RedisTimeoutError):
+                if attempt == max_retries:
+                    logger.error("Failed to create consumer group after %d attempts, giving up.", max_retries)
+                    raise
+                logger.warning("Failed to create consumer group (attempt %d/%d), retrying...", attempt, max_retries)
+                await asyncio.sleep(1)
 
         while True:
             try:
@@ -97,7 +110,14 @@ async def subscribe_activations(
             except ResponseError as e:
                 if "NOGROUP" in str(e):
                     logger.warning("Consumer group lost, re-creating...")
-                    await self._ensure_consumer_group()
+                    try:
+                        await self._ensure_consumer_group()
+                    except (
+                        ResponseError,
+                        RedisConnectionError,
+                        RedisTimeoutError,
+                    ):
+                        await asyncio.sleep(1)
                 else:
                     logger.exception(
                         "Error while subscribing to activations, retrying..."

From 48c31d48d85728fedacc1625512b7c0e0680d5a5 Mon Sep 17 00:00:00 2001
From: Carmen Iniesta <carmen@nuclia.com>
Date: Thu, 9 Jul 2026 15:10:11 +0200
Subject: [PATCH 2/2] Update redis.py

---
 hyperforge/src/hyperforge/broker/redis.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/hyperforge/src/hyperforge/broker/redis.py b/hyperforge/src/hyperforge/broker/redis.py
index 34428f1..e37deb0 100644
--- a/hyperforge/src/hyperforge/broker/redis.py
+++ b/hyperforge/src/hyperforge/broker/redis.py
@@ -84,7 +84,7 @@ async def subscribe_activations(
                     logger.error("Failed to create consumer group after %d attempts, giving up.", max_retries)
                     raise
                 logger.warning("Failed to create consumer group (attempt %d/%d), retrying...", attempt, max_retries)
-                await asyncio.sleep(1)
+                await asyncio.sleep(10)
 
         while True:
             try: