Add workflow failure Teams notifications

.github/workflows/hyperforge_workflow.yaml

@@ -131,3 +131,34 @@ jobs:
         run: |
           docker push "ghcr.io/$REPOSITORY_OWNER/hyperforge:$IMAGE_SHA_TAG"
           docker push "ghcr.io/$REPOSITORY_OWNER/hyperforge:$IMAGE_REF_TAG"
+
+  notify-teams-fail:
+    name: Notify Teams on failure
+    needs:
+      - test
+      - build_wheels
+      - build_and_push_docker_image
+    if: (failure() || cancelled()) && github.event_name == 'push'
+    runs-on: ubuntu-24.04
+    steps:
+      - name: Send Teams notification
+        env:
+          REF_NAME: ${{ github.ref_name }}
+          RUN_NUMBER: ${{ github.run_number }}
+          RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
+          COMMIT_SHA: ${{ github.sha }}
+          TEAMS_WEBHOOK_URL: ${{ secrets.TEAMS_WEBHOOK_URL }}
+        run: |
+          curl -s -o /dev/null -w "%{http_code}" \
+            -H "Content-Type: application/json" \
+            -d "{
+              \"@type\": \"MessageCard\",
+              \"@context\": \"http://schema.org/extensions\",
+              \"themeColor\": \"FF0000\",
+              \"summary\": \"Hyperforge CI failed\",
+              \"sections\": [{
+                \"activityTitle\": \"**Hyperforge CI failed on \`${REF_NAME}\`**\",
+                \"activityText\": \"Run [#${RUN_NUMBER}](${RUN_URL}) failed for commit \`${COMMIT_SHA}\`.\"
+              }]
+            }" \
+            "$TEAMS_WEBHOOK_URL"

hyperforge/pyproject.toml

@@ -44,6 +44,7 @@ dependencies = [
     "sentry_sdk",
     "fastapi",
     "starlette>1.0.0",
+    "uvicorn>=0.30.0",
 ]
 
 

hyperforge/src/hyperforge/api/session.py

@@ -1,5 +1,6 @@
 """Session management functions for ARAG agents with NucliaDB memory."""
 
+from uuid import UUID
 from typing import Optional
 
 from nucliadb_models import (
@@ -19,6 +20,14 @@
 from hyperforge.memory.memory import QUESTION_ANSWERS_FIELD
 
 
+def _is_uuid(value: str) -> bool:
+    try:
+        UUID(value)
+        return True
+    except (ValueError, TypeError):
+        return False
+
+
 async def create_session_resource(
     ndb: NucliaDBAsync,
     agent_id: str,
@@ -110,11 +119,18 @@ async def session_exists(
         True if session exists, False otherwise
     """
     try:
-        await ndb.get_resource_by_id(
-            rid=session_id,
-            kbid=agent_id,
-            query_params={"show": ["basic"]},
-        )
+        if _is_uuid(session_id):
+            await ndb.get_resource_by_id(
+                rid=session_id,
+                kbid=agent_id,
+                query_params={"show": ["basic"]},
+            )
+        else:
+            await ndb.get_resource_by_slug(
+                slug=session_id,
+                kbid=agent_id,
+                query_params={"show": ["basic"]},
+            )
         return True
     except NotFoundError:
         return False

hyperforge/src/hyperforge/broker/redis.py

@@ -1,17 +1,30 @@
 import asyncio
 import json
-from typing import AsyncIterator, cast
+import time
+from typing import Any, AsyncIterator, cast
 
 import opentelemetry.propagate
 from pydantic import TypeAdapter
 from redis.asyncio import Redis, ResponseError
+from redis.asyncio.retry import Retry
+from redis.backoff import ExponentialBackoff
+from redis.exceptions import ConnectionError as RedisConnectionError
 
 from hyperforge import logger
 from hyperforge.broker import AgentTimeoutError, Broker
 from hyperforge.pubsub import AgentMessage, StartInteraction
 from hyperforge.redis_utils import ManualStreamKeysRedisCluster
 
 
+_REDIS_CONNECT_TIMEOUT_SECONDS = 5
+_REDIS_HEALTH_CHECK_INTERVAL_SECONDS = 30
+_REDIS_RETRY_ATTEMPTS = 3
+_REDIS_RETRY_BACKOFF_BASE_SECONDS = 0.1
+_REDIS_RETRY_BACKOFF_CAP_SECONDS = 1.0
+_REDIS_SUBSCRIBE_ACTIVATIONS_RETRY_SLEEP_SECONDS = 1
+_REPLY_READ_BLOCK_MAX_MS = 2000
+
+
 class RedisBroker(Broker):
     def __init__(self, client: Redis, activate_subject: str, keepalive_ms: int):
         self._client = client
@@ -30,23 +43,42 @@ def from_url(
         keepalive_ms: int,
         cluster_mode: bool = False,
     ) -> "RedisBroker":
+        client_kwargs = cls._client_kwargs(keepalive_ms)
         if cluster_mode:
             client = cast(
                 Redis,
                 ManualStreamKeysRedisCluster.from_url(
                     url=url,
-                    decode_responses=True,
                     dynamic_startup_nodes=False,
+                    **client_kwargs,
                 ),
             )
         else:
             client = Redis.from_url(
                 url,
-                decode_responses=True,
-                socket_timeout=keepalive_ms / 1000,
-            )
+                **client_kwargs,
+            )  # type: ignore[call-overload]
         return cls(client, activate_subject, keepalive_ms)
 
+    @staticmethod
+    def _client_kwargs(keepalive_ms: int) -> dict[str, Any]:
+        retry = Retry(
+            backoff=ExponentialBackoff(
+                base=_REDIS_RETRY_BACKOFF_BASE_SECONDS,
+                cap=_REDIS_RETRY_BACKOFF_CAP_SECONDS,
+            ),
+            retries=_REDIS_RETRY_ATTEMPTS,
+        )
+        return {
+            "decode_responses": True,
+            "socket_timeout": keepalive_ms / 1000,
+            "socket_connect_timeout": _REDIS_CONNECT_TIMEOUT_SECONDS,
+            "health_check_interval": _REDIS_HEALTH_CHECK_INTERVAL_SECONDS,
+            "socket_keepalive": True,
+            "retry": retry,
+            "retry_on_timeout": True,
+        }
+
     async def publish_activation(
         self, msg: StartInteraction, trace: dict[str, str]
     ) -> None:
@@ -102,10 +134,12 @@ async def subscribe_activations(
                     logger.exception(
                         "Error while subscribing to activations, retrying..."
                     )
-                    await asyncio.sleep(1)
+                    await asyncio.sleep(
+                        _REDIS_SUBSCRIBE_ACTIVATIONS_RETRY_SLEEP_SECONDS
+                    )
             except Exception:
                 logger.exception("Error while subscribing to activations, retrying...")
-                await asyncio.sleep(1)
+                await asyncio.sleep(_REDIS_SUBSCRIBE_ACTIVATIONS_RETRY_SLEEP_SECONDS)
 
     async def publish(self, topic: str, message: AgentMessage) -> None:
         async with self._client.pipeline() as pipe:
@@ -159,14 +193,38 @@ async def send_reply(self, key: str, payload: str) -> None:
         )
 
     async def receive_reply(self, key: str, timeout_ms: int) -> str | None:
-        response = await self._client.xread(
-            {key: "$"},
-            block=timeout_ms,
-            count=1,
-        )
-        if not response:
-            return None
-        return response[0][1][0][1]["msg"]
+        deadline = time.monotonic() + (timeout_ms / 1000)
+        reconnect_attempts = 0
+
+        while True:
+            remaining_s = deadline - time.monotonic()
+            if remaining_s <= 0:
+                return None
+
+            # Keep each blocking read short so transient socket closures/failovers
+            # can be recovered within the same overall timeout budget.
+            block_ms = min(_REPLY_READ_BLOCK_MAX_MS, max(1, int(remaining_s * 1000)))
+
+            try:
+                response = await self._client.xread(
+                    {key: "$"},
+                    block=block_ms,
+                    count=1,
+                )
+            except RedisConnectionError:
+                reconnect_attempts += 1
+                backoff_s = min(1.0, 0.1 * reconnect_attempts)
+                logger.warning(
+                    "Redis connection closed while receiving reply key=%s; retrying (attempt=%d)",
+                    key,
+                    reconnect_attempts,
+                )
+                await asyncio.sleep(backoff_s)
+                continue
+
+            if not response:
+                continue
+            return response[0][1][0][1]["msg"]
 
     async def initialize(self) -> None:
         pass

hyperforge/src/hyperforge/server/session.py

@@ -10,7 +10,7 @@
 import nucliadb_telemetry.context
 import nucliadb_telemetry.metrics
 import prometheus_client
-from aiohttp.web import Server
+from hyperforge.server.web import WebServer
 from lru import LRU
 from nucliadb_telemetry import errors
 from nucliadb_telemetry.utils import get_telemetry
@@ -62,7 +62,7 @@ def tracer():
 
 
 class SessionManager:
-    server: Optional[Server] = None
+    server: Optional[WebServer] = None
     tasks: List[Task]
     hooks: Optional[Dict[str, List[Callable]]] = None
 

hyperforge/src/hyperforge/server/web.py

@@ -1,35 +1,66 @@
+import asyncio
 import os
 
 import prometheus_client  # type: ignore
-from aiohttp import web
+import uvicorn
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import PlainTextResponse, Response
+from starlette.routing import Route
 
 from hyperforge.server import logger
 
 PORT = os.environ.get("HEALTH_CHECK_PORT", "8000")
 
 
-async def http_handler(request: web.Request):
-    if request.path == "/metrics":
-        output = prometheus_client.exposition.generate_latest()
-        return web.Response(text=output.decode("utf8"))
-    elif request.path in ("/health/alive", "/health/ready"):
-        # implement health check here
-        return web.Response(text="OK")
-    else:
-        return web.Response(text="OK", status=404)
+async def metrics(request: Request) -> Response:
+    output = prometheus_client.exposition.generate_latest()
+    return PlainTextResponse(output.decode("utf8"))
 
 
-async def start_web_server() -> web.Server:
-    server = web.Server(http_handler)  # type: ignore
-    runner = web.ServerRunner(server)
-    await runner.setup()
-    site = web.TCPSite(runner, "0.0.0.0", int(PORT))
-    await site.start()
+async def health(request: Request) -> Response:
+    return PlainTextResponse("OK")
 
+
+async def not_found(request: Request, exc: Exception) -> Response:
+    return PlainTextResponse("OK", status_code=404)
+
+
+app = Starlette(
+    routes=[
+        Route("/metrics", metrics),
+        Route("/health/alive", health),
+        Route("/health/ready", health),
+    ],
+    exception_handlers={404: not_found},
+)
+
+
+class WebServer:
+    """Wraps a uvicorn server running as an asyncio task in the current event loop."""
+
+    def __init__(self, server: uvicorn.Server) -> None:
+        self._server = server
+        self._task: asyncio.Task | None = None
+
+    def start(self) -> None:
+        self._task = asyncio.get_event_loop().create_task(self._server.serve())
+
+    async def shutdown(self) -> None:
+        self._server.should_exit = True
+        if self._task is not None:
+            await self._task
+            self._task = None
+
+
+async def start_web_server() -> WebServer:
+    config = uvicorn.Config(app, host="0.0.0.0", port=int(PORT), log_level="warning")
+    server = uvicorn.Server(config)
+    web = WebServer(server)
+    web.start()
     logger.info(f"======= Serving on http://0.0.0.0:{PORT}/ ======")
-    return server
+    return web
 
 
-async def start_health_check():
-    server = await start_web_server()
-    return server
+async def start_health_check() -> WebServer:
+    return await start_web_server()